zenith-runtime 0.1.0

//! WorkerRuntime - eBPF + XSK + Worker 全链路集成运行时
//!
//! 本模块实现 Zenith 数据面的核心运行时，将所有子系统集成为统一的管道：
//!
//! ## 数据流架构
//! ```text
//! +------------------+     bpf_link      +------------------+
//! |   eBPF/XDP       | ──────────────→ |   AF_XDP         |
//! |  (内核态导流)     │   原子挂载        |  (零拷贝收发)     |
//! +------------------+                  +------------------+
//!                                              │
//!                                              ▼
//! +------------------+     准入判定      +------------------+
//! | SourceAdmission  | ←────────────── |    Worker        |
//! |   Engine         |                 |  (数据面循环)     |
//! +------------------+                 +------------------+
//!                                              │
//!                                              ▼
//! +------------------+     帧归还       +------------------+
//! |  FramePool       | ←────────────── |   Quarantine    |
//! |  (帧池管理)       |                 |  (隔离区)        |
//! +------------------+                  +------------------+
//! ```
//!
//! ## 核心能力
//! - eBPF bpf_link 原子挂载与零丢包热更新
//! - AF_XDP 零拷贝数据通路
//! - 单 Owner 所有权守恒
//! - 五级资源配额管理
//! - 来源准入引擎 O(1) 五元组匹配
//! - Quarantine 隔离区异常帧管理
//! - AutoOptimizer 自适应 QoS + 运行期重规划（控制线程驱动）
//! - ExtremePlanner 极致拓扑规划 + CpuAffinity 硬绑定
//! - 负载感知的缓存预测预热

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, atomic::{AtomicBool, AtomicU8}};
use std::thread::JoinHandle;
use std::time::Instant;

use zenith_capability::detection::{CapabilitySnapshot, EnvironmentDetector};
use zenith_capability::profile::{Profile, ProfileSelector};
use zenith_foundation::{FramePool, LedgerQuota, LedgerType, ResourceLedger, ResourceType};
#[cfg(target_os = "linux")]
use zenith_ebpf::{BankId, DualBankManager, XdpLoader, XdpProgramName};
#[cfg(target_os = "linux")]
use zenith_linux::xsk::XskConfig;
use zenith_net::source_admission::{AdmissionRule, SourceAdmissionEngine};
use zenith_net::worker::{Worker, WorkerState, WorkerStats};
use zenith_observability::{FaultRecorder, FaultType, MetricsCollector};

use crate::auto_optimizer::{AutoOptimizer, LoadSample, QosLevel};
// ChangeSet 仅用于 Linux 专属的 eBPF 双 Bank 热切换路径
#[cfg(target_os = "linux")]
use crate::changeset::ChangeSet;
use crate::cpu_affinity::{CpuAffinityConfig, CpuAffinityError};
use crate::extreme_planner::ExtremePlanner;
use crate::graph::{
    DomainSnapshot, ExecutionDomain, PlannedTopology, ResourceNode, RuntimeGraph,
};
use crate::supervisor::{
    DomainSupervisor, ExitReason, HealthReport, NodeSupervisor, QueueSupervisor, RestartStrategy,
    SupervisorConfig,
};

/// 帧大小（字节）：与 zenith-net Worker 数据面 FRAME_SIZE 一致
const FRAME_BYTES: u64 = 4096;
/// 默认连接配额（每队列账本）
const DEFAULT_CONNECTION_QUOTA: u64 = 1024;
/// 缓存预热回调类型（控制线程以预测热键调用）
type CacheWarmupHook = Box<dyn Fn(&[Box<str>]) + Send>;

/// WorkerRuntime 配置
#[derive(Debug, Clone)]
pub struct WorkerRuntimeConfig {
    /// 网卡接口索引
    pub ifindex: u32,
    /// 队列 ID
    pub queue_id: u32,
    /// XSK 配置（仅 Linux 数据面）
    #[cfg(target_os = "linux")]
    pub xsk_config: XskConfig,
    /// 帧池容量
    pub frame_pool_capacity: u32,
    /// eBPF 程序名称（仅 Linux 数据面）
    #[cfg(target_os = "linux")]
    pub ebpf_program: XdpProgramName,
    /// 是否启用双 Bank 热更新
    pub enable_dual_bank: bool,
    /// 是否启用 Quarantine 隔离区
    pub enable_quarantine: bool,
    /// 最大循环周期数（0=无限）
    pub max_cycles: u64,
    /// 域 ID
    pub domain_id: u32,
    /// Worker 故障重启策略（Permanent/Transient/Temporary，经三级 Supervisor 落实）
    pub restart_strategy: RestartStrategy,
}

impl Default for WorkerRuntimeConfig {
    fn default() -> Self {
        Self {
            ifindex: 0,
            queue_id: 0,
            #[cfg(target_os = "linux")]
            xsk_config: XskConfig::default(),
            frame_pool_capacity: 1024,
            #[cfg(target_os = "linux")]
            ebpf_program: XdpProgramName::Main,
            enable_dual_bank: false,
            enable_quarantine: true,
            max_cycles: 0,
            domain_id: 0,
            restart_strategy: RestartStrategy::Permanent,
        }
    }
}

impl WorkerRuntimeConfig {
    /// 创建新的运行时配置，使用指定网卡接口索引，其余字段使用默认值
    ///
    /// # Arguments
    /// * `ifindex` - 网卡接口索引
    pub fn new(ifindex: u32) -> Self {
        Self {
            ifindex,
            #[cfg(target_os = "linux")]
            xsk_config: XskConfig {
                ifindex,
                ..Default::default()
            },
            ..Default::default()
        }
    }
}

/// 运行时状态
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RuntimeState {
    /// 已创建
    Created,
    /// 初始化完成（eBPF 已加载但未挂载）
    Initialized,
    /// 运行中
    Running,
    /// 暂停
    Paused,
    /// 已停止
    Stopped,
    /// 错误状态
    Error,
}

impl RuntimeState {
    /// 判断当前状态是否为 `Running`
    pub fn is_running(&self) -> bool {
        matches!(self, RuntimeState::Running)
    }
}

/// 运行时统计信息
#[derive(Debug, Clone, Default)]
pub struct RuntimeStats {
    /// 总处理包数
    pub total_packets: u64,
    /// 总接收包数
    pub total_rx: u64,
    /// 总发送包数
    pub total_tx: u64,
    /// 总准入拒绝数
    pub total_rejected: u64,
    /// 总解析错误数
    pub total_parse_errors: u64,
    /// 总隔离帧数
    pub total_quarantined: u64,
    /// 完成循环周期数
    pub cycles_completed: u64,
    /// eBPF 热切换次数
    pub ebpf_switches: u64,
    /// 当前活动 Bank（仅 Linux 数据面）
    #[cfg(target_os = "linux")]
    pub active_bank: Option<BankId>,
    /// 运行时间（秒）
    pub uptime_seconds: u64,
}

/// 待重启 worker 计划（非阻塞退避）
///
/// 故障处理只登记计划，`run_cycle` 每周期检查到点再执行重启，
/// 禁止 `std::thread::sleep` 阻塞数据面调用线程。
#[derive(Debug, Clone, Copy)]
struct PendingRestart {
    /// worker 索引：0 = 主 worker，n ≥ 1 = `extra_workers[n - 1]`
    worker_index: usize,
    /// Supervisor 逻辑 worker ID（治理视图记账）
    supervised_id: u64,
    /// 下次可重启的单调时钟时刻（退避到期点）
    next_retry_at: Instant,
}

/// WorkerRuntime - 全链路数据面运行时
///
/// ## 设计原则
/// - **单 Owner**：所有资源归属本 Runtime，无共享状态
/// - **零堆热路径**：数据面循环无堆分配、无锁、无原子操作
/// - **类型安全**：通过 Rust 类型系统强制资源所有权
/// - **可观测性**：完整的统计与状态暴露，治理闭环经 zenith-observability 可选注入
/// - **自适应优化**：控制线程驱动 QoS 调整、拓扑重规划、缓存预热
pub struct WorkerRuntime {
    /// 配置
    config: WorkerRuntimeConfig,
    /// 运行时状态
    state: RuntimeState,
    /// eBPF XDP 加载器（仅 Linux 数据面）
    #[cfg(target_os = "linux")]
    xdp_loader: Option<XdpLoader>,
    /// 双 Bank 管理器（仅 Linux 数据面）
    #[cfg(target_os = "linux")]
    dual_bank: Option<DualBankManager>,
    /// 主 Worker 数据面循环（不可移除，承载准入/帧池等单例访问语义）
    worker: Worker,
    /// 扩容的额外数据面 Worker（`reconcile_worker_count` 真实构造/停止移除；
    /// 索引对齐：`extra_workers[i]` ↔ `supervised_worker_ids[i + 1]`）
    extra_workers: Vec<Worker>,
    /// 已配置的准入规则（扩容新 worker 时重放，保证多 worker 准入一致）
    admission_rules: Vec<AdmissionRule>,
    /// L7 UDP 数据出口（仅 Linux 数据面；None = 纯 L4 转发模式）
    ///
    /// 注入后 `run_cycle` 自动切换为 `process_cycle_with_udp_sink`，
    /// 准入放行的 UDP 数据报经此 sink 交付 L7 服务（如 QUIC/H3）。
    /// 由生产二进制（或 AF_XDP 桥）在 `start()` 前注入。
    #[cfg(target_os = "linux")]
    udp_sink: Option<Box<dyn FnMut(zenith_net::worker::UdpDatagram) + Send>>,
    /// 资源账本
    ledger: ResourceLedger,
    /// Quarantine 隔离区计数
    quarantined_frames: AtomicU64,
    /// 统计信息
    stats: RuntimeStats,
    /// 运行周期计数
    cycle_count: AtomicU64,

    // ==================== 自动优化集成 ====================
    /// 自动优化器（负载监控 + QoS + 缓存预测）
    auto_optimizer: Arc<AutoOptimizer>,
    /// 当前 QoS 等级（AtomicU8 供 Worker 热路径 Relaxed 读取）
    qos_level: Arc<AtomicU8>,
    /// 控制线程运行标志
    control_running: Arc<AtomicBool>,
    /// 控制线程句柄
    control_thread: Option<JoinHandle<()>>,
    /// 拓扑规划器
    planner: ExtremePlanner,
    /// CPU 亲和配置
    affinity_config: Option<CpuAffinityConfig>,
    /// 上次拓扑规划结果（真实保存，供后续 worker 分配与查询）
    last_topology: Option<PlannedTopology>,
    /// 重规划待处理标记（控制线程置位，run_cycle 消费并真正执行 replan）
    replan_pending: Arc<AtomicBool>,
    /// 目标 worker 数（控制线程按负载趋势写入，run_cycle 经 Supervisor 落实）
    target_worker_count: Arc<AtomicU64>,
    /// 内部拓扑图（控制线程触发的重规划输入）
    topology_graph: RuntimeGraph<64, 256>,
    /// 域需求快照（控制线程触发的重规划输入）
    domain_snapshots: Vec<DomainSnapshot>,
    /// 缓存预热回调（控制线程以预测热键调用；None = 未注册）
    cache_warmup_hook: Arc<Mutex<Option<CacheWarmupHook>>>,
    /// 上次已应用到 Worker 的 QoS 等级（避免每周期重复设置）
    last_applied_qos: QosLevel,
    /// 重规划亲和应用错误（fail-visible：不吞错，供观测）
    replan_apply_error: Option<String>,

    // ==================== Supervisor 层级 ====================
    /// 节点 Supervisor（顶层）
    node_supervisor: NodeSupervisor,
    /// 域 Supervisor（中层）
    domain_supervisor: DomainSupervisor,
    /// 队列 Supervisor（底层，直接管理本 Worker）
    queue_supervisor: QueueSupervisor,
    /// Supervisor 在管的逻辑 worker ID 列表（扩缩容记账；
    /// 索引对齐：`[0]` ↔ 主 worker，`[i]` ↔ `extra_workers[i - 1]`）
    supervised_worker_ids: Vec<u64>,
    /// 下次分配的逻辑 worker ID（单调递增）
    next_supervised_id: u64,
    /// 最近一次故障重启的退避时长
    last_restart_backoff: Option<std::time::Duration>,
    /// 非阻塞退避的待重启计划（run_cycle 到点执行，禁止 sleep 阻塞数据面）
    pending_restarts: Vec<PendingRestart>,

    // ==================== 能力检测 ====================
    /// 环境能力快照（构造时经 zenith-capability 检测）
    capability_snapshot: CapabilitySnapshot,
    /// 数据面 Profile（按能力快照选择：Performance/Balanced/Minimal）
    data_plane_profile: Profile,

    // ==================== 资源账本 ====================
    /// 队列子账本名称（ledger 分层 roll-up 路径）
    queue_ledger_name: String,
    /// 上次与帧池同步的已分配帧数
    ledger_synced_frames: u64,

    // ==================== 可观测性（治理闭环） ====================
    /// 指标采集器（可选注入；`None` = 零开销空实现，每周期仅一次分支判断）
    metrics: Option<Arc<Mutex<MetricsCollector>>>,
    /// 故障记录器（可选注入；`None` = 零开销空实现）
    fault_recorder: Option<Arc<Mutex<FaultRecorder>>>,
    /// 上次指标上报时的累计处理包数（counter 增量差分基准）
    metrics_last_processed: u64,
    /// 上次指标上报时的累计准入拒绝数（counter 增量差分基准）
    metrics_last_rejected: u64,
    /// 上次指标上报时的累计丢包数/解析错误（counter 增量差分基准）
    metrics_last_dropped: u64,
}

impl WorkerRuntime {
    /// 创建新的 WorkerRuntime
    ///
    /// 构造时经 `zenith-capability` 检测环境能力并选择数据面 Profile，
    /// 初始化三级 Supervisor 层级与真实配额的资源账本。
    pub fn new(config: WorkerRuntimeConfig) -> Result<Self, RuntimeError> {
        let admission = SourceAdmissionEngine::allow_all();
        // Linux：真实 AF_XDP Worker（携带 XSK 配置）；非 Linux：可移植桩 Worker
        // （帧池/准入真实可用，数据面 process_cycle fail-closed）
        #[cfg(target_os = "linux")]
        let worker = Worker::new(
            config.queue_id,
            config.xsk_config.clone(),
            config.frame_pool_capacity,
            admission,
        )?;
        #[cfg(not(target_os = "linux"))]
        let worker = Worker::new(config.queue_id, config.frame_pool_capacity, admission)?;

        // 环境能力检测 → 数据面 Profile 选择（Performance/Balanced/Minimal）
        let capability_snapshot = EnvironmentDetector::new().detect();
        let data_plane_profile = ProfileSelector::new().select(&capability_snapshot);

        // 真实资源配额：帧配额 = 帧池容量，内存配额 = 容量 × 帧大小，连接配额固定
        let frame_capacity = u64::from(config.frame_pool_capacity);
        let quota = LedgerQuota {
            memory_bytes: frame_capacity.saturating_mul(FRAME_BYTES),
            frame_count: frame_capacity,
            connection_count: DEFAULT_CONNECTION_QUOTA,
        };
        let mut ledger = ResourceLedger::new("worker-runtime", LedgerType::Domain, quota.clone());
        // 队列子账本：帧/连接分配经 allocate_in_child 分层 roll-up
        let queue_ledger_name = format!("queue-{}", config.queue_id);
        ledger
            .create_child(&queue_ledger_name, LedgerType::Queue, quota)
            .map_err(|e| RuntimeError::InvalidConfig(format!("ledger init: {e}")))?;

        // 三级 Supervisor 层级：Node → Domain → Queue → Worker
        let mut node_supervisor = NodeSupervisor::new(u64::from(config.domain_id));
        node_supervisor
            .spawn_domain(u64::from(config.domain_id))
            .map_err(|e| RuntimeError::InvalidConfig(format!("supervisor init: {e}")))?;
        let mut domain_supervisor = DomainSupervisor::new(config.domain_id);
        domain_supervisor
            .spawn_queue(u64::from(config.queue_id))
            .map_err(|e| RuntimeError::InvalidConfig(format!("supervisor init: {e}")))?;
        let mut queue_supervisor = QueueSupervisor::with_config(
            config.queue_id,
            SupervisorConfig {
                restart_strategy: config.restart_strategy,
                ..Default::default()
            },
        );
        let initial_worker_id = u64::from(config.queue_id);
        queue_supervisor
            .spawn_worker(initial_worker_id)
            .map_err(|e| RuntimeError::InvalidConfig(format!("supervisor init: {e}")))?;

        // 内部拓扑图：以能力检测结果构造本机资源节点（重规划输入）
        let mut topology_graph: RuntimeGraph<64, 256> = RuntimeGraph::new();
        let node = ResourceNode::new(
            0,
            ExecutionDomain::DataPlane,
            (capability_snapshot.cpu_cores as u32).max(1),
            (capability_snapshot.locked_memory_kb / 1024)
                .clamp(4096, u64::from(u32::MAX)) as u32,
            (capability_snapshot.nic_queues as u32).max(1),
        )
        .with_numa(0);
        topology_graph
            .add_node(node)
            .map_err(|e| RuntimeError::InvalidConfig(format!("topology init: {e}")))?;
        let domain_snapshots = vec![DomainSnapshot::from_domain(ExecutionDomain::DataPlane, 1)];

        // 热阈值 10：访问计数衰减不动点约 20，阈值 100 永远不可达（预热路径永不触发）
        let auto_optimizer = Arc::new(AutoOptimizer::new(1000).with_hot_threshold(10));
        // repr(u8) 显式判别值：to_u8() 与强转等价但不再依赖声明顺序
        let qos_level = Arc::new(AtomicU8::new(QosLevel::Normal.to_u8()));
        let control_running = Arc::new(AtomicBool::new(false));
        let planner = ExtremePlanner;

        Ok(Self {
            config,
            state: RuntimeState::Created,
            #[cfg(target_os = "linux")]
            xdp_loader: None,
            #[cfg(target_os = "linux")]
            dual_bank: None,
            worker,
            extra_workers: Vec::new(),
            admission_rules: Vec::new(),
            #[cfg(target_os = "linux")]
            udp_sink: None,
            ledger,
            quarantined_frames: AtomicU64::new(0),
            stats: RuntimeStats::default(),
            cycle_count: AtomicU64::new(0),
            auto_optimizer,
            qos_level,
            control_running,
            control_thread: None,
            planner,
            affinity_config: None,
            last_topology: None,
            replan_pending: Arc::new(AtomicBool::new(false)),
            target_worker_count: Arc::new(AtomicU64::new(1)),
            topology_graph,
            domain_snapshots,
            cache_warmup_hook: Arc::new(Mutex::new(None)),
            last_applied_qos: QosLevel::Normal,
            replan_apply_error: None,
            node_supervisor,
            domain_supervisor,
            queue_supervisor,
            supervised_worker_ids: vec![initial_worker_id],
            next_supervised_id: initial_worker_id.saturating_add(1),
            last_restart_backoff: None,
            pending_restarts: Vec::new(),
            capability_snapshot,
            data_plane_profile,
            queue_ledger_name,
            ledger_synced_frames: 0,
            metrics: None,
            fault_recorder: None,
            metrics_last_processed: 0,
            metrics_last_rejected: 0,
            metrics_last_dropped: 0,
        })
    }

    /// 注入指标采集器（治理闭环：run_cycle / 控制线程 / 故障处理真实上报）
    ///
    /// 可选注入：默认 `None` 零开销（每周期仅一次分支判断）。
    /// 注入后每周期上报处理包数 / 丢包 / 准入拒绝 / p99 延迟，
    /// 控制线程上报 QoS 等级变更、replan 事件与 worker 扩缩事件。
    pub fn with_metrics(mut self, metrics: Arc<Mutex<MetricsCollector>>) -> Self {
        self.metrics = Some(metrics);
        self
    }

    /// 注入故障记录器（故障处理路径真实记录 `FaultSnapshot`）
    ///
    /// 可选注入：默认 `None` 零开销。
    pub fn with_fault_recorder(mut self, recorder: Arc<Mutex<FaultRecorder>>) -> Self {
        self.fault_recorder = Some(recorder);
        self
    }

    /// 初始化 eBPF 组件（加载 XDP 程序、挂载到网卡）
    #[cfg(target_os = "linux")]
    pub fn init_ebpf(&mut self) -> Result<(), RuntimeError> {
        // 加载 XDP 程序
        let mut loader = XdpLoader::load(self.config.ebpf_program)?;

        // 挂载 XDP 到网卡
        let ifindex = self.config.ifindex as i32;
        loader.attach_xdp(ifindex)?;

        self.xdp_loader = Some(loader);
        self.state = RuntimeState::Initialized;
        Ok(())
    }

    /// 初始化 eBPF 组件（非 Linux：fail-closed 不支持）
    #[cfg(not(target_os = "linux"))]
    pub fn init_ebpf(&mut self) -> Result<(), RuntimeError> {
        Err(RuntimeError::EbpfError(
            "eBPF/XDP 数据面仅 Linux 支持，当前平台不可用".into(),
        ))
    }

    /// 初始化双 Bank 热更新能力
    ///
    /// 按构造时检测的能力快照门控：环境不支持 BPF 时 fail-closed 返回错误，
    /// 而不是创建出静默降级的空管理器。
    #[cfg(target_os = "linux")]
    pub fn init_dual_bank(&mut self) -> Result<(), RuntimeError> {
        if !self.config.enable_dual_bank {
            return Ok(());
        }
        if !self.capability_snapshot.bpf_supported {
            return Err(RuntimeError::EbpfError(
                "当前环境不支持 BPF，无法启用双 Bank 热更新".into(),
            ));
        }

        let manager = DualBankManager::new()?;
        self.dual_bank = Some(manager);

        Ok(())
    }

    /// 初始化双 Bank 热更新能力（非 Linux：fail-closed 不支持）
    #[cfg(not(target_os = "linux"))]
    pub fn init_dual_bank(&mut self) -> Result<(), RuntimeError> {
        if !self.config.enable_dual_bank {
            return Ok(());
        }
        Err(RuntimeError::EbpfError(
            "eBPF 双 Bank 热更新仅 Linux 支持，当前平台不可用".into(),
        ))
    }

    /// 注入 L7 UDP 数据出口（仅 Linux 数据面）
    ///
    /// 注入后 `run_cycle` 自动切换为 `process_cycle_with_udp_sink`，
    /// 准入放行的 UDP 数据报经此 sink 交付 L7 服务（如 QUIC/H3）。
    /// 须在 `start()` 前调用；运行中注入将在下一周期生效。
    ///
    /// # 参数
    /// * `sink` - UDP 数据报消费者（`FnMut(UdpDatagram) + Send`）
    #[cfg(target_os = "linux")]
    pub fn set_udp_sink(&mut self, sink: Box<dyn FnMut(zenith_net::worker::UdpDatagram) + Send>) {
        self.udp_sink = Some(sink);
    }

    /// 启动运行时
    pub fn start(&mut self) -> Result<(), RuntimeError> {
        self.worker.start();
        for w in &mut self.extra_workers {
            w.start();
        }
        self.state = RuntimeState::Running;

        // 启动自动优化控制线程（spawn 失败回滚状态并返回 Err）
        if let Err(e) = self.start_control_thread() {
            // fail-closed：控制线程启动失败时回滚到 Created，
            // 避免调用方误判为已启动后 run_cycle 状态门控通过但治理功能静默失效
            self.worker.stop();
            for w in &mut self.extra_workers {
                w.stop();
            }
            self.state = RuntimeState::Created;
            return Err(e);
        }

        Ok(())
    }

    /// 停止运行时
    pub fn stop(&mut self) {
        // 停止自动优化控制线程
        self.stop_control_thread();

        self.worker.stop();
        for w in &mut self.extra_workers {
            w.stop();
        }
        self.state = RuntimeState::Stopped;

        // 三级 Supervisor 全量关闭（Queue → Domain → Node）
        self.queue_supervisor.shutdown();
        self.domain_supervisor.shutdown();
        self.node_supervisor.shutdown();

        // 清理 eBPF 资源（仅 Linux 数据面）
        #[cfg(target_os = "linux")]
        if let Some(loader) = &mut self.xdp_loader {
            let _ = loader.destroy_link();
        }
    }

    /// 执行一个数据面处理周期
    ///
    /// 多 worker 轮询驱动：主 worker + 全部扩容 worker 各执行一个数据面周期，
    /// 处理包数汇总返回；任一 worker 故障经三级 Supervisor 上报并按策略
    /// 登记非阻塞退避重启（本周期返回 Err，重启在后续周期到点执行）。
    pub fn run_cycle(&mut self) -> Result<u32, RuntimeError> {
        if !self.state.is_running() {
            return Err(RuntimeError::NotRunning);
        }

        // 非阻塞退避：执行已到点的重启计划（不到点跳过，绝不 sleep）
        self.process_pending_restarts();

        // 多 worker 轮询驱动：跳过非 Running（待重启/已停止）的 worker；
        // Linux 且注入了 L7 sink 时，准入放行的 UDP 数据报交付 L7 服务
        let mut total_processed = 0u32;
        let mut fault: Option<(usize, zenith_net::error::NetError)> = None;

        #[cfg(target_os = "linux")]
        {
            let sink = &mut self.udp_sink;
            if self.worker.state() == WorkerState::Running {
                let r = if let Some(s) = sink.as_mut() {
                    self.worker.process_cycle_with_udp_sink(s)
                } else {
                    self.worker.process_cycle()
                };
                match r {
                    Ok(p) => total_processed = p,
                    Err(e) => fault = Some((0, e)),
                }
            }
            if fault.is_none() {
                for (i, w) in self.extra_workers.iter_mut().enumerate() {
                    if w.state() != WorkerState::Running {
                        continue;
                    }
                    let r = if let Some(s) = sink.as_mut() {
                        w.process_cycle_with_udp_sink(s)
                    } else {
                        w.process_cycle()
                    };
                    match r {
                        Ok(p) => total_processed = total_processed.saturating_add(p),
                        Err(e) => {
                            fault = Some((i.saturating_add(1), e));
                            break;
                        }
                    }
                }
            }
        }
        #[cfg(not(target_os = "linux"))]
        {
            if self.worker.state() == WorkerState::Running {
                match self.worker.process_cycle() {
                    Ok(p) => total_processed = p,
                    Err(e) => fault = Some((0, e)),
                }
            }
            if fault.is_none() {
                for (i, w) in self.extra_workers.iter_mut().enumerate() {
                    if w.state() != WorkerState::Running {
                        continue;
                    }
                    match w.process_cycle() {
                        Ok(p) => total_processed = total_processed.saturating_add(p),
                        Err(e) => {
                            fault = Some((i.saturating_add(1), e));
                            break;
                        }
                    }
                }
            }
        }

        if let Some((worker_index, e)) = fault {
            self.handle_worker_fault(worker_index);
            return Err(e.into());
        }
        let processed = total_processed;

        // QoS 等级真实生效：按当前等级调整全部 Worker 准入严格度
        self.apply_qos_to_worker();

        // 资源账本与帧池对账（allocate_in_child / release_in_child 分层 roll-up）
        self.sync_ledger_with_pool();

        // 消费控制线程的重规划请求（真正执行 replan）
        self.run_pending_replan();

        // 按控制线程计算的目标 worker 数真实扩缩容（Supervisor 记账同步维护）
        self.reconcile_worker_count();

        // 更新统计（多 worker 聚合视图）
        let wstats = self.aggregated_worker_stats();
        self.update_runtime_stats(&wstats);

        // 治理闭环：每周期上报处理包数/丢包/准入拒绝/p99 延迟（未注入零开销）
        self.report_cycle_metrics(&wstats);

        // 周期计数
        let cycles = self.cycle_count.fetch_add(1, Ordering::Relaxed).saturating_add(1);
        self.stats.cycles_completed = cycles;

        // 向自动优化器记录负载样本（热路径操作，仅一次原子写入）
        self.record_load_sample(&wstats);

        // 检查最大周期数限制
        if self.config.max_cycles > 0 && cycles >= self.config.max_cycles {
            self.stop();
        }

        Ok(processed)
    }

    /// 执行完整的启动-运行-停止生命周期
    pub fn run_full_lifecycle(&mut self, cycles: u64) -> Result<u64, RuntimeError> {
        if cycles == 0 {
            return Err(RuntimeError::InvalidConfig("cycles must be > 0".into()));
        }

        self.start()?;
        // 确保 run_cycle 失败时 stop() 仍被调用，避免控制线程泄漏。
        // Drop 会兜底调用 stop()，但若调用方持有 WorkerRuntime 不释放，
        // 控制线程将持续运行。
        let result = (|| {
            for _ in 0..cycles {
                self.run_cycle()?;
            }
            Ok(self.stats.total_packets)
        })();
        self.stop();
        result
    }

    // ==================== eBPF 热更新 ====================

    /// 执行 eBPF 热切换（双 Bank 原子替换）
    ///
    /// 全程经 [`ChangeSet`] 状态机驱动：
    /// `Prepare → Validate → ResourceProof → Shadow → Activate → StopOldAdmission → Drain → Commit`，
    /// 内核态替换失败时执行 `rollback` 回到激活前生成号（fail-closed）。
    #[cfg(target_os = "linux")]
    pub fn switch_ebpf_bank(&mut self) -> Result<BankId, RuntimeError> {
        let manager = self
            .dual_bank
            .as_mut()
            .ok_or(RuntimeError::DualBankNotEnabled)?;

        let active_gen = u64::from(manager.global_version());
        let target_gen = active_gen.saturating_add(1);
        let mut cs = ChangeSet::new(active_gen, target_gen);

        cs.prepare()
            .map_err(|e| RuntimeError::EbpfError(format!("changeset prepare: {e}")))?;
        cs.validate()
            .map_err(|e| RuntimeError::EbpfError(format!("changeset validate: {e}")))?;
        // 资源证明：两个 Bank 的 eBPF 程序都必须真实加载
        cs.resource_proof(|| {
            if manager.is_ebpf_fully_loaded() {
                Ok(())
            } else {
                Err("ebpf banks not fully loaded")
            }
        })
        .map_err(|e| RuntimeError::EbpfError(format!("changeset resource proof: {e}")))?;
        // 阴影验证：活动 Bank 在线（双写前提）
        cs.shadow(|| {
            if manager.is_ebpf_active_bank_loaded() {
                Ok(())
            } else {
                Err("active bank not loaded")
            }
        })
        .map_err(|e| RuntimeError::EbpfError(format!("changeset shadow: {e}")))?;
        cs.activate(active_gen)
            .map_err(|e| RuntimeError::EbpfError(format!("changeset activate: {e}")))?;

        // 原子激活：内核态零丢包替换；失败 → 回滚到激活前生成号
        let ifindex = self.config.ifindex as i32;
        match manager.switch(ifindex) {
            Ok(new_bank) => {
                // 后续 stop_old_admission/drain/commit 任一失败 → 回滚 ChangeSet（fail-closed）
                if let Err(e) = cs
                    .stop_old_admission()
                    .and_then(|_| cs.drain(|| Ok(None)))
                    .and_then(|_| cs.commit())
                {
                    let _ = cs.rollback(None);
                    return Err(RuntimeError::EbpfError(format!("changeset post-switch: {e}")));
                }
                manager.increment_version();
                self.stats.ebpf_switches = self.stats.ebpf_switches.saturating_add(1);
                self.stats.active_bank = Some(new_bank);
                Ok(new_bank)
            }
            Err(e) => {
                // 回滚到激活前生成号（ChangeSet 语义保证）
                let _outcome = cs.rollback(None);
                Err(RuntimeError::from(e))
            }
        }
    }

    /// 获取当前活动的 Bank ID
    #[cfg(target_os = "linux")]
    pub fn active_bank(&self) -> Option<BankId> {
        self.dual_bank.as_ref().map(|m| m.active_bank_id())
    }

    /// 获取当前 Bank 切换次数
    #[cfg(target_os = "linux")]
    pub fn ebpf_switch_count(&self) -> u32 {
        self.dual_bank.as_ref().map(|m| m.switch_count()).unwrap_or(0)
    }

    /// 获取当前 Bank 切换次数（非 Linux：恒 0）
    #[cfg(not(target_os = "linux"))]
    pub fn ebpf_switch_count(&self) -> u32 {
        0
    }

    // ==================== 准入引擎管理 ====================

    /// 添加准入规则
    ///
    /// 规则应用到全部数据面 worker（多 worker 准入一致），并记录于
    /// `admission_rules`：后续扩容的新 worker 构造时重放全部规则。
    pub fn add_admission_rule(
        &mut self,
        rule: AdmissionRule,
    ) -> Result<(), RuntimeError> {
        self.worker.admission_mut().add_rule(rule)?;
        for w in &mut self.extra_workers {
            w.admission_mut().add_rule(rule)?;
        }
        self.admission_rules.push(rule);
        Ok(())
    }

    /// 获取可变准入引擎引用
    pub fn admission_mut(&mut self) -> &mut SourceAdmissionEngine {
        self.worker.admission_mut()
    }

    /// 获取准入引擎引用
    pub fn admission(&self) -> &SourceAdmissionEngine {
        self.worker.admission()
    }

    /// 批量添加准入规则（应用到全部 worker 并记录，供扩容重放）
    pub fn add_admission_rules(
        &mut self,
        rules: Vec<AdmissionRule>,
    ) -> Result<(), RuntimeError> {
        for rule in rules {
            self.add_admission_rule(rule)?;
        }
        Ok(())
    }

    // ==================== Quarantine 隔离区 ====================

    /// 隔离一个异常帧
    ///
    /// 真正调用 `FramePool::quarantine_by_id` 将帧状态迁移至 `FrameState::Quarantine`，
    /// 隔离后的帧不再参与分配 / 收发，但仍保留在池中可被显式 `recover_frame` 恢复。
    ///
    /// # 参数
    /// * `frame_id` - 待隔离的帧 ID
    /// * `reason` - 隔离原因（用于审计日志与 FrameInfo.reason 字段）
    pub fn quarantine_frame(
        &mut self,
        frame_id: zenith_foundation::FrameId,
        reason: &str,
    ) -> Result<(), RuntimeError> {
        if !self.config.enable_quarantine {
            return Ok(());
        }
        // 先查询当前代际，quarantine_by_id 会严格校验代际一致性防止陈旧 FrameID 误用
        let expected_gen = self
            .worker
            .frame_pool()
            .get_frame_info(frame_id)
            .ok_or_else(|| RuntimeError::EbpfError(format!("frame {} not found", frame_id.value())))?
            .generation();
        // 真正迁移帧状态到 Quarantine；代际与所有权由 FramePool 内部校验
        self.worker
            .frame_pool_mut()
            .quarantine_by_id(frame_id, expected_gen, reason)
            .map_err(|e| RuntimeError::EbpfError(format!("quarantine failed: {e:?}")))?;
        self.quarantined_frames.fetch_add(1, Ordering::SeqCst);
        self.stats.total_quarantined = self.quarantined_frames.load(Ordering::SeqCst);
        Ok(())
    }

    /// 获取隔离帧数量
    ///
    /// 直接读取 `FramePool::quarantined_count()`（单一真相源），
    /// 同时与内部 AtomicU64 计数器交叉校验以发现任何漂移。
    pub fn quarantined_count(&self) -> u64 {
        // 以 FramePool 为权威源；AtomicU64 仅为热路径快速读取的近似值
        self.worker.frame_pool().quarantined_count() as u64
    }

    /// 恢复指定帧从隔离区
    ///
    /// 真正调用 `FramePool::recover_from_quarantine` 将帧状态从 `Quarantine` 迁回 `Free`，
    /// 恢复后该帧可重新被分配使用。代际不会前进（避免误将旧代际帧当作新代际使用）。
    pub fn recover_frame(&mut self, frame_id: zenith_foundation::FrameId) -> Result<(), RuntimeError> {
        self.worker
            .frame_pool_mut()
            .recover_from_quarantine(frame_id)
            .map_err(|e| RuntimeError::EbpfError(format!("recover failed: {e:?}")))?;
        // CAS 饱和递减：替代 fetch_sub + store 的瞬态下溢窗口，
        // 与 task_channel::dec_pending 的 CAS 模式对齐（不 panic、不下溢回绕 u64::MAX）。
        let mut current = self.quarantined_frames.load(Ordering::SeqCst);
        loop {
            let new = current.saturating_sub(1);
            match self.quarantined_frames.compare_exchange(
                current,
                new,
                Ordering::SeqCst,
                Ordering::SeqCst,
            ) {
                Ok(_) => break,
                Err(actual) => current = actual,
            }
        }
        Ok(())
    }

    /// 恢复所有隔离帧
    ///
    /// 遍历 `FramePool` 中所有 `FrameState::Quarantine` 的帧，逐个迁移回 `Free` 状态。
    /// 返回实际恢复的帧数。
    pub fn recover_all_quarantined(&mut self) -> Result<u64, RuntimeError> {
        let mut recovered: u64 = 0;
        // 收集所有 Quarantine 状态的 frame_id（避免在迭代中修改 frame_pool）
        let quarantined_ids: Vec<u32> = self
            .worker
            .frame_pool()
            .frames()
            .iter()
            .filter(|info| info.state() == zenith_foundation::FrameState::Quarantine)
            .map(|info| info.id().value())
            .collect();

        for fid in quarantined_ids {
            // recover_from_quarantine 内部会校验状态合法性
            if self
                .worker
                .frame_pool_mut()
                .recover_from_quarantine(zenith_foundation::FrameId::new(fid))
                .is_ok()
            {
                recovered += 1;
            }
        }

        // 遍历 extra_workers 的帧池，恢复隔离帧（与主 Worker 同语义）
        for w in &mut self.extra_workers {
            let extra_quarantined_ids: Vec<u32> = w
                .frame_pool()
                .frames()
                .iter()
                .filter(|info| info.state() == zenith_foundation::FrameState::Quarantine)
                .map(|info| info.id().value())
                .collect();
            for fid in extra_quarantined_ids {
                if w
                    .frame_pool_mut()
                    .recover_from_quarantine(zenith_foundation::FrameId::new(fid))
                    .is_ok()
                {
                    recovered += 1;
                }
            }
        }

        // 重置计数器为全部 Worker FramePool 真实剩余隔离数之和（与 FramePool 真相一致）
        let total_quarantined = self
            .extra_workers
            .iter()
            .fold(
                u64::from(self.worker.frame_pool().quarantined_count()),
                |acc, w| acc.saturating_add(u64::from(w.frame_pool().quarantined_count())),
            );
        self.quarantined_frames
            .store(total_quarantined, Ordering::SeqCst);
        Ok(recovered)
    }

    // ==================== 资源账本 ====================

    /// 获取资源账本引用
    pub fn ledger(&self) -> &ResourceLedger {
        &self.ledger
    }

    /// 获取帧池引用
    pub fn frame_pool(&self) -> &FramePool {
        self.worker.frame_pool()
    }

    /// 获取帧池可变引用（用于外部主动分配 / 隔离 / 恢复帧）
    pub fn frame_pool_mut(&mut self) -> &mut FramePool {
        self.worker.frame_pool_mut()
    }

    /// 获取当前已完成的循环周期数（pub 访问器）
    #[inline]
    pub fn cycle_count(&self) -> u64 {
        self.cycle_count.load(Ordering::Relaxed)
    }

    /// 验证所有权守恒（全部数据面 worker 逐一校验）
    pub fn verify_conservation(&self) -> bool {
        self.worker.verify_conservation()
            && self.extra_workers.iter().all(|w| w.verify_conservation())
    }

    // ==================== 状态查询 ====================

    /// 获取运行时状态
    pub fn state(&self) -> RuntimeState {
        self.state
    }

    /// 获取完整统计信息
    pub fn stats(&self) -> RuntimeStats {
        self.stats.clone()
    }

    /// 获取 Worker ID
    pub fn worker_id(&self) -> u32 {
        self.worker.id()
    }

    /// 获取 Worker 状态
    pub fn worker_state(&self) -> zenith_net::worker::WorkerState {
        self.worker.state()
    }

    /// 获取 Worker 统计
    pub fn worker_stats(&self) -> WorkerStats {
        self.worker.stats()
    }

    /// 获取配置引用
    pub fn config(&self) -> &WorkerRuntimeConfig {
        &self.config
    }

    /// 获取 XDP 加载器引用
    #[cfg(target_os = "linux")]
    pub fn xdp_loader(&self) -> Option<&XdpLoader> {
        self.xdp_loader.as_ref()
    }

    /// 获取双 Bank 管理器引用
    #[cfg(target_os = "linux")]
    pub fn dual_bank_manager(&self) -> Option<&DualBankManager> {
        self.dual_bank.as_ref()
    }

    // ==================== 自动优化集成 ====================

    /// 获取当前 QoS 等级（热路径使用 Relaxed 读取，零开销）
    #[inline]
    pub fn current_qos_level(&self) -> QosLevel {
        QosLevel::from_priority(self.qos_level.load(Ordering::Relaxed))
    }

    /// 设置 QoS 等级
    pub fn set_qos_level(&self, level: QosLevel) {
        // repr(u8) 显式判别值：to_u8() 与强转等价但不再依赖声明顺序
        self.qos_level.store(level.to_u8(), Ordering::Relaxed);
        self.auto_optimizer.set_qos_level(level);
    }

    /// 获取自动优化器引用
    pub fn auto_optimizer(&self) -> &Arc<AutoOptimizer> {
        &self.auto_optimizer
    }

    /// 向自动优化器记录负载样本
    #[inline]
    pub fn record_load_sample(&self, stats: &WorkerStats) {
        let sample = LoadSample {
            timestamp: Instant::now(),
            cpu_usage: stats.cpu_usage,
            queue_depth: stats.queue_depth,
            latency_p99: stats.latency_p99,
            throughput: stats.throughput,
        };
        self.auto_optimizer.record_sample(sample);
    }

    /// 记录缓存访问模式（用于缓存预测预热）
    #[inline]
    pub fn record_cache_access(&self, key: &str) {
        self.auto_optimizer.record_cache_access(key);
    }

    /// 获取预测的热缓存键列表（`Box<str>` 零成本 Clone）
    pub fn get_hot_cache_keys(&self) -> Vec<Box<str>> {
        self.auto_optimizer.get_hot_cache_keys()
    }

    /// 注册缓存预热回调
    ///
    /// 控制线程每轮评估时将预测热键推送给该回调（真实消费路径），
    /// 嵌入方可在回调中执行实际的缓存预热（如 zenith-cache 预加载）。
    pub fn set_cache_warmup_hook(&self, hook: CacheWarmupHook) {
        if let Ok(mut guard) = self.cache_warmup_hook.lock() {
            *guard = Some(hook);
        }
    }

    // ==================== Supervisor / 能力 / 规划观测 ====================

    /// 节点级健康报告（三级 Supervisor 聚合）
    pub fn supervisor_health(&mut self) -> HealthReport {
        self.node_supervisor.health_check()
    }

    /// 队列 Supervisor 当前在管 worker 数
    pub fn supervised_worker_count(&self) -> usize {
        self.supervised_worker_ids.len()
    }

    /// 当前真实数据面 worker 总数（主 worker + 扩容 worker）
    pub fn worker_count(&self) -> usize {
        self.extra_workers.len().saturating_add(1)
    }

    /// 控制线程按负载趋势计算的目标 worker 数
    pub fn target_worker_count(&self) -> u64 {
        self.target_worker_count.load(Ordering::Relaxed)
    }

    /// 最近一次故障重启的退避时长
    pub fn last_restart_backoff(&self) -> Option<std::time::Duration> {
        self.last_restart_backoff
    }

    /// 上次拓扑规划结果（真实保存的规划输出）
    pub fn last_planned_topology(&self) -> Option<&PlannedTopology> {
        self.last_topology.as_ref()
    }

    /// 重规划亲和应用错误（None = 上次应用成功或尚未重规划）
    pub fn replan_apply_error(&self) -> Option<&str> {
        self.replan_apply_error.as_deref()
    }

    /// 环境能力快照（构造时经 zenith-capability 检测）
    pub fn capability_snapshot(&self) -> &CapabilitySnapshot {
        &self.capability_snapshot
    }

    /// 数据面 Profile（按能力快照选择）
    pub fn data_plane_profile(&self) -> Profile {
        self.data_plane_profile
    }

    /// 启动自动优化控制线程
    ///
    /// 控制线程定期执行：
    /// - 评估 QoS 等级并更新原子标记（run_cycle 将其应用于 Worker 准入严格度）
    /// - 到期时置位 `replan_pending`（run_cycle 消费并真正执行拓扑重规划）
    /// - 按负载趋势计算目标 worker 数（run_cycle 真实扩缩容，Supervisor 记账同步）
    /// - 将预测热键推送给已注册的缓存预热回调
    /// - 治理闭环：上报 QoS 等级变更、replan 事件、worker 扩缩事件（未注入零开销）
    ///
    /// # Errors
    /// 线程 spawn 失败时回滚 `control_running` 并返回 `RuntimeError::WorkerError`
    fn start_control_thread(&mut self) -> Result<(), RuntimeError> {
        if self.control_running.load(Ordering::SeqCst) {
            return Ok(());
        }
        self.control_running.store(true, Ordering::SeqCst);

        let optimizer = self.auto_optimizer.clone();
        let qos_level = self.qos_level.clone();
        let running = self.control_running.clone();
        let replan_pending = self.replan_pending.clone();
        let target_worker_count = self.target_worker_count.clone();
        let warmup_hook = self.cache_warmup_hook.clone();
        let metrics = self.metrics.clone();

        match std::thread::Builder::new()
            .name("zenith-auto-optimizer".to_string())
            .spawn(move || {
                // 上次已上报的 QoS 等级（仅变更时上报，避免每轮重复计数）
                let mut last_reported_qos: Option<QosLevel> = None;
                while running.load(Ordering::SeqCst) {
                    // QoS 评估 + 标记更新（Worker 热路径 Relaxed 读取）
                    // repr(u8) 显式判别值：to_u8() 与强转等价但不再依赖声明顺序
                    let new_level = optimizer.evaluate_qos();
                    qos_level.store(new_level.to_u8(), Ordering::Relaxed);

                    // 到期 → 置位重规划标记（由 run_cycle 真正执行 replan_topology）
                    let replan_due = optimizer.should_replan();
                    if replan_due {
                        replan_pending.store(true, Ordering::SeqCst);
                    }

                    // 按负载趋势计算目标 worker 数（run_cycle 真实扩缩容）
                    let current = target_worker_count.load(Ordering::Relaxed) as usize;
                    let optimal = optimizer.get_optimal_worker_count(current.max(1));
                    target_worker_count.store(optimal.max(1) as u64, Ordering::Relaxed);

                    // 治理闭环：QoS 等级变更 / replan 事件 / worker 扩缩事件上报
                    if let Some(mtx) = &metrics
                        && let Ok(mut m) = mtx.lock()
                    {
                        if last_reported_qos != Some(new_level) {
                            let _ = m.counter_inc(
                                "zenith_runtime_qos_level_changes_total",
                                &[("level", new_level.as_str())],
                            );
                            let _ = m.gauge_set(
                                "zenith_runtime_qos_level",
                                i64::from(new_level.to_u8()),
                                &[],
                            );
                            last_reported_qos = Some(new_level);
                        }
                        if replan_due {
                            let _ = m.counter_inc("zenith_runtime_replan_total", &[]);
                        }
                        if optimal != current {
                            let direction = if optimal > current { "up" } else { "down" };
                            let _ = m.counter_inc(
                                "zenith_runtime_worker_scale_events_total",
                                &[("direction", direction)],
                            );
                            let _ = m.gauge_set(
                                "zenith_runtime_target_workers",
                                (optimal as u64).min(i64::MAX as u64) as i64,
                                &[],
                            );
                        }
                    }

                    // 缓存预热：将预测热键推送给注册的回调（真实消费路径）
                    let hot_keys = optimizer.get_hot_cache_keys();
                    if !hot_keys.is_empty()
                        && let Ok(guard) = warmup_hook.lock()
                        && let Some(hook) = guard.as_ref()
                    {
                        hook(&hot_keys);
                    }

                    // 自适应轮询间隔：按 QoS 等级动态调整
                    let poll_interval_ms = match new_level {
                        QosLevel::Critical => 10,        // Critical: 10ms
                        QosLevel::High | QosLevel::Normal => 100,  // Normal: 100ms
                        QosLevel::Low | QosLevel::Background => 1000,  // Background: 1000ms
                    };
                    std::thread::sleep(std::time::Duration::from_millis(poll_interval_ms));
                }
            }) {
            Ok(handle) => {
                self.control_thread = Some(handle);
                Ok(())
            }
            Err(e) => {
                // spawn 失败：回滚运行标志并上报错误（不得吞错留下假运行状态）
                self.control_running.store(false, Ordering::SeqCst);
                Err(RuntimeError::WorkerError(format!(
                    "failed to spawn control thread: {e}"
                )))
            }
        }
    }

    /// 停止自动优化控制线程
    fn stop_control_thread(&mut self) {
        self.control_running.store(false, Ordering::SeqCst);
        if let Some(handle) = self.control_thread.take() {
            let _ = handle.join();
        }
    }

    /// 应用 CPU 亲和配置
    pub fn apply_affinity(&mut self, config: CpuAffinityConfig) -> Result<(), CpuAffinityError> {
        let result = config.apply();
        if result.is_ok() {
            self.affinity_config = Some(config);
        }
        result
    }

    /// 基于拓扑规划结果重新应用 CPU 亲和
    ///
    /// 规划结果真实生效：
    /// 1. 从 `PlannedTopology` 取 DataPlane 域中本 Worker 的分配（节点/NUMA）
    /// 2. 按分配节点推导 CPU 亲和并实际调用 `sched_setaffinity`
    /// 3. 规划结果保存于 `last_topology`，供后续 worker 分配查询
    pub fn replan_topology(
        &mut self,
        graph: &RuntimeGraph<64, 256>,
        snapshots: &[DomainSnapshot],
    ) -> Result<(), RuntimeError> {
        let topology = self.planner.plan(graph, snapshots)
            .map_err(|e| RuntimeError::WorkerError(e.to_string()))?;

        self.apply_planned_topology(topology)
    }

    /// 将规划结果真实应用于 CPU 亲和与 worker 分配记录
    fn apply_planned_topology(&mut self, topology: PlannedTopology) -> Result<(), RuntimeError> {
        // 从规划结果取 DataPlane 域中本 Worker 的分配
        if let Some(assignment) = topology.find_assignment(ExecutionDomain::DataPlane)
            && !assignment.workers.is_empty()
        {
            let idx = (self.config.queue_id as usize) % assignment.workers.len();
            let planned = &assignment.workers[idx];
            // SYS-016：规划直接携带按 worker 分布的 CPU 编号——以全局唯一的
            // worker_id（而非 node_id）对在线 CPU 取模，避免同一规划节点上的
            // 多 worker 因 node_id 相同而映射到同一核心（原 node_id%online 缺陷）。
            // 越界校验由 `CpuAffinityConfig::apply` 在应用前 fail-closed 完成。
            let online = crate::cpu_affinity::get_online_cpus().max(1);
            let cpu_id = (planned.worker_id as usize) % online;
            let affinity = CpuAffinityConfig::new(
                self.config.queue_id,
                vec![cpu_id],
                planned.numa_node,
            );
            // 亲和应用失败不阻断数据面，但记录错误（fail-visible，不吞错）
            match self.apply_affinity(affinity) {
                Ok(()) => self.replan_apply_error = None,
                Err(e) => self.replan_apply_error = Some(e.to_string()),
            }
        }

        // 规划结果真实保存（供后续 worker 分配与外部查询）
        self.last_topology = Some(topology);
        Ok(())
    }

    /// 消费控制线程置位的重规划请求并真正执行
    fn run_pending_replan(&mut self) {
        if !self.replan_pending.swap(false, Ordering::SeqCst) {
            return;
        }
        let planner = ExtremePlanner;
        let topology = match planner.plan(&self.topology_graph, &self.domain_snapshots) {
            Ok(t) => t,
            Err(e) => {
                self.replan_apply_error = Some(e.to_string());
                return;
            }
        };
        let _ = self.apply_planned_topology(topology);
    }

    /// QoS 等级真实生效路径：按当前等级调整全部 Worker 准入严格度
    ///
    /// - Critical/High：拒绝与解析错误的帧一律隔离（最严格准入）
    /// - Normal：仅隔离解析错误帧
    /// - Low/Background：不隔离（最大吞吐）
    fn apply_qos_to_worker(&mut self) {
        let level = self.current_qos_level();
        if level == self.last_applied_qos {
            return;
        }
        self.last_applied_qos = level;
        let (on_error, on_deny) = match level {
            QosLevel::Critical | QosLevel::High => (true, true),
            QosLevel::Normal => (true, false),
            QosLevel::Low | QosLevel::Background => (false, false),
        };
        self.worker.set_quarantine_on_error(on_error);
        self.worker.set_quarantine_on_deny(on_deny);
        for w in &mut self.extra_workers {
            w.set_quarantine_on_error(on_error);
            w.set_quarantine_on_deny(on_deny);
        }
    }

    /// 资源账本与帧池对账：帧分配/释放经 `allocate_in_child`/`release_in_child`
    /// 分层 roll-up 到域账本，使账本 `used` 始终反映全部帧池真实占用（多 worker 汇总）。
    fn sync_ledger_with_pool(&mut self) {
        let actual = self
            .extra_workers
            .iter()
            .fold(u64::from(self.worker.frame_pool().allocated_count()), |acc, w| {
                acc.saturating_add(u64::from(w.frame_pool().allocated_count()))
            });
        let path = [self.queue_ledger_name.as_str()];
        if actual > self.ledger_synced_frames {
            let delta = actual - self.ledger_synced_frames;
            if self
                .ledger
                .allocate_in_child(&path, ResourceType::Frames, delta)
                .is_ok()
            {
                self.ledger_synced_frames = actual;
            }
            // 配额不足时保持旧值，下周期重试（fail-visible：ledger.used 可读）
        } else if actual < self.ledger_synced_frames {
            let delta = self.ledger_synced_frames - actual;
            if self
                .ledger
                .release_in_child(&path, ResourceType::Frames, delta)
                .is_ok()
            {
                self.ledger_synced_frames = actual;
            }
        }
    }

    /// 按控制线程计算的目标 worker 数落实真实扩缩容：
    /// 扩容真实构造数据面 Worker，缩容真实停止并移除；
    /// Supervisor 子项记账作为治理视图同步维护。
    fn reconcile_worker_count(&mut self) {
        // 至少保留主 worker（target 下限 1）
        let target = (self.target_worker_count.load(Ordering::Relaxed) as usize).max(1);
        let current = self.worker_count();
        if target > current {
            for _ in current..target {
                if self.spawn_worker().is_err() {
                    break;
                }
            }
        } else if target < current {
            for _ in target..current {
                // 缩容：从尾部停止并移除（Supervisor 退避语义由故障路径负责）
                self.remove_worker_at(self.worker_count().saturating_sub(1));
            }
        }
        // 健康状态沿层级向上聚合
        let _ = self.domain_supervisor.health_check();
        let _ = self.node_supervisor.health_check();
        // 扩容后强制对新 Worker 重放当前 QoS 设置：
        // apply_qos_to_worker 在 level == last_applied_qos 时跳过，
        // 但新 Worker 使用默认设置（quarantine_on_error=false），
        // 必须在此强制重放以确保 Critical/High QoS 下新 Worker 正确隔离。
        if target > current {
            self.force_apply_qos_to_new_workers();
        }
    }

    /// 强制对所有 Worker 应用当前 QoS（不检查 last_applied_qos）。
    /// 用于 reconcile_worker_count 扩容后对新 Worker 的 QoS 继承。
    fn force_apply_qos_to_new_workers(&mut self) {
        let level = self.current_qos_level();
        let (on_error, on_deny) = match level {
            QosLevel::Critical | QosLevel::High => (true, true),
            QosLevel::Normal => (true, false),
            QosLevel::Low | QosLevel::Background => (false, false),
        };
        self.worker.set_quarantine_on_error(on_error);
        self.worker.set_quarantine_on_deny(on_deny);
        for w in &mut self.extra_workers {
            w.set_quarantine_on_error(on_error);
            w.set_quarantine_on_deny(on_deny);
        }
    }

    /// 扩容一个真实数据面 Worker：
    /// 1. QueueSupervisor 登记治理视图记账（容量满 fail-closed 不再扩容）
    /// 2. 真实构造 zenith-net Worker（queue_id 按拓扑分配，准入规则重放）
    /// 3. 构造失败回滚 Supervisor 记账（不留幽灵记账）
    fn spawn_worker(&mut self) -> Result<(), RuntimeError> {
        let supervised_id = self.next_supervised_id;
        self.queue_supervisor
            .spawn_worker(supervised_id)
            .map_err(|e| RuntimeError::WorkerError(format!("supervisor spawn: {e}")))?;
        self.next_supervised_id = self.next_supervised_id.saturating_add(1);

        // queue_id 按拓扑分配：主 worker 占 config.queue_id，
        // 扩容 worker 依次占用后续网卡队列（对检测到的队列数取模回绕）
        let worker_index = self.worker_count();
        let nic_queues = (self.capability_snapshot.nic_queues as u32).max(1);
        let queue_id = self
            .config
            .queue_id
            .saturating_add(worker_index as u32)
            % nic_queues;
        let worker_id = supervised_id as u32;

        match self.build_worker(worker_id, queue_id) {
            Ok(mut worker) => {
                worker.start();
                self.extra_workers.push(worker);
                self.supervised_worker_ids.push(supervised_id);
                Ok(())
            }
            Err(e) => {
                // 构造失败：回滚 Supervisor 记账（fail-closed，不留幽灵记账）
                let _ = self.queue_supervisor.stop_worker(supervised_id);
                Err(e)
            }
        }
    }

    /// 构造一个数据面 Worker（Linux：携带 XSK 配置的 AF_XDP Worker；非 Linux：可移植桩），
    /// 并重放已配置的准入规则（多 worker 准入一致）
    #[cfg(target_os = "linux")]
    fn build_worker(&self, worker_id: u32, queue_id: u32) -> Result<Worker, RuntimeError> {
        let xsk_config = XskConfig {
            queue_id,
            ..self.config.xsk_config.clone()
        };
        let mut worker = Worker::new(
            worker_id,
            xsk_config,
            self.config.frame_pool_capacity,
            SourceAdmissionEngine::allow_all(),
        )?;
        for rule in &self.admission_rules {
            worker.admission_mut().add_rule(*rule)?;
        }
        Ok(worker)
    }

    /// 构造一个数据面 Worker（非 Linux 可移植桩），并重放已配置的准入规则
    #[cfg(not(target_os = "linux"))]
    fn build_worker(&self, worker_id: u32, _queue_id: u32) -> Result<Worker, RuntimeError> {
        let mut worker = Worker::new(
            worker_id,
            self.config.frame_pool_capacity,
            SourceAdmissionEngine::allow_all(),
        )?;
        for rule in &self.admission_rules {
            worker.admission_mut().add_rule(*rule)?;
        }
        Ok(worker)
    }

    /// 停止并移除指定索引的 worker（缩容 / Temporary 故障移除共用）
    ///
    /// 索引语义：0 = 主 worker（不可移除，仅停止，fail-closed），
    /// n ≥ 1 = `extra_workers[n - 1]`（真实停止移除 + Supervisor 记账同步）。
    /// 移除后修复待重启计划的索引对齐。
    fn remove_worker_at(&mut self, worker_index: usize) {
        if worker_index == 0 {
            // 主 worker 不可移除：仅停止（保留索引对齐不变式），fail-closed 不再处理报文
            self.worker.stop();
            return;
        }
        let extra_idx = worker_index.saturating_sub(1);
        if extra_idx >= self.extra_workers.len() {
            return;
        }
        let mut worker = self.extra_workers.remove(extra_idx);
        worker.stop();
        if worker_index < self.supervised_worker_ids.len() {
            let supervised_id = self.supervised_worker_ids.remove(worker_index);
            let _ = self.queue_supervisor.stop_worker(supervised_id);
        }
        // 待重启计划索引对齐：被移除索引的计划作废，其后索引前移
        self.pending_restarts.retain_mut(|pr| {
            match pr.worker_index.cmp(&worker_index) {
                std::cmp::Ordering::Less => true,
                std::cmp::Ordering::Equal => false,
                std::cmp::Ordering::Greater => {
                    pr.worker_index = pr.worker_index.saturating_sub(1);
                    true
                }
            }
        });
    }

    /// 停止指定索引的 worker（0 = 主 worker，n ≥ 1 = 扩容 worker）
    fn stop_worker_at(&mut self, worker_index: usize) {
        if worker_index == 0 {
            self.worker.stop();
        } else if let Some(w) = self.extra_workers.get_mut(worker_index.saturating_sub(1)) {
            w.stop();
        }
    }

    /// 启动指定索引的 worker（0 = 主 worker，n ≥ 1 = 扩容 worker）
    fn start_worker_at(&mut self, worker_index: usize) {
        if worker_index == 0 {
            self.worker.start();
        } else if let Some(w) = self.extra_workers.get_mut(worker_index.saturating_sub(1)) {
            w.start();
        }
    }

    /// 非阻塞退避重启：`run_cycle` 每周期检查到期的重启计划并执行
    /// （不到点跳过；禁止 `std::thread::sleep` 阻塞数据面调用线程）
    fn process_pending_restarts(&mut self) {
        if self.pending_restarts.is_empty() {
            return;
        }
        let now = Instant::now();
        let mut i = 0;
        while i < self.pending_restarts.len() {
            if now < self.pending_restarts[i].next_retry_at {
                i = i.saturating_add(1);
                continue;
            }
            let pr = self.pending_restarts.remove(i);
            // 索引可能因缩容失效：越界放弃（fail-closed）
            if pr.worker_index >= self.worker_count() {
                continue;
            }
            // 退避到期：重启数据面 Worker 并沿三级层级恢复健康标记
            self.stop_worker_at(pr.worker_index);
            self.start_worker_at(pr.worker_index);
            let _ = self.queue_supervisor.health_check_worker(pr.supervised_id, true);
            let _ = self
                .domain_supervisor
                .mark_queue_healthy(u64::from(self.config.queue_id), true);
            let _ = self
                .node_supervisor
                .mark_domain_healthy(u64::from(self.config.domain_id), true);
        }
    }

    /// Worker 故障处理：经三级 Supervisor 上报（Queue → Domain → Node），
    /// 按配置的重启策略（Permanent/Transient/Temporary + 指数退避）登记
    /// 非阻塞重启计划（`run_cycle` 到点执行），并上报故障事件到治理闭环。
    ///
    /// 数据面 `process_cycle` 错误属异常退出（`ExitReason::Abnormal`）：
    /// Permanent/Transient 退避重启，Temporary 停止移除。
    fn handle_worker_fault(&mut self, worker_index: usize) {
        let Some(&supervised_id) = self.supervised_worker_ids.get(worker_index) else {
            return;
        };
        let queue_id = u64::from(self.config.queue_id);
        let domain_id = u64::from(self.config.domain_id);

        // 故障沿三级层级上报
        let _ = self.queue_supervisor.health_check_worker(supervised_id, false);
        let _ = self.domain_supervisor.mark_queue_healthy(queue_id, false);
        let _ = self.node_supervisor.mark_domain_healthy(domain_id, false);

        // 治理闭环：故障事件真实上报（FaultRecorder 快照 + 故障计数器）
        if let Some(recorder) = &self.fault_recorder
            && let Ok(mut r) = recorder.lock()
        {
            r.record(FaultType::WorkerPanic, "zenith-runtime.worker");
        }
        if let Some(metrics) = &self.metrics
            && let Ok(mut m) = metrics.lock()
        {
            let _ = m.counter_inc("zenith_runtime_worker_faults_total", &[]);
        }

        // 按策略决策（Temporary / Transient+Normal 下 restart 返回 ZERO 且子项被移除）
        match self
            .queue_supervisor
            .restart_worker(supervised_id, ExitReason::Abnormal)
        {
            Ok(backoff) => {
                self.last_restart_backoff = Some(backoff);
                if backoff == std::time::Duration::ZERO {
                    // Temporary：子项已被 Supervisor 移除，真实 worker 同步停止移除
                    self.remove_worker_at(worker_index);
                    return;
                }
                // Permanent/Transient：停止故障 worker 并登记非阻塞退避重启计划
                // （run_cycle 到点执行，禁止 sleep 阻塞数据面调用线程）
                self.stop_worker_at(worker_index);
                self.pending_restarts.push(PendingRestart {
                    worker_index,
                    supervised_id,
                    next_retry_at: Instant::now()
                        .checked_add(backoff)
                        .unwrap_or_else(Instant::now),
                });
            }
            Err(_) => {
                // 熔断（重启阈值超限）：Supervisor 已进入 Failed，不再重启
                self.last_restart_backoff = None;
            }
        }
    }

    // ==================== 内部方法 ====================

    /// 汇总全部数据面 worker 的累计统计（多 worker 聚合视图）
    ///
    /// 计数类字段 saturating 求和；p99 延迟取各 worker 最差值（保守语义）；
    /// CPU 占用率取均值；队列深度/吞吐量求和。
    fn aggregated_worker_stats(&self) -> WorkerStats {
        let mut agg = self.worker.stats();
        let mut count = 1u64;
        for w in &self.extra_workers {
            let s = w.stats();
            agg.rx_packets = agg.rx_packets.saturating_add(s.rx_packets);
            agg.tx_packets = agg.tx_packets.saturating_add(s.tx_packets);
            agg.rejected_packets = agg.rejected_packets.saturating_add(s.rejected_packets);
            agg.parse_errors = agg.parse_errors.saturating_add(s.parse_errors);
            agg.frame_allocs = agg.frame_allocs.saturating_add(s.frame_allocs);
            agg.frame_frees = agg.frame_frees.saturating_add(s.frame_frees);
            agg.quarantined_frames = agg.quarantined_frames.saturating_add(s.quarantined_frames);
            agg.cycles_completed = agg.cycles_completed.saturating_add(s.cycles_completed);
            agg.queue_depth = agg.queue_depth.saturating_add(s.queue_depth);
            agg.throughput = agg.throughput.saturating_add(s.throughput);
            agg.latency_p99 = agg.latency_p99.max(s.latency_p99);
            agg.cpu_usage += s.cpu_usage;
            count = count.saturating_add(1);
        }
        agg.cpu_usage /= count as f64;
        agg
    }

    /// 治理闭环：每周期向指标采集器上报处理包数 / 丢包 / 准入拒绝 / p99 延迟
    ///
    /// counter 以累计值 saturating 差分上报（多 worker 聚合口径）；
    /// p99 延迟为 gauge 瞬时值。未注入采集器时零开销直接返回。
    fn report_cycle_metrics(&mut self, wstats: &WorkerStats) {
        let Some(metrics) = &self.metrics else {
            return;
        };
        let Ok(mut m) = metrics.lock() else {
            return;
        };
        let processed = wstats.rx_packets.saturating_add(wstats.tx_packets);
        let _ = m.counter_add(
            "zenith_runtime_packets_processed_total",
            processed.saturating_sub(self.metrics_last_processed),
            &[],
        );
        let _ = m.counter_add(
            "zenith_runtime_admission_rejected_total",
            wstats.rejected_packets.saturating_sub(self.metrics_last_rejected),
            &[],
        );
        let _ = m.counter_add(
            "zenith_runtime_parse_errors_total",
            wstats.parse_errors.saturating_sub(self.metrics_last_dropped),
            &[],
        );
        let _ = m.gauge_set(
            "zenith_runtime_latency_p99_us",
            wstats.latency_p99.min(i64::MAX as u64) as i64,
            &[],
        );
        self.metrics_last_processed = processed;
        self.metrics_last_rejected = wstats.rejected_packets;
        self.metrics_last_dropped = wstats.parse_errors;
    }

    fn update_runtime_stats(&mut self, wstats: &WorkerStats) {
        self.stats.total_rx = wstats.rx_packets;
        self.stats.total_tx = wstats.tx_packets;
        self.stats.total_rejected = wstats.rejected_packets;
        self.stats.total_parse_errors = wstats.parse_errors;
        self.stats.total_packets = wstats.rx_packets.saturating_add(wstats.tx_packets);
    }
}

impl Drop for WorkerRuntime {
    fn drop(&mut self) {
        self.stop();
    }
}

impl std::fmt::Debug for WorkerRuntime {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("WorkerRuntime")
            .field("state", &self.state)
            .field("worker_id", &self.worker.id())
            .field("stats", &self.stats)
            .field("quarantined", &self.quarantined_frames)
            .finish()
    }
}

/// WorkerRuntime 错误
#[derive(Debug)]
pub enum RuntimeError {
    /// 运行时未运行（处于非 Running 状态）
    NotRunning,
    /// 运行时未初始化
    NotInitialized,
    /// Worker 数据面循环错误
    WorkerError(String),
    /// eBPF 子系统错误
    EbpfError(String),
    /// eBPF Map 操作错误
    MapError(String),
    /// AF_XDP Socket 错误
    XskError(String),
    /// 来源准入引擎错误
    AdmissionError(String),
    /// 帧池或帧管理错误
    FrameError(String),
    /// 双 Bank 热更新未启用
    DualBankNotEnabled,
    /// XSK 文件描述符未找到
    XskFdNotFound,
    /// 配置参数无效
    InvalidConfig(String),
}

impl std::fmt::Display for RuntimeError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            RuntimeError::NotRunning => write!(f, "runtime not running"),
            RuntimeError::NotInitialized => write!(f, "runtime not initialized"),
            RuntimeError::WorkerError(e) => write!(f, "worker error: {}", e),
            RuntimeError::EbpfError(e) => write!(f, "ebpf error: {}", e),
            RuntimeError::MapError(e) => write!(f, "map error: {}", e),
            RuntimeError::XskError(e) => write!(f, "xsk error: {}", e),
            RuntimeError::AdmissionError(e) => write!(f, "admission error: {}", e),
            RuntimeError::FrameError(e) => write!(f, "frame error: {}", e),
            RuntimeError::DualBankNotEnabled => write!(f, "dual bank not enabled"),
            RuntimeError::XskFdNotFound => write!(f, "XSK file descriptor not found"),
            RuntimeError::InvalidConfig(e) => write!(f, "invalid config: {}", e),
        }
    }
}

impl std::error::Error for RuntimeError {}

impl From<zenith_net::error::NetError> for RuntimeError {
    fn from(e: zenith_net::error::NetError) -> Self {
        RuntimeError::WorkerError(e.to_string())
    }
}

#[cfg(target_os = "linux")]
impl From<zenith_ebpf::error::LoadError> for RuntimeError {
    fn from(e: zenith_ebpf::error::LoadError) -> Self {
        RuntimeError::EbpfError(e.to_string())
    }
}

#[cfg(target_os = "linux")]
impl From<zenith_ebpf::error::AttachError> for RuntimeError {
    fn from(e: zenith_ebpf::error::AttachError) -> Self {
        RuntimeError::EbpfError(e.to_string())
    }
}

#[cfg(target_os = "linux")]
impl From<zenith_ebpf::error::ManagerError> for RuntimeError {
    fn from(e: zenith_ebpf::error::ManagerError) -> Self {
        RuntimeError::EbpfError(e.to_string())
    }
}

impl From<zenith_foundation::error::CoreError> for RuntimeError {
    fn from(e: zenith_foundation::error::CoreError) -> Self {
        RuntimeError::FrameError(e.to_string())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use zenith_net::source_admission::{AdmissionAction, IpAddr};

    fn create_test_config() -> WorkerRuntimeConfig {
        WorkerRuntimeConfig::default()
    }

    #[test]
    fn test_runtime_creation() {
        let config = create_test_config();
        let runtime = WorkerRuntime::new(config);
        assert!(runtime.is_ok());
        let runtime = runtime.unwrap();
        assert_eq!(runtime.state(), RuntimeState::Created);
        assert_eq!(runtime.worker_id(), 0);
    }

    #[test]
    fn test_runtime_start_stop() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();
        assert_eq!(runtime.state(), RuntimeState::Created);

        runtime.start().unwrap();
        assert_eq!(runtime.state(), RuntimeState::Running);

        runtime.stop();
        assert_eq!(runtime.state(), RuntimeState::Stopped);
    }

    #[test]
    fn test_runtime_not_running() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();

        let result = runtime.run_cycle();
        assert!(result.is_err());
    }

    #[test]
    fn test_runtime_full_lifecycle() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();

        let result = runtime.run_full_lifecycle(5);
        assert!(result.is_ok());

        let packets = result.unwrap();
        assert_eq!(packets, 0);
        assert_eq!(runtime.state(), RuntimeState::Stopped);
    }

    // simulate_rx_transfer 为真实 AF_XDP Worker 的测试辅助，仅 Linux 可用
    #[cfg(target_os = "linux")]
    #[test]
    fn test_runtime_full_lifecycle_with_data() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();
        runtime.start().unwrap();

        runtime.run_cycle().unwrap();

        runtime.worker.simulate_rx_transfer(2, 4096, |_idx, data| {
            data.fill(0xAB);
        });

        let processed = runtime.run_cycle().unwrap();
        assert_eq!(processed, 2);

        let stats = runtime.stats();
        assert_eq!(stats.total_rx, 2);

        runtime.stop();
    }

    #[test]
    fn test_runtime_with_admission_rules() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();

        let rule = AdmissionRule {
            id: 1,
            src_ip: IpAddr::V4_WILDCARD,
            prefix_len: 0,
            src_port: 0,
            dst_port: 8080,
            proto: zenith_net::source_admission::ProtoMatch::Tcp,
            action: AdmissionAction::Allow,
            enabled: true,
        };

        runtime.add_admission_rule(rule).unwrap();
        let admission = runtime.admission();
        assert!(admission.rule_count() >= 1);
    }

    #[test]
    fn test_runtime_quarantine() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();

        assert_eq!(runtime.quarantined_count(), 0);

        // 启动后帧池有帧可查询
        runtime.start().unwrap();

        // 先分配一个帧（改变状态到 Allocated），才能隔离
        let token = runtime.frame_pool_mut().allocate(0).unwrap();
        let frame_id = token.frame_id();
        runtime.quarantine_frame(frame_id, "test").unwrap();

        // 隔离成功，计数为 1
        assert_eq!(runtime.quarantined_count(), 1);

        // 恢复所有隔离帧
        let recovered = runtime.recover_all_quarantined().unwrap();
        assert_eq!(recovered, 1);
        assert_eq!(runtime.quarantined_count(), 0);

        runtime.stop();
    }

    #[test]
    fn test_runtime_debug() {
        let config = create_test_config();
        let runtime = WorkerRuntime::new(config).unwrap();
        let debug_str = format!("{:?}", runtime);
        assert!(debug_str.contains("WorkerRuntime"));
        assert!(debug_str.contains("Created"));
    }

    #[test]
    fn test_runtime_stats() {
        let config = create_test_config();
        let mut runtime = WorkerRuntime::new(config).unwrap();

        runtime.start().unwrap();
        runtime.run_cycle().unwrap();

        let stats = runtime.stats();
        assert!(stats.cycles_completed >= 1);
        // total_packets 为 u64，永远 >= 0；这里验证零数据包场景下的统计正确性
        assert_eq!(stats.total_packets, 0);
    }

    #[test]
    fn test_verify_conservation() {
        let config = create_test_config();
        let runtime = WorkerRuntime::new(config).unwrap();
        assert!(runtime.verify_conservation());
    }

    #[test]
    fn test_capability_detection_integrated() {
        // WorkerRuntime::new 必须接入 zenith-capability 检测并选择数据面 Profile
        let runtime = WorkerRuntime::new(create_test_config()).unwrap();
        assert!(runtime.capability_snapshot().cpu_cores >= 1);
        // Profile 必为三档之一（按能力快照选择）
        match runtime.data_plane_profile() {
            zenith_capability::profile::Profile::Performance
            | zenith_capability::profile::Profile::Balanced
            | zenith_capability::profile::Profile::Minimal => {}
        }
    }

    #[test]
    fn test_ledger_real_quota_and_rollup() {
        // 真实配额：帧配额 = 帧池容量；run_cycle 后账本 used 与帧池占用一致
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        assert_eq!(
            runtime.ledger().total().frame_count,
            u64::from(runtime.config().frame_pool_capacity)
        );
        assert_eq!(
            runtime.ledger().total().memory_bytes,
            u64::from(runtime.config().frame_pool_capacity) * FRAME_BYTES
        );

        runtime.start().unwrap();
        runtime.run_cycle().unwrap();
        let pool_allocated = u64::from(runtime.frame_pool().allocated_count());
        assert_eq!(runtime.ledger().used().frame_count, pool_allocated);
        // 子账本 roll-up：queue 子账本与域账本记账一致
        let child = runtime
            .ledger()
            .get_child("queue-0")
            .unwrap();
        assert_eq!(child.used().frame_count, pool_allocated);
        runtime.stop();
    }

    #[test]
    fn test_supervisor_three_level_integrated() {
        // 三级 Supervisor 已挂载：Node → Domain → Queue → Worker 全部健康
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        let report = runtime.supervisor_health();
        assert!(report.total >= 1);
        assert!(report.is_healthy());
        assert_eq!(runtime.supervised_worker_count(), 1);
    }

    #[test]
    fn test_worker_fault_restart_via_supervisor() {
        // Worker 故障经三级层级上报并按 Permanent 策略登记非阻塞退避重启
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.start().unwrap();
        runtime.handle_worker_fault(0);
        // Permanent：首次重启退避上界 = initial 10ms（全抖动后 ∈ [0, 10ms]）
        let backoff = runtime
            .last_restart_backoff()
            .expect("重启后必须记录退避时长");
        assert!(
            backoff <= std::time::Duration::from_millis(10),
            "首次退避不得超过 initial 10ms，实际 {backoff:?}"
        );
        // 非阻塞退避：故障后登记待重启计划（不 sleep 阻塞），worker 先停止
        assert_eq!(runtime.pending_restarts.len(), 1);
        assert_eq!(runtime.worker_state(), zenith_net::worker::WorkerState::Stopped);
        // 退避到点后 run_cycle 真实执行重启并恢复健康（上界 10ms，留足余量）
        std::thread::sleep(std::time::Duration::from_millis(15));
        runtime.run_cycle().unwrap();
        assert!(runtime.pending_restarts.is_empty());
        assert_eq!(runtime.worker_state(), zenith_net::worker::WorkerState::Running);
        assert!(runtime.supervisor_health().is_healthy());
        runtime.stop();
    }

    #[test]
    fn test_fault_backoff_does_not_block_run_cycle() {
        // 非阻塞退避回归测试：handle_worker_fault 与退避期内的 run_cycle
        // 都不得 sleep 阻塞（原实现在数据面调用线程 std::thread::sleep(backoff)）
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.start().unwrap();

        let t0 = Instant::now();
        runtime.handle_worker_fault(0);
        let fault_elapsed = t0.elapsed();
        assert!(
            fault_elapsed < std::time::Duration::from_millis(5),
            "handle_worker_fault 疑似阻塞退避: {fault_elapsed:?}"
        );

        // 退避期内（backoff 可能 > 0）立即 run_cycle：跳过未到期计划，不阻塞
        let t1 = Instant::now();
        runtime.run_cycle().unwrap();
        let cycle_elapsed = t1.elapsed();
        assert!(
            cycle_elapsed < std::time::Duration::from_millis(50),
            "run_cycle 疑似被退避 sleep 阻塞: {cycle_elapsed:?}"
        );
        runtime.stop();
    }

    #[test]
    fn test_qos_level_applied_to_worker() {
        // QoS 等级真实生效：Critical → 准入拒绝帧也隔离（最严格准入）
        // （直接调用 apply_qos_to_worker，避免与控制线程的自动评估产生竞态）
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        assert!(!runtime.worker.quarantine_on_deny());
        runtime.set_qos_level(QosLevel::Critical);
        runtime.apply_qos_to_worker();
        assert!(runtime.worker.quarantine_on_deny());
        assert!(runtime.worker.quarantine_on_error());

        // 回落到 Background → 不隔离（最大吞吐）
        runtime.set_qos_level(QosLevel::Background);
        runtime.apply_qos_to_worker();
        assert!(!runtime.worker.quarantine_on_deny());
        assert!(!runtime.worker.quarantine_on_error());
    }

    #[test]
    fn test_replan_topology_applies_result() {
        // 规划结果真实应用：亲和来自规划节点（非 config.queue_id 直取），
        // 且 PlannedTopology 被真实保存（非 format! 字符串）
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        let mut graph: RuntimeGraph<64, 256> = RuntimeGraph::new();
        graph
            .add_node(ResourceNode::new(0, ExecutionDomain::DataPlane, 8, 16384, 8))
            .unwrap();
        let snapshots = vec![DomainSnapshot::from_domain(ExecutionDomain::DataPlane, 1)];
        runtime.replan_topology(&graph, &snapshots).unwrap();

        let topology = runtime.last_planned_topology().unwrap();
        assert_eq!(topology.domain_count(), 1);
        assert_eq!(topology.total_workers(), 1);
        // 亲和配置已应用（Linux 上 sched_setaffinity 成功）
        #[cfg(target_os = "linux")]
        assert!(runtime.replan_apply_error().is_none());
    }

    #[test]
    fn test_pending_replan_consumed_by_run_cycle() {
        // 控制线程置位的 replan_pending 必须被 run_cycle 真正消费
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.start().unwrap();
        runtime.replan_pending.store(true, Ordering::SeqCst);
        runtime.run_cycle().unwrap();
        assert!(!runtime.replan_pending.load(Ordering::SeqCst));
        // 规划已执行：结果保存或错误记录（取决于检测到的本机资源）
        assert!(
            runtime.last_planned_topology().is_some() || runtime.replan_apply_error().is_some()
        );
        runtime.stop();
    }

    #[cfg(target_os = "linux")]
    #[test]
    fn test_switch_ebpf_bank_requires_dual_bank() {
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        let err = runtime.switch_ebpf_bank().unwrap_err();
        assert!(matches!(err, RuntimeError::DualBankNotEnabled));
    }

    #[test]
    fn test_cache_warmup_hook_invoked() {
        // 缓存预热真实消费路径：热键经控制线程推送给注册回调
        use std::sync::atomic::AtomicU64 as AU64;
        let hits = Arc::new(AU64::new(0));
        let hits_clone = hits.clone();

        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.set_cache_warmup_hook(Box::new(move |keys| {
            if !keys.is_empty() {
                hits_clone.fetch_add(1, Ordering::SeqCst);
            }
        }));

        runtime.start().unwrap();
        // 记录足够访问使键变热（AutoOptimizer 热阈值 10，衰减不动点约 20）
        for _ in 0..30 {
            runtime.record_cache_access("hot-session-key");
        }
        std::thread::sleep(std::time::Duration::from_millis(350));
        runtime.stop();

        assert!(
            hits.load(Ordering::SeqCst) > 0,
            "缓存预热回调未被控制线程调用"
        );
    }

    #[test]
    fn test_target_worker_count_reconciled() {
        // 目标 worker 数落实：手动设置目标后 run_cycle 真实扩缩容 + Supervisor 记账同步
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.start().unwrap();
        runtime.target_worker_count.store(3, Ordering::Relaxed);
        runtime.run_cycle().unwrap();
        assert_eq!(runtime.supervised_worker_count(), 3);
        assert_eq!(runtime.worker_count(), 3);

        runtime.target_worker_count.store(1, Ordering::Relaxed);
        runtime.run_cycle().unwrap();
        assert_eq!(runtime.supervised_worker_count(), 1);
        assert_eq!(runtime.worker_count(), 1);
        runtime.stop();
    }

    #[test]
    fn test_multi_worker_spawn_and_drive() {
        // 真实多 worker：扩容构造真实数据面 Worker（非仅 Supervisor 记账），
        // run_cycle 轮询驱动全部 worker 并汇总周期统计
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.start().unwrap();
        assert_eq!(runtime.worker_count(), 1);

        // 扩容到 3：真实构造 2 个额外 Worker 并启动
        runtime.target_worker_count.store(3, Ordering::Relaxed);
        runtime.run_cycle().unwrap();
        assert_eq!(runtime.worker_count(), 3);
        assert_eq!(runtime.supervised_worker_count(), 3);
        assert!(runtime.verify_conservation());

        // 第二周期：3 个 worker 全部被真实驱动（周期计数汇总 = 1 + 3）
        runtime.run_cycle().unwrap();
        let agg = runtime.aggregated_worker_stats();
        assert!(
            agg.cycles_completed >= 4,
            "多 worker 周期未被全部驱动: {}",
            agg.cycles_completed
        );

        // 缩容到 1：真实停止并移除额外 worker
        runtime.target_worker_count.store(1, Ordering::Relaxed);
        runtime.run_cycle().unwrap();
        assert_eq!(runtime.worker_count(), 1);
        assert_eq!(runtime.supervised_worker_count(), 1);
        runtime.stop();
    }

    // simulate_rx_transfer 为真实 AF_XDP Worker 的测试辅助，仅 Linux 可用
    #[cfg(target_os = "linux")]
    #[test]
    fn test_multi_worker_data_aggregation() {
        // 多 worker 数据面聚合：主 worker 与扩容 worker 的处理包数汇总上报
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        runtime.start().unwrap();
        runtime.target_worker_count.store(2, Ordering::Relaxed);
        runtime.run_cycle().unwrap();
        assert_eq!(runtime.worker_count(), 2);

        // 扩容 worker 冷启动：reconcile（构造扩容 worker）发生在周期驱动循环之后，
        // 扩容当周期 extra worker 尚未执行 process_cycle，其 fill ring 为空——
        // simulate 注入依赖 fill ring 预填帧，需先跑一个周期完成 Phase1 冷填
        runtime.run_cycle().unwrap();

        runtime.worker.simulate_rx_transfer(2, 4096, |_idx, data| {
            data.fill(0xAB);
        });
        runtime.extra_workers[0].simulate_rx_transfer(3, 4096, |_idx, data| {
            data.fill(0xCD);
        });

        let processed = runtime.run_cycle().unwrap();
        assert_eq!(processed, 5, "多 worker 处理包数必须汇总");
        let stats = runtime.stats();
        assert_eq!(stats.total_rx, 5);
        runtime.stop();
    }

    #[test]
    fn test_metrics_reporting_accumulates() {
        // 治理闭环：注入 MetricsCollector 后 run_cycle 每周期真实上报四项核心指标
        let metrics = Arc::new(Mutex::new(MetricsCollector::new()));
        let mut runtime = WorkerRuntime::new(create_test_config())
            .unwrap()
            .with_metrics(metrics.clone());
        runtime.start().unwrap();
        for _ in 0..3 {
            runtime.run_cycle().unwrap();
        }
        runtime.stop();

        let m = metrics.lock().unwrap();
        let snapshot = m.snapshot();
        for name in [
            "zenith_runtime_packets_processed_total",
            "zenith_runtime_admission_rejected_total",
            "zenith_runtime_parse_errors_total",
            "zenith_runtime_latency_p99_us",
        ] {
            assert!(
                snapshot.iter().any(|s| s.name == name),
                "指标 {name} 未上报，快照: {snapshot:?}"
            );
        }
    }

    // simulate_rx_transfer 为真实 AF_XDP Worker 的测试辅助，仅 Linux 可用
    #[cfg(target_os = "linux")]
    #[test]
    fn test_metrics_counter_values_accumulate() {
        // 指标真实累计：模拟 2 包后 processed counter 精确累计为 2
        let metrics = Arc::new(Mutex::new(MetricsCollector::new()));
        let mut runtime = WorkerRuntime::new(create_test_config())
            .unwrap()
            .with_metrics(metrics.clone());
        runtime.start().unwrap();
        runtime.run_cycle().unwrap();
        runtime.worker.simulate_rx_transfer(2, 4096, |_idx, data| {
            data.fill(0xAB);
        });
        runtime.run_cycle().unwrap();
        runtime.stop();

        let m = metrics.lock().unwrap();
        let snapshot = m.snapshot();
        let processed = snapshot
            .iter()
            .find(|s| s.name == "zenith_runtime_packets_processed_total")
            .expect("processed 指标必须存在");
        assert_eq!(processed.value, 2, "processed counter 必须真实累计");
    }

    #[test]
    fn test_control_thread_reports_qos_events() {
        // 治理闭环：控制线程上报 QoS 等级变更（首次评估即产生基线变更事件）
        let metrics = Arc::new(Mutex::new(MetricsCollector::new()));
        let mut runtime = WorkerRuntime::new(create_test_config())
            .unwrap()
            .with_metrics(metrics.clone());
        runtime.start().unwrap();
        // 控制线程每 100ms 评估一轮，留足两轮以上
        std::thread::sleep(std::time::Duration::from_millis(350));
        runtime.stop();

        let m = metrics.lock().unwrap();
        let snapshot = m.snapshot();
        assert!(
            snapshot
                .iter()
                .any(|s| s.name == "zenith_runtime_qos_level_changes_total" && s.value >= 1),
            "QoS 等级变更事件未上报，快照: {snapshot:?}"
        );
        assert!(
            snapshot.iter().any(|s| s.name == "zenith_runtime_qos_level"),
            "QoS 等级 gauge 未上报，快照: {snapshot:?}"
        );
    }

    #[test]
    fn test_fault_recorded_to_observability() {
        // 治理闭环：故障处理真实记录 FaultRecorder 快照并累计故障计数器
        let metrics = Arc::new(Mutex::new(MetricsCollector::new()));
        let recorder = Arc::new(Mutex::new(FaultRecorder::new()));
        let mut runtime = WorkerRuntime::new(create_test_config())
            .unwrap()
            .with_metrics(metrics.clone())
            .with_fault_recorder(recorder.clone());
        runtime.start().unwrap();
        runtime.handle_worker_fault(0);

        assert_eq!(
            recorder.lock().unwrap().len(),
            1,
            "故障事件未记录到 FaultRecorder"
        );
        let m = metrics.lock().unwrap();
        let snapshot = m.snapshot();
        assert!(
            snapshot
                .iter()
                .any(|s| s.name == "zenith_runtime_worker_faults_total" && s.value >= 1),
            "故障计数器未累计，快照: {snapshot:?}"
        );
        runtime.stop();
    }

    #[test]
    fn test_spawned_worker_replays_admission_rules() {
        // 扩容的新 worker 必须重放已配置准入规则（多 worker 准入一致）
        let mut runtime = WorkerRuntime::new(create_test_config()).unwrap();
        let rule = AdmissionRule {
            id: 7,
            src_ip: IpAddr::V4_WILDCARD,
            prefix_len: 0,
            src_port: 0,
            dst_port: 443,
            proto: zenith_net::source_admission::ProtoMatch::Tcp,
            action: AdmissionAction::Allow,
            enabled: true,
        };
        runtime.add_admission_rule(rule).unwrap();

        runtime.start().unwrap();
        runtime.target_worker_count.store(2, Ordering::Relaxed);
        runtime.run_cycle().unwrap();
        assert_eq!(runtime.worker_count(), 2);
        assert_eq!(
            runtime.extra_workers[0].admission().rule_count(),
            runtime.admission().rule_count(),
            "扩容 worker 未重放准入规则"
        );
        runtime.stop();
    }
}