rekuiper-server 1.0.0-rc4

//! Process-based reliability qualification and soak harness for rekuiper.
//!
//! Features:
//! 1. Launches real `kuiperd` child process with isolated config, storage, and ephemeral ports.
//! 2. Captures engine stdout/stderr to persistent disk logs for full post-mortem diagnostics.
//! 3. Monitors actual `kuiperd` process RSS memory, not the test harness.
//! 4. Generates paced traffic for the requested duration and rate; measures achieved throughput.
//! 5. Independently subscribes to the MQTT sink and validates sequence IDs, duplicates, and loss.
//! 6. Uses bounded $O(1)$ gap tracking so 24h-72h soak runs never exhaust memory.
//! 7. Binds and measures in-flight records during abrupt process kill and restart.
//! 8. Measures true broker recovery time until actual message delivery resumes.
//! 9. Executes real upgrade and rollback verification across distinct binaries.
//! 10. Strictly enforces comprehensive acceptance criteria across all failure modes.

use async_trait::async_trait;
use rekuiper_core::{KvStore, SqliteKvStore, StreamDefinition, StreamManager};
use rumqttc::{AsyncClient, MqttOptions, Packet, QoS};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::collections::{BTreeSet, HashMap};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use sysinfo::Pid;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;
use tokio::process::{Child, Command};
use tokio::sync::Mutex;

/// Qualification metrics produced by the harness.
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct QualificationMetrics {
    pub mode: String,
    pub source_revision: String,
    pub dirty_tree: bool,
    pub binary_path: String,
    pub engine_log_path: String,
    pub broker_type: String,
    pub host_os: String,
    pub host_cpus: usize,
    pub host_memory_mb: f64,
    pub requested_duration_seconds: f64,
    pub measured_duration_seconds: f64,
    pub requested_rate: u64,
    pub achieved_rate: f64,
    pub accepted_input: u64,
    pub delivered_output: u64,
    pub unique_delivered: u64,
    pub missing_records: u64,
    pub missing_sample_ids: Vec<u64>,
    pub duplicate_records: u64,
    pub in_flight_at_shutdown: u64,
    pub in_flight_ids_at_shutdown: Vec<u64>,
    pub unexplained_missing_records: Vec<u64>,
    pub out_of_order_count: u64,
    pub initial_rss_mb: f64,
    pub peak_rss_mb: f64,
    pub final_rss_mb: f64,
    pub memory_growth_mb: f64,
    pub error_count: u64,
    pub broker_recovery_time_ms: u64,
    pub engine_recovery_time_ms: u64,
    pub storage_fault_errors_caught: u64,
    pub upgrade_rollback_status: String,
    pub upgrade_rollback_verified: bool,
    pub status: String,
    pub failure_reasons: Vec<String>,
    pub timestamp_utc: String,
}

type Tasks = Arc<parking_lot::Mutex<Vec<tokio::task::JoinHandle<()>>>>;

/// Bidirectional TCP proxy to a real Mosquitto broker that supports simulating network/broker outages.
struct FaultInjectingProxy {
    port: u16,
    upstream: String,
    tasks: Tasks,
}

impl FaultInjectingProxy {
    async fn start(upstream: String) -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let proxy = Self {
            port,
            upstream,
            tasks: Tasks::default(),
        };
        proxy.serve(listener);
        proxy
    }

    fn serve(&self, listener: TcpListener) {
        let upstream = self.upstream.clone();
        let tasks = self.tasks.clone();
        let tasks_for_accept = tasks.clone();
        let accept = tokio::spawn(async move {
            while let Ok((mut inbound, _)) = listener.accept().await {
                let upstream = upstream.clone();
                let conn = tokio::spawn(async move {
                    if let Ok(mut outbound) = tokio::net::TcpStream::connect(&upstream).await {
                        let _ = tokio::io::copy_bidirectional(&mut inbound, &mut outbound).await;
                    }
                });
                tasks_for_accept.lock().push(conn);
            }
        });
        tasks.lock().push(accept);
    }

    fn simulate_outage(&self) {
        for task in self.tasks.lock().drain(..) {
            task.abort();
        }
    }

    fn restore(&self) {
        let socket = tokio::net::TcpSocket::new_v4().unwrap();
        let _ = socket.set_reuseaddr(true);
        socket
            .bind(format!("127.0.0.1:{}", self.port).parse().unwrap())
            .unwrap();
        self.serve(socket.listen(64).unwrap());
    }
}

impl Drop for FaultInjectingProxy {
    fn drop(&mut self) {
        self.simulate_outage();
    }
}

/// Fallback self-contained mock MQTT broker supporting controlled outages, QoS 1 pubacks,
/// and fan-out of PUBLISH packets to connected independent subscribers when no real broker is present.
struct FaultInjectingBroker {
    port: u16,
    is_down: Arc<AtomicBool>,
    _publishes_received: Arc<AtomicU64>,
    shutdown: Option<tokio::sync::oneshot::Sender<()>>,
}

impl FaultInjectingBroker {
    async fn start() -> Self {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let is_down = Arc::new(AtomicBool::new(false));
        let publishes_received = Arc::new(AtomicU64::new(0));
        let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel::<()>();

        // Connected subscribers for message fan-out
        let subscribers = Arc::new(Mutex::new(Vec::<tokio::net::tcp::OwnedWriteHalf>::new()));

        let is_down_clone = is_down.clone();
        let pub_clone = publishes_received.clone();
        let subs_clone = subscribers.clone();

        tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = &mut shutdown_rx => break,
                    accept_res = listener.accept() => {
                        let (stream, _) = match accept_res {
                            Ok(s) => s,
                            Err(_) => break,
                        };

                        if is_down_clone.load(Ordering::Relaxed) {
                            drop(stream);
                            continue;
                        }

                        let is_down_conn = is_down_clone.clone();
                        let pub_conn = pub_clone.clone();
                        let subs_conn = subs_clone.clone();

                        tokio::spawn(async move {
                            let (mut reader, mut writer) = stream.into_split();
                            let mut buf = [0u8; 16384];

                            // Read CONNECT packet
                            let _n = match reader.read(&mut buf).await {
                                Ok(n) if n > 0 => n,
                                _ => return,
                            };

                            if buf[0] == 0x10 {
                                // Send CONNACK: 0x20, 0x02, 0x00, 0x00
                                if writer.write_all(&[0x20, 0x02, 0x00, 0x00]).await.is_err() {
                                    return;
                                }
                            }

                            loop {
                                if is_down_conn.load(Ordering::Relaxed) {
                                    break;
                                }

                                match tokio::time::timeout(Duration::from_millis(50), reader.read(&mut buf)).await {
                                    Ok(Ok(0)) => break,
                                    Ok(Ok(n)) => {
                                        let mut cursor = 0;
                                        while cursor < n {
                                            let packet_type = buf[cursor] & 0xF0;

                                            if packet_type == 0xC0 {
                                                // PINGREQ -> PINGRESP
                                                let _ = writer.write_all(&[0xD0, 0x00]).await;
                                                cursor += 2;
                                            } else if packet_type == 0x80 {
                                                // SUBSCRIBE packet from independent test client
                                                let pkid_msb = if cursor + 2 < n { buf[cursor + 2] } else { 0 };
                                                let pkid_lsb = if cursor + 3 < n { buf[cursor + 3] } else { 1 };
                                                let _ = writer.write_all(&[0x90, 0x03, pkid_msb, pkid_lsb, 0x01]).await;

                                                // Register this writer as a subscriber
                                                subs_conn.lock().await.push(writer);
                                                return; // Writer moved to subscribers pool
                                            } else if packet_type == 0x40 {
                                                // PUBACK from client, ignore
                                                cursor += 4;
                                            } else if packet_type == 0x30 {
                                                // PUBLISH packet
                                                pub_conn.fetch_add(1, Ordering::Relaxed);
                                                let qos = (buf[cursor] & 0x06) >> 1;

                                                // If QoS 1, reply with PUBACK
                                                if qos == 1 {
                                                    let topic_len_offset = cursor + 2;
                                                    if topic_len_offset + 2 <= n {
                                                        let topic_len = ((buf[topic_len_offset] as usize) << 8)
                                                            | (buf[topic_len_offset + 1] as usize);
                                                        let pkid_offset = topic_len_offset + 2 + topic_len;
                                                        if pkid_offset + 2 <= n {
                                                            let pkid_msb = buf[pkid_offset];
                                                            let pkid_lsb = buf[pkid_offset + 1];
                                                            let _ = writer.write_all(&[0x40, 0x02, pkid_msb, pkid_lsb]).await;
                                                        }
                                                    }
                                                }

                                                // Fan-out publish packet to subscribers
                                                let mut subs = subs_conn.lock().await;
                                                let packet_slice = &buf[cursor..n];
                                                let mut i = 0;
                                                while i < subs.len() {
                                                    if subs[i].write_all(packet_slice).await.is_err() {
                                                        subs.swap_remove(i);
                                                    } else {
                                                        i += 1;
                                                    }
                                                }
                                                break;
                                            } else {
                                                break;
                                            }
                                        }
                                    }
                                    Ok(Err(_)) => break,
                                    Err(_) => {}
                                }
                            }
                        });
                    }
                }
            }
        });

        Self {
            port,
            is_down,
            _publishes_received: publishes_received,
            shutdown: Some(shutdown_tx),
        }
    }

    fn simulate_outage(&self) {
        self.is_down.store(true, Ordering::Relaxed);
    }

    fn restore(&self) {
        self.is_down.store(false, Ordering::Relaxed);
    }
}

impl Drop for FaultInjectingBroker {
    fn drop(&mut self) {
        if let Some(tx) = self.shutdown.take() {
            let _ = tx.send(());
        }
    }
}

async fn probe_tcp(addr: &str) -> bool {
    tokio::time::timeout(
        Duration::from_millis(500),
        tokio::net::TcpStream::connect(addr),
    )
    .await
    .map(|res| res.is_ok())
    .unwrap_or(false)
}

/// Isolated temporary test directory.
struct TestEnvDir {
    path: PathBuf,
}

impl TestEnvDir {
    fn new(prefix: &str) -> Self {
        let path = std::env::temp_dir().join(format!(
            "rekuiper_qual_{}_{}_{}",
            prefix,
            std::process::id(),
            chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0)
        ));
        let _ = std::fs::remove_dir_all(&path);
        std::fs::create_dir_all(&path).unwrap();
        std::fs::create_dir_all(path.join("etc")).unwrap();
        std::fs::create_dir_all(path.join("data")).unwrap();
        std::fs::create_dir_all(path.join("log")).unwrap();
        Self { path }
    }
}

impl Drop for TestEnvDir {
    fn drop(&mut self) {
        let _ = std::fs::remove_dir_all(&self.path);
    }
}

fn get_qualification_log_path() -> PathBuf {
    let log_dir = PathBuf::from("target/qualification_logs");
    let _ = std::fs::create_dir_all(&log_dir);
    log_dir.join("kuiperd_child.log")
}

/// Measures resident memory (RSS in MB) of a specific process PID.
fn get_process_rss_mb(pid: u32) -> f64 {
    let mut sys = sysinfo::System::new();
    sys.refresh_processes();
    let p = Pid::from_u32(pid);
    if let Some(proc) = sys.process(p) {
        proc.memory() as f64 / (1024.0 * 1024.0)
    } else {
        0.0
    }
}

fn find_kuiperd_bin() -> PathBuf {
    let manifest_dir = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    let workspace_root = manifest_dir
        .parent()
        .and_then(|p| p.parent())
        .map(PathBuf::from)
        .unwrap_or_else(|| PathBuf::from("."));

    let check_all = || -> Option<PathBuf> {
        if let Ok(current_exe) = std::env::current_exe() {
            if let Some(target_dir) = current_exe.parent().and_then(|p| p.parent()) {
                for sub in ["kuiperd", "kuiperd.exe", "rekuiperd", "rekuiperd.exe"] {
                    let candidate = target_dir.join(sub);
                    if candidate.exists() {
                        return Some(candidate.canonicalize().unwrap_or(candidate));
                    }
                }
            }
        }
        for env_var in [
            "REKUIPER_BIN",
            "CARGO_BIN_EXE_rekuiperd",
            "KUIPERD_BIN",
            "CARGO_BIN_EXE_kuiperd",
        ] {
            if let Ok(p) = std::env::var(env_var) {
                let pb = PathBuf::from(p);
                if pb.exists() {
                    return Some(pb.canonicalize().unwrap_or(pb));
                }
            }
        }
        if let Ok(td) = std::env::var("CARGO_TARGET_DIR") {
            let p = PathBuf::from(td);
            for sub in [
                "debug/kuiperd",
                "debug/kuiperd.exe",
                "release/kuiperd",
                "release/kuiperd.exe",
            ] {
                let candidate = p.join(sub);
                if candidate.exists() {
                    return Some(candidate.canonicalize().unwrap_or(candidate));
                }
            }
        }
        let candidates = [
            workspace_root.join("target/debug/rekuiperd.exe"),
            workspace_root.join("target/debug/rekuiperd"),
            workspace_root.join("target/release/rekuiperd.exe"),
            workspace_root.join("target/release/rekuiperd"),
            workspace_root.join("target/debug/kuiperd.exe"),
            workspace_root.join("target/debug/kuiperd"),
            workspace_root.join("target/release/kuiperd.exe"),
            workspace_root.join("target/release/kuiperd"),
            PathBuf::from("target/debug/kuiperd.exe"),
            PathBuf::from("target/debug/kuiperd"),
            PathBuf::from("../../target/debug/kuiperd.exe"),
            PathBuf::from("../../target/debug/kuiperd"),
            PathBuf::from("../target/debug/kuiperd.exe"),
            PathBuf::from("../target/debug/kuiperd"),
            PathBuf::from("target/release/kuiperd.exe"),
            PathBuf::from("target/release/kuiperd"),
            PathBuf::from("../../target/release/kuiperd.exe"),
            PathBuf::from("../../target/release/kuiperd"),
        ];
        for c in &candidates {
            if c.exists() {
                return Some(c.canonicalize().unwrap_or_else(|_| c.clone()));
            }
        }
        None
    };

    if let Some(found) = check_all() {
        return found;
    }

    let mut cmd = std::process::Command::new("cargo");
    cmd.args(["build", "-p", "kuiperd"])
        .current_dir(&workspace_root);
    if let Ok(td) = std::env::var("CARGO_TARGET_DIR") {
        cmd.env("CARGO_TARGET_DIR", td);
    }
    let _ = cmd.status();

    if let Some(found) = check_all() {
        return found;
    }

    panic!("kuiperd binary not found. Build it with `cargo build -p kuiperd` first.");
}

async fn get_free_port() -> u16 {
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    listener.local_addr().unwrap().port()
}

/// Spawns kuiperd child process with isolated config and preserves output in persistent log file.
async fn start_kuiperd_child(
    bin: &Path,
    dir: &Path,
    rest_port: u16,
    rpc_port: u16,
    sse_port: u16,
) -> Child {
    let kuiper_yaml = format!(
        "basic:\n  ip: \"127.0.0.1\"\n  port: {}\n  restIp: \"127.0.0.1\"\n  restPort: {}\n  httpServerIp: \"127.0.0.1\"\n  httpServerPort: {}\n  prometheus: false\n  consoleLog: true\n",
        rpc_port, rest_port, sse_port
    );
    std::fs::write(dir.join("etc/kuiper.yaml"), kuiper_yaml).unwrap();

    let log_path = get_qualification_log_path();
    let log_file = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&log_path)
        .expect("Failed to open child log file");
    let log_err = log_file
        .try_clone()
        .expect("Failed to clone child log handle");

    Command::new(bin)
        .current_dir(dir)
        .env("RUST_LOG", "info")
        .arg("--etc")
        .arg("etc")
        .arg("--data")
        .arg("data")
        .arg("--log")
        .arg("log")
        .stdout(log_file)
        .stderr(log_err)
        .spawn()
        .expect("Failed to spawn kuiperd child process")
}

async fn wait_for_healthy(base_url: &str, timeout: Duration) -> bool {
    let client = reqwest::Client::new();
    let start = Instant::now();
    while start.elapsed() < timeout {
        if let Ok(resp) = client.get(format!("{}/ping", base_url)).send().await {
            if resp.status().is_success() {
                return true;
            }
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    false
}

fn gather_git_info() -> (String, bool) {
    let rev = std::process::Command::new("git")
        .args(["rev-parse", "HEAD"])
        .output()
        .ok()
        .and_then(|o| String::from_utf8(o.stdout).ok())
        .map(|s| s.trim().to_string())
        .unwrap_or_else(|| "unknown".to_string());

    let dirty = std::process::Command::new("git")
        .args(["status", "--porcelain"])
        .output()
        .ok()
        .map(|o| !o.stdout.is_empty())
        .unwrap_or(false);

    (rev, dirty)
}

/// Executes real upgrade and rollback verification across two distinct binaries.
async fn verify_upgrade_rollback(old_bin: &Path, new_bin: &Path) -> Result<(), String> {
    if !old_bin.exists() {
        return Err(format!("Old binary does not exist at {:?}", old_bin));
    }
    if !new_bin.exists() {
        return Err(format!("New binary does not exist at {:?}", new_bin));
    }
    let old_canon = old_bin.canonicalize().map_err(|e| e.to_string())?;
    let new_canon = new_bin.canonicalize().map_err(|e| e.to_string())?;
    if old_canon == new_canon {
        return Err(
            "Old binary and new binary must be distinct file paths for upgrade qualification"
                .to_string(),
        );
    }
    let old_bytes = std::fs::read(&old_canon).map_err(|e| e.to_string())?;
    let new_bytes = std::fs::read(&new_canon).map_err(|e| e.to_string())?;
    if old_bytes == new_bytes {
        return Err("Old binary and new binary have identical binary content; distinct builds required for upgrade qualification".to_string());
    }

    let test_dir = TestEnvDir::new("upgrade_test");
    let rest_port = get_free_port().await;
    let rpc_port = get_free_port().await;
    let sse_port = get_free_port().await;
    let base_url = format!("http://127.0.0.1:{}", rest_port);
    let client = reqwest::Client::new();

    // 1. Start old binary
    let mut old_child =
        start_kuiperd_child(old_bin, &test_dir.path, rest_port, rpc_port, sse_port).await;
    if !wait_for_healthy(&base_url, Duration::from_secs(10)).await {
        let _ = old_child.kill().await;
        return Err(format!("Old binary at {:?} failed to start", old_bin));
    }

    // 2. Create stream and rule in old binary
    let s_res = client
        .post(format!("{}/streams", base_url))
        .json(&json!({"sql": "CREATE STREAM up_s () WITH (TYPE=\"memory\", FORMAT=\"json\")"}))
        .send()
        .await;
    if s_res.is_err() || !s_res.unwrap().status().is_success() {
        let _ = old_child.kill().await;
        return Err("Failed to create stream in old binary".to_string());
    }

    let r_res = client
        .post(format!("{}/rules", base_url))
        .json(&json!({
            "id": "up_r",
            "sql": "SELECT v FROM up_s",
            "actions": [{"log": {}}]
        }))
        .send()
        .await;
    if r_res.is_err() || !r_res.unwrap().status().is_success() {
        let _ = old_child.kill().await;
        return Err("Failed to create rule in old binary".to_string());
    }

    // Terminate old binary
    let _ = old_child.kill().await;
    let _ = old_child.wait().await;

    // 3. Upgrade: start new binary on same test directory
    let mut new_child =
        start_kuiperd_child(new_bin, &test_dir.path, rest_port, rpc_port, sse_port).await;
    if !wait_for_healthy(&base_url, Duration::from_secs(10)).await {
        let _ = new_child.kill().await;
        return Err(format!(
            "New upgraded binary at {:?} failed to start",
            new_bin
        ));
    }

    // 4. Verify catalog definitions survived upgrade
    let rule_resp = client.get(format!("{}/rules/up_r", base_url)).send().await;
    if rule_resp.is_err() || !rule_resp.unwrap().status().is_success() {
        let _ = new_child.kill().await;
        return Err("Rule created in old binary was not found in upgraded binary".to_string());
    }

    // Add new stream in upgraded binary to test forward schema compatibility
    let s2_res = client
        .post(format!("{}/streams", base_url))
        .json(&json!({"sql": "CREATE STREAM up_s2 () WITH (TYPE=\"memory\", FORMAT=\"json\")"}))
        .send()
        .await;
    if s2_res.is_err() || !s2_res.unwrap().status().is_success() {
        let _ = new_child.kill().await;
        return Err("Failed to create new stream in upgraded binary".to_string());
    }

    // Terminate new binary
    let _ = new_child.kill().await;
    let _ = new_child.wait().await;

    // 5. Rollback: start old binary again on same test directory
    let mut rollback_child =
        start_kuiperd_child(old_bin, &test_dir.path, rest_port, rpc_port, sse_port).await;
    if !wait_for_healthy(&base_url, Duration::from_secs(10)).await {
        let _ = rollback_child.kill().await;
        return Err("Old binary failed to start after rollback".to_string());
    }

    let rule_resp2 = client.get(format!("{}/rules/up_r", base_url)).send().await;
    if rule_resp2.is_err() || !rule_resp2.unwrap().status().is_success() {
        let _ = rollback_child.kill().await;
        return Err("Rule damaged after rollback to old binary".to_string());
    }

    let _ = rollback_child.kill().await;
    let _ = rollback_child.wait().await;

    Ok(())
}

/// Bounded $O(1)$ memory sequence tracker: records gaps, out-of-order deliveries,
/// and duplicates without retaining millions of seen sequence IDs.
#[derive(Default)]
struct BoundedSequenceTracker {
    total_received: u64,
    next_expected: u64,
    duplicates: u64,
    out_of_order: u64,
    missing_gaps: BTreeSet<u64>,
    overflow_gaps_count: u64,
}

impl BoundedSequenceTracker {
    fn record(&mut self, seq: u64) {
        self.total_received += 1;
        if seq == self.next_expected {
            self.next_expected += 1;
        } else if seq > self.next_expected {
            for gap in self.next_expected..seq {
                if self.missing_gaps.len() < 20_000 {
                    self.missing_gaps.insert(gap);
                } else {
                    self.overflow_gaps_count += 1;
                }
            }
            self.next_expected = seq + 1;
        } else {
            // seq < next_expected: either out-of-order filling a gap, or a duplicate
            if self.missing_gaps.remove(&seq) {
                self.out_of_order += 1;
            } else {
                self.duplicates += 1;
            }
        }
    }

    fn first_unconfirmed(&self) -> u64 {
        self.missing_gaps
            .iter()
            .next()
            .copied()
            .unwrap_or(self.next_expected)
    }

    fn finalize(&mut self, total_sent: u64) {
        if self.next_expected < total_sent {
            for tail in self.next_expected..total_sent {
                if self.missing_gaps.len() < 20_000 {
                    self.missing_gaps.insert(tail);
                } else {
                    self.overflow_gaps_count += 1;
                }
            }
        }
    }

    fn total_missing(&self) -> u64 {
        self.missing_gaps.len() as u64 + self.overflow_gaps_count
    }
}

/// Executes process-based qualification with real duration, pacing, child kill, and MQTT validation.
pub async fn run_qualification(
    mode: &str,
    requested_duration: Duration,
    target_rate: u64,
) -> QualificationMetrics {
    println!("============================================================");
    println!("Starting Process-Based Reliability Qualification Harness");
    println!(
        "Mode: {} | Requested Duration: {:?} | Target Rate: {} msg/s",
        mode, requested_duration, target_rate
    );
    println!("============================================================");

    let kuiperd_bin = find_kuiperd_bin();
    let (git_rev, dirty_tree) = gather_git_info();

    let mut sys = sysinfo::System::new();
    sys.refresh_all();

    let log_path = get_qualification_log_path();

    let mut metrics = QualificationMetrics {
        mode: mode.to_string(),
        source_revision: git_rev,
        dirty_tree,
        binary_path: kuiperd_bin.to_string_lossy().to_string(),
        engine_log_path: log_path.to_string_lossy().to_string(),
        host_os: std::env::consts::OS.to_string(),
        host_cpus: sys.cpus().len(),
        host_memory_mb: sys.total_memory() as f64 / (1024.0 * 1024.0),
        requested_duration_seconds: requested_duration.as_secs_f64(),
        requested_rate: target_rate,
        timestamp_utc: chrono::Utc::now().to_rfc3339(),
        status: "RUNNING".to_string(),
        ..Default::default()
    };

    // 1. Setup isolated directories & broker
    let test_dir = TestEnvDir::new("proc_qual");
    let rest_port = get_free_port().await;
    let rpc_port = get_free_port().await;
    let sse_port = get_free_port().await;
    let base_url = format!("http://127.0.0.1:{}", rest_port);

    // Support real external Mosquitto (via local fault-injecting proxy) or embedded mock
    let configured_mqtt = std::env::var("REKUIPER_TEST_MQTT").ok();
    let upstream_broker = if let Some(ref url) = configured_mqtt {
        Some(url.clone())
    } else if probe_tcp("127.0.0.1:1883").await {
        Some("tcp://127.0.0.1:1883".to_string())
    } else {
        None
    };

    let (broker_url, proxy_opt, mock_broker_opt, broker_type) =
        if let Some(ref upstream) = upstream_broker {
            println!(
                "-> Detected real Mosquitto at {}; starting fault-injecting proxy...",
                upstream
            );
            let (_, upstream_port) = rekuiper_connectors::parse_mqtt_server_url(upstream).unwrap();
            let upstream_addr = format!("127.0.0.1:{}", upstream_port);
            let proxy = FaultInjectingProxy::start(upstream_addr).await;
            let local_url = format!("tcp://127.0.0.1:{}", proxy.port);
            (
                local_url,
                Some(proxy),
                None,
                "real_mosquitto_proxy".to_string(),
            )
        } else {
            if std::env::var("REKUIPER_REQUIRE_BROKER").as_deref() == Ok("1")
                || std::env::var("REKUIPER_REQUIRE_MQTT").as_deref() == Ok("1")
            {
                panic!(
                "REKUIPER_REQUIRE_BROKER is set, but no real MQTT broker is running or reachable!"
            );
            }
            println!("-> Real MQTT broker not detected; starting embedded mock broker...");
            let b = FaultInjectingBroker::start().await;
            let url = format!("tcp://127.0.0.1:{}", b.port);
            (url, None, Some(b), "embedded_mock".to_string())
        };

    metrics.broker_type = broker_type;
    println!("-> Preserving engine logs to: {}", metrics.engine_log_path);

    // 2. Start independent MQTT subscriber client
    let (broker_host, broker_port) =
        rekuiper_connectors::parse_mqtt_server_url(&broker_url).unwrap();
    let mut mqtt_opts = MqttOptions::new("qual_independent_sub", broker_host, broker_port);
    mqtt_opts.set_keep_alive(Duration::from_secs(5));
    mqtt_opts.set_clean_session(false);
    let (mqtt_client, mut mqtt_eventloop) = AsyncClient::new(mqtt_opts, 100);

    let tracker = Arc::new(Mutex::new(BoundedSequenceTracker::default()));
    let sub_running = Arc::new(AtomicBool::new(true));
    let restore_time_state = Arc::new(Mutex::new(None::<Instant>));
    let broker_recovery_measured = Arc::new(AtomicBool::new(false));
    let measured_broker_recovery_ms = Arc::new(AtomicU64::new(0));

    let trk_clone = tracker.clone();
    let run_clone = sub_running.clone();
    let rec_state_clone = restore_time_state.clone();
    let rec_meas_clone = broker_recovery_measured.clone();
    let rec_ms_clone = measured_broker_recovery_ms.clone();

    // Background task for independent MQTT subscriber
    tokio::spawn(async move {
        let _ = mqtt_client.subscribe("qual/sink", QoS::AtLeastOnce).await;

        while run_clone.load(Ordering::Relaxed) {
            match tokio::time::timeout(Duration::from_millis(100), mqtt_eventloop.poll()).await {
                Ok(Ok(rumqttc::Event::Incoming(Packet::Publish(publish)))) => {
                    // Record broker recovery time on first message received after restore
                    if !rec_meas_clone.load(Ordering::Relaxed) {
                        if let Some(restored_at) = *rec_state_clone.lock().await {
                            let elapsed_ms = restored_at.elapsed().as_millis() as u64;
                            rec_ms_clone.store(elapsed_ms.max(1), Ordering::Relaxed);
                            rec_meas_clone.store(true, Ordering::Relaxed);
                            println!(
                                "-> [PHASE 3] Delivery restored to MQTT subscriber in {} ms!",
                                elapsed_ms
                            );
                        }
                    }

                    if let Ok(v) = serde_json::from_slice::<Value>(&publish.payload) {
                        let item = if let Some(arr) = v.as_array() {
                            arr.first()
                        } else {
                            Some(&v)
                        };
                        if let Some(seq) = item.and_then(|m| m.get("seq")).and_then(|s| s.as_u64())
                        {
                            trk_clone.lock().await.record(seq);
                        }
                    }
                }
                Ok(Ok(rumqttc::Event::Incoming(Packet::ConnAck(_)))) => {
                    let _ = mqtt_client.subscribe("qual/sink", QoS::AtLeastOnce).await;
                }
                _ => {}
            }
        }
    });

    // 3. Launch kuiperd child process
    println!("-> Spawning kuiperd child process on port {}...", rest_port);
    let mut child =
        start_kuiperd_child(&kuiperd_bin, &test_dir.path, rest_port, rpc_port, sse_port).await;

    assert!(
        wait_for_healthy(&base_url, Duration::from_secs(10)).await,
        "kuiperd child process failed to start within 10s"
    );

    let mut current_child_pid = child.id().expect("child PID");
    let initial_rss = get_process_rss_mb(current_child_pid);
    metrics.initial_rss_mb = initial_rss;
    metrics.peak_rss_mb = initial_rss;

    let client = reqwest::Client::new();
    let stream_id = "qual_stream";
    let rule_id = "qual_rule";

    // 4. Create Stream & Rule with MQTT Sink (with offline caching)
    let create_stream_resp = client
        .post(format!("{}/streams", base_url))
        .json(&json!({
            "sql": format!("CREATE STREAM {} () WITH (DATASOURCE=\"{}\", FORMAT=\"json\")", stream_id, stream_id)
        }))
        .send()
        .await
        .unwrap();
    assert!(create_stream_resp.status().is_success());

    let create_rule_resp = client
        .post(format!("{}/rules", base_url))
        .json(&json!({
            "id": rule_id,
            "sql": format!("SELECT seq, payload, resent FROM {}", stream_id),
            "actions": [
                {
                    "mqtt": {
                        "server": broker_url,
                        "topic": "qual/sink",
                        "qos": 1,
                        "clientId": "qual_sink_worker",
                        "cleanSession": false,
                        "enableCache": true,
                        "resendPriority": 1,
                        "memoryCacheThreshold": 1000,
                        "bufferPageSize": 100,
                        "resendIndicatorField": "resent"
                    }
                }
            ]
        }))
        .send()
        .await
        .unwrap();
    assert!(create_rule_resp.status().is_success());

    // 5. Continuous Paced Traffic Generation
    println!("-> Starting traffic generation at {} msg/s...", target_rate);
    let run_start = Instant::now();
    let interval_micros = (1_000_000.0 / target_rate as f64).max(100.0) as u64;
    let mut ticker = tokio::time::interval(Duration::from_micros(interval_micros));
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Burst);

    let mut current_seq = 0u64;
    let stream_url = format!("{}/streams/{}/data", base_url, stream_id);

    let outage_point = requested_duration.mul_f64(0.25);
    let restore_point = requested_duration.mul_f64(0.45);
    let crash_point = requested_duration.mul_f64(0.75);

    let mut outage_injected = false;
    let mut outage_restored = false;
    let mut engine_crashed = false;
    let mut last_mem_sample = Instant::now();

    while run_start.elapsed() < requested_duration {
        let elapsed = run_start.elapsed();

        // Sample memory of actual kuiperd child process periodically
        if last_mem_sample.elapsed() > Duration::from_millis(500) {
            let cur_rss = get_process_rss_mb(current_child_pid);
            if cur_rss > metrics.peak_rss_mb {
                metrics.peak_rss_mb = cur_rss;
            }
            last_mem_sample = Instant::now();
        }

        // Phase 2: Broker Outage
        if !outage_injected && elapsed >= outage_point && elapsed < restore_point {
            println!("-> [PHASE 2] Injecting broker outage...");
            if let Some(ref p) = proxy_opt {
                p.simulate_outage();
            } else if let Some(ref b) = mock_broker_opt {
                b.simulate_outage();
            }
            outage_injected = true;
        }

        // Phase 3: Broker Restore
        if outage_injected && !outage_restored && elapsed >= restore_point && elapsed < crash_point
        {
            println!("-> [PHASE 3] Restoring broker and measuring delivery recovery...");
            let rec_start = Instant::now();
            *restore_time_state.lock().await = Some(rec_start);
            if let Some(ref p) = proxy_opt {
                p.restore();
            } else if let Some(ref b) = mock_broker_opt {
                b.restore();
            }
            outage_restored = true;
        }

        // Phase 4: Abrupt Process Kill and Restart
        if outage_restored && !engine_crashed && elapsed >= crash_point {
            println!("-> [PHASE 4] Terminating kuiperd process abruptly (kill)...");
            let pre_crash_rss = get_process_rss_mb(current_child_pid);
            if pre_crash_rss > metrics.peak_rss_mb {
                metrics.peak_rss_mb = pre_crash_rss;
            }
            let in_flight_start = tracker.lock().await.first_unconfirmed().min(current_seq);
            let in_flight = current_seq.saturating_sub(in_flight_start).min(500);
            metrics.in_flight_at_shutdown = in_flight;
            let bounded_start = current_seq.saturating_sub(in_flight);
            let in_flight_ids: Vec<u64> = (bounded_start..current_seq).collect();
            metrics.in_flight_ids_at_shutdown = in_flight_ids;
            println!(
                "-> [PHASE 4] In-flight before crash: {} record(s) (IDs: {:?})",
                metrics.in_flight_at_shutdown, metrics.in_flight_ids_at_shutdown
            );

            let _ = child.kill().await;
            let _ = child.wait().await;

            let rec_start = Instant::now();
            println!("-> [PHASE 4] Restarting kuiperd process with persisted state...");
            child =
                start_kuiperd_child(&kuiperd_bin, &test_dir.path, rest_port, rpc_port, sse_port)
                    .await;

            assert!(
                wait_for_healthy(&base_url, Duration::from_secs(10)).await,
                "Restarted kuiperd child process failed to become healthy"
            );
            metrics.engine_recovery_time_ms = rec_start.elapsed().as_millis() as u64;
            current_child_pid = child.id().expect("restarted child PID");
            println!(
                "-> [PHASE 4] Engine restored in {} ms (PID: {})",
                metrics.engine_recovery_time_ms, current_child_pid
            );
            engine_crashed = true;
        }

        ticker.tick().await;

        let payload = json!({
            "seq": current_seq,
            "payload": format!("sensor_reading_{}", current_seq),
            "ts": chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0)
        });

        match client.post(&stream_url).json(&payload).send().await {
            Ok(resp) if resp.status().is_success() => {
                metrics.accepted_input += 1;
                current_seq += 1;
            }
            Ok(resp) => {
                metrics.error_count += 1;
                tracing::warn!("HTTP push error: {}", resp.status());
            }
            Err(_) => {
                // Network drops during the exact crash millisecond are expected in-flight
                if !engine_crashed {
                    metrics.error_count += 1;
                }
            }
        }
    }

    metrics.measured_duration_seconds = run_start.elapsed().as_secs_f64();
    metrics.achieved_rate =
        metrics.accepted_input as f64 / metrics.measured_duration_seconds.max(0.001);

    // 6. Settle Phase: Drain sink buffer
    println!("-> Waiting for sink replay and drain to complete...");
    tokio::time::sleep(Duration::from_millis(1500)).await;

    // Stop subscriber
    sub_running.store(false, Ordering::Relaxed);

    // 7. Validate Independent MQTT Sink Output via Bounded Tracker
    {
        let mut trk = tracker.lock().await;
        trk.finalize(current_seq);
        metrics.missing_sample_ids = trk.missing_gaps.iter().take(20).copied().collect();
        metrics.missing_records = trk.total_missing();
        metrics.delivered_output = trk.total_received;
        metrics.duplicate_records = trk.duplicates;
        metrics.out_of_order_count = trk.out_of_order;
        metrics.unique_delivered = trk.total_received.saturating_sub(trk.duplicates);
    }

    // Collect measured broker recovery latency (time until actual post-restore delivery)
    metrics.broker_recovery_time_ms = measured_broker_recovery_ms.load(Ordering::Relaxed);

    // 8. Storage Fault Injection (Independent verification in isolated DB)
    println!("-> Phase 5: Exercising storage write failure in isolated storage...");
    struct FaultyStore {
        fail: Arc<AtomicBool>,
        inner: Arc<dyn KvStore>,
    }

    #[async_trait]
    impl KvStore for FaultyStore {
        async fn get(&self, namespace: &str, key: &str) -> anyhow::Result<Option<String>> {
            self.inner.get(namespace, key).await
        }
        async fn set(&self, namespace: &str, key: &str, val: &str) -> anyhow::Result<()> {
            if self.fail.load(Ordering::Relaxed) {
                anyhow::bail!("Simulated ENOSPC write error in qualification");
            }
            self.inner.set(namespace, key, val).await
        }
        async fn delete(&self, namespace: &str, key: &str) -> anyhow::Result<()> {
            if self.fail.load(Ordering::Relaxed) {
                anyhow::bail!("Simulated ENOSPC delete error in qualification");
            }
            self.inner.delete(namespace, key).await
        }
        async fn list_all(&self, namespace: &str) -> anyhow::Result<Vec<(String, String)>> {
            self.inner.list_all(namespace).await
        }
    }

    let fault_flag = Arc::new(AtomicBool::new(true));
    let fault_db_path = test_dir.path.join("qual_fault.db");
    let base_kv: Arc<dyn KvStore> = Arc::new(SqliteKvStore::new(&fault_db_path).await.unwrap());
    let faulty_store: Arc<dyn KvStore> = Arc::new(FaultyStore {
        fail: fault_flag.clone(),
        inner: base_kv,
    });

    let faulty_mgr = StreamManager::new_with_kv(faulty_store.clone());
    let bad_res = faulty_mgr
        .create_stream(StreamDefinition {
            name: "fail_stream".to_string(),
            sql: "CREATE STREAM fail_stream () WITH (FORMAT=\"json\")".to_string(),
            stream_fields: vec![],
            options: HashMap::new(),
        })
        .await;
    assert!(bad_res.is_err());
    metrics.storage_fault_errors_caught += 1;

    // 9. Real Upgrade / Rollback Verification
    let old_bin_env = std::env::var("REKUIPER_OLD_BIN").ok();
    let new_bin_env = std::env::var("REKUIPER_NEW_BIN").ok();
    match (old_bin_env, new_bin_env) {
        (Some(old), Some(new)) if Path::new(&old).exists() && Path::new(&new).exists() => {
            println!(
                "-> Verifying real upgrade/rollback across {} and {}",
                old, new
            );
            match verify_upgrade_rollback(Path::new(&old), Path::new(&new)).await {
                Ok(_) => {
                    metrics.upgrade_rollback_status = "VERIFIED".to_string();
                    metrics.upgrade_rollback_verified = true;
                }
                Err(e) => {
                    metrics.upgrade_rollback_status = format!("FAILED: {}", e);
                    metrics.upgrade_rollback_verified = false;
                }
            }
        }
        (Some(_), Some(_)) => {
            metrics.upgrade_rollback_status =
                "FAILED: configured binary paths do not exist".to_string();
            metrics.upgrade_rollback_verified = false;
        }
        _ => {
            metrics.upgrade_rollback_status =
                "PENDING (set REKUIPER_OLD_BIN and REKUIPER_NEW_BIN to qualify)".to_string();
            metrics.upgrade_rollback_verified = false;
        }
    }

    // 10. Process Memory Growth & Teardown
    metrics.final_rss_mb = get_process_rss_mb(current_child_pid);
    if metrics.final_rss_mb > metrics.peak_rss_mb {
        metrics.peak_rss_mb = metrics.final_rss_mb;
    }
    metrics.memory_growth_mb = (metrics.peak_rss_mb - metrics.initial_rss_mb).max(0.0);

    // Clean up child process
    let _ = child.kill().await;
    let _ = child.wait().await;

    // 11. Strict Acceptance Criteria Evaluation
    let mut failure_reasons = Vec::new();

    if metrics.delivered_output == 0 {
        failure_reasons.push("Zero records delivered to MQTT sink".to_string());
    }
    let in_flight_set: std::collections::HashSet<u64> =
        metrics.in_flight_ids_at_shutdown.iter().copied().collect();
    let unexplained: Vec<u64> = tracker
        .lock()
        .await
        .missing_gaps
        .iter()
        .filter(|id| !in_flight_set.contains(id))
        .copied()
        .collect();
    metrics.unexplained_missing_records = unexplained.clone();

    if !unexplained.is_empty() {
        failure_reasons.push(format!(
            "Detected {} unexplained lost record(s): sample {:?} (missing records were not in-flight before crash {:?})",
            unexplained.len(),
            unexplained.iter().take(10).collect::<Vec<_>>(),
            metrics.in_flight_ids_at_shutdown
        ));
    }
    if metrics.missing_records > metrics.in_flight_at_shutdown {
        failure_reasons.push(format!(
            "Missing records ({}) exceed bounded in-flight at crash ({})",
            metrics.missing_records, metrics.in_flight_at_shutdown
        ));
    }
    if !engine_crashed && metrics.duplicate_records > 0 {
        failure_reasons.push(format!(
            "Detected {} duplicate records in sink delivery without crash",
            metrics.duplicate_records
        ));
    } else if metrics.duplicate_records > metrics.in_flight_at_shutdown.max(10) {
        failure_reasons.push(format!(
            "Detected {} duplicate records in sink delivery exceeding bounded in-flight at crash ({})",
            metrics.duplicate_records, metrics.in_flight_at_shutdown
        ));
    }
    if metrics.out_of_order_count > 0 {
        failure_reasons.push(format!(
            "Detected {} out-of-order records in sink delivery",
            metrics.out_of_order_count
        ));
    }
    if metrics.error_count > 0 {
        failure_reasons.push(format!(
            "Encountered {} unexpected engine push errors",
            metrics.error_count
        ));
    }
    let min_rate_factor = if engine_crashed || outage_injected {
        0.50
    } else {
        0.70
    };
    if metrics.achieved_rate < (metrics.requested_rate as f64 * min_rate_factor) {
        failure_reasons.push(format!(
            "Achieved rate ({:.2} msg/s) fell below {:.0}% of requested rate ({} msg/s)",
            metrics.achieved_rate,
            min_rate_factor * 100.0,
            metrics.requested_rate
        ));
    }
    if outage_injected {
        if !broker_recovery_measured.load(Ordering::Relaxed) || metrics.broker_recovery_time_ms == 0
        {
            failure_reasons.push(
                "Broker recovery failed: no messages delivered to subscriber after broker restore"
                    .to_string(),
            );
        } else if metrics.broker_recovery_time_ms > 10_000 {
            failure_reasons.push(format!(
                "Broker recovery took too long: {} ms",
                metrics.broker_recovery_time_ms
            ));
        }
    }
    if engine_crashed
        && (metrics.engine_recovery_time_ms == 0 || metrics.engine_recovery_time_ms > 10_000)
    {
        failure_reasons.push(format!(
            "Engine recovery took too long or failed: {} ms",
            metrics.engine_recovery_time_ms
        ));
    }
    if metrics.storage_fault_errors_caught == 0 {
        failure_reasons.push("Storage fault error was not caught".to_string());
    }
    if metrics.initial_rss_mb == 0.0 || metrics.final_rss_mb == 0.0 {
        failure_reasons.push(format!(
            "Process memory monitoring failed (initial: {:.2} MB, final: {:.2} MB)",
            metrics.initial_rss_mb, metrics.final_rss_mb
        ));
    }
    if metrics.memory_growth_mb > 50.0 {
        failure_reasons.push(format!(
            "Excessive memory growth ({:.2} MB > 50 MB threshold, initial: {:.2} MB, peak: {:.2} MB)",
            metrics.memory_growth_mb, metrics.initial_rss_mb, metrics.peak_rss_mb
        ));
    }
    if mode == "upgrade" && !metrics.upgrade_rollback_verified {
        failure_reasons.push(format!(
            "Upgrade qualification requested but verification failed or binaries missing: {}",
            metrics.upgrade_rollback_status
        ));
    }

    metrics.status = if failure_reasons.is_empty() {
        "PASSED".to_string()
    } else {
        "FAILED".to_string()
    };
    metrics.failure_reasons = failure_reasons;

    println!("============================================================");
    println!("Qualification Run Summary (Mode: {})", metrics.mode);
    println!("  Status:                   {}", metrics.status);
    println!(
        "  Duration:                 {:.2}s measured (requested: {:.2}s)",
        metrics.measured_duration_seconds, metrics.requested_duration_seconds
    );
    println!(
        "  Achieved Rate:            {:.2} msg/s (requested: {} msg/s)",
        metrics.achieved_rate, metrics.requested_rate
    );
    println!("  Accepted Input:           {}", metrics.accepted_input);
    println!("  Delivered Output:         {}", metrics.delivered_output);
    println!("  Unique Delivered:         {}", metrics.unique_delivered);
    println!("  Missing Records:          {}", metrics.missing_records);
    if !metrics.missing_sample_ids.is_empty() {
        println!(
            "  Missing Sample IDs:       {:?}",
            metrics.missing_sample_ids
        );
    }
    if !metrics.unexplained_missing_records.is_empty() {
        println!(
            "  Unexplained Lost IDs:     {:?}",
            metrics.unexplained_missing_records
        );
    }
    println!("  Duplicate Records:        {}", metrics.duplicate_records);
    println!(
        "  In-Flight at Crash:       {} (IDs: {:?})",
        metrics.in_flight_at_shutdown, metrics.in_flight_ids_at_shutdown
    );
    println!("  Out-of-Order Count:       {}", metrics.out_of_order_count);
    println!(
        "  Storage Faults Caught:    {}",
        metrics.storage_fault_errors_caught
    );
    println!(
        "  Upgrade/Rollback Status:  {}",
        metrics.upgrade_rollback_status
    );
    println!("  Broker Type:              {}", metrics.broker_type);
    println!(
        "  Broker Recovery:          {} ms",
        metrics.broker_recovery_time_ms
    );
    println!(
        "  Engine Recovery:          {} ms",
        metrics.engine_recovery_time_ms
    );
    println!(
        "  Engine PID {} Memory:     {:.2} MB -> {:.2} MB (Peak: {:.2} MB, Growth: {:.2} MB)",
        current_child_pid,
        metrics.initial_rss_mb,
        metrics.final_rss_mb,
        metrics.peak_rss_mb,
        metrics.memory_growth_mb
    );
    println!("  Engine Logs Preserved:    {}", metrics.engine_log_path);
    if !metrics.failure_reasons.is_empty() {
        println!("  Failure Reasons:          {:?}", metrics.failure_reasons);
    }
    println!("============================================================");

    metrics
}

#[tokio::test]
async fn qualification_short_test() {
    let mode = std::env::var("REKUIPER_QUAL_MODE").unwrap_or_else(|_| "short".to_string());
    let duration_secs: u64 = std::env::var("REKUIPER_QUAL_DURATION_SECS")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(5);
    let target_rate: u64 = std::env::var("REKUIPER_QUAL_RATE")
        .ok()
        .and_then(|s| s.parse().ok())
        .unwrap_or(50);

    let metrics = run_qualification(&mode, Duration::from_secs(duration_secs), target_rate).await;

    // Save machine-readable JSON metrics report
    let out_path = std::env::var("REKUIPER_QUAL_OUTPUT")
        .unwrap_or_else(|_| "target/qualification_results.json".to_string());
    let out_dir = Path::new(&out_path)
        .parent()
        .unwrap_or_else(|| Path::new("."));
    let _ = std::fs::create_dir_all(out_dir);

    let json_str = serde_json::to_string_pretty(&metrics).unwrap();
    let _ = std::fs::write(&out_path, &json_str);

    let root_target = Path::new("../../target/qualification_results.json");
    if let Some(parent) = root_target.parent() {
        if parent.exists() {
            let _ = std::fs::write(root_target, &json_str);
        }
    }
    println!(
        "Machine-readable qualification report written to: {}",
        out_path
    );

    assert_eq!(
        metrics.status, "PASSED",
        "Qualification run failed: {:?}",
        metrics.failure_reasons
    );
}