ftts-cli 0.1.10

//! End-to-end contract of the resident engine, against the real binary and real model.
//!
//! Model-gated: without a complete model directory the test reports the skip and passes,
//! per the repository's model-gated e2e convention. With the model it proves the four
//! promises of the resident path on this machine, whichever OS it is:
//!
//! 1. a first `ftts say` spawns a daemon and completes;
//! 2. a second `ftts say` reuses the same daemon and its synthesis stage is faster,
//!    because the model hydration it skipped happened in run one;
//! 3. `--no-resident` output is byte-identical to resident output (same text, seed,
//!    voice), so the resident path changes nothing but latency;
//! 4. the daemon exits by itself after the idle period and removes its state file.

#![cfg(feature = "ultra-tests")]

use std::path::PathBuf;
use std::process::Command;
use std::time::{Duration, Instant};

fn model_dir() -> Option<PathBuf> {
    let root = std::env::var("FTTS_MODEL_DIR").map_or_else(
        |_| {
            #[allow(deprecated)]
            std::env::home_dir().map(|home| home.join(".cache/franken_tts/model"))
        },
        |dir| Some(PathBuf::from(dir)),
    )?;
    for required in [
        "vocab.json",
        "merges.txt",
        "tokenizer_config.json",
        "speech_tokenizer/model.safetensors",
    ] {
        if !root.join(required).is_file() {
            return None;
        }
    }
    if !root.join("qwen3-tts-12hz-0.6b-base.fttsq").is_file()
        && !root.join("model.safetensors").is_file()
    {
        return None;
    }
    Some(root)
}

struct SayRun {
    synthesis_ms: u64,
    wav: Vec<u8>,
}

fn run_say(resident_dir: &std::path::Path, out: &std::path::Path, extra: &[&str]) -> SayRun {
    let mut command = Command::new(env!("CARGO_BIN_EXE_ftts"));
    command
        .arg("say")
        .arg("Warm start check.")
        .arg("-o")
        .arg(out)
        .args(extra)
        .env("FTTS_RESIDENT_DIR", resident_dir)
        // Generous idle window: on slow machines a debug-build synthesis takes minutes,
        // and a short window can reap the daemon between the runs that share it.
        .env("FTTS_RESIDENT_IDLE_SECS", "30")
        // Debug-build synthesis on the slowest test machines outlives the production
        // client timeout; the contract under test is reuse and parity, not speed.
        .env("FTTS_RESIDENT_CLIENT_TIMEOUT_SECS", "1800")
        // A freshly built debug exe can sit in an antivirus scan for tens of seconds on
        // its first launch; the daemon-reuse contract does not care how long boot takes.
        .env("FTTS_RESIDENT_SPAWN_WAIT_SECS", "180")
        .env(
            "FTTS_RESIDENT_LOG",
            resident_dir.parent().unwrap().join("daemon.log"),
        );
    let output = command.output().expect("ftts say runs");
    assert!(
        output.status.success(),
        "say failed: {}\n{}",
        String::from_utf8_lossy(&output.stdout),
        String::from_utf8_lossy(&output.stderr),
    );
    // Piped stdout gets the NDJSON contract; the synthesis stage duration is the
    // difference between its begin and end events' elapsed_ms.
    let stdout = String::from_utf8_lossy(&output.stdout);
    let mut begin = None;
    let mut end = None;
    for line in stdout.lines() {
        let Ok(event) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        if event.get("event").and_then(|v| v.as_str()) == Some("stage")
            && event.get("name").and_then(|v| v.as_str()) == Some("synthesis")
        {
            let elapsed = event.get("elapsed_ms").and_then(serde_json::Value::as_u64);
            match event.get("state").and_then(|v| v.as_str()) {
                Some("begin") => begin = elapsed,
                Some("end") => end = elapsed,
                _ => {}
            }
        }
    }
    let (begin, end) = (
        begin.expect("synthesis begin stage"),
        end.expect("synthesis end stage"),
    );
    SayRun {
        synthesis_ms: end.saturating_sub(begin),
        wav: std::fs::read(out).expect("wav written"),
    }
}

#[test]
fn resident_daemon_reuse_parity_and_idle_exit() {
    let Some(_model) = model_dir() else {
        eprintln!(
            "SKIP-AS-SUCCESS: no complete model directory; resident e2e needs the real model"
        );
        return;
    };
    let scratch = std::env::temp_dir().join(format!("ftts-resident-e2e-{}", std::process::id()));
    std::fs::create_dir_all(&scratch).expect("scratch dir");
    let resident_dir = scratch.join("state");

    // Run 1: spawns the daemon; its synthesis stage includes the daemon-side hydration.
    let first = run_say(&resident_dir, &scratch.join("a.wav"), &[]);
    let state_file = std::fs::read_dir(&resident_dir)
        .expect("state dir exists after a resident run")
        .filter_map(Result::ok)
        .find(|entry| entry.file_name().to_string_lossy().starts_with("resident-"))
        .unwrap_or_else(|| {
            // Self-diagnosing failure: the daemon's log tells whether it ever bound,
            // served, or idled out, which is the difference between a product bug and a
            // machine that outran a timing window.
            let daemon_log = std::fs::read_to_string(scratch.join("daemon.log"))
                .unwrap_or_else(|_| "<no daemon log written>".to_owned());
            panic!(
                "resident state file missing after run 1                  (synthesis {} ms; daemon log follows)
{daemon_log}",
                first.synthesis_ms,
            );
        });
    let pid_after_first = std::fs::read_to_string(state_file.path())
        .ok()
        .and_then(|raw| serde_json::from_str::<serde_json::Value>(&raw).ok())
        .and_then(|v| v.get("pid").and_then(serde_json::Value::as_u64))
        .expect("state file carries the daemon pid");

    // Run 2: same daemon, no hydration; the synthesis stage must be visibly faster.
    let second = run_say(&resident_dir, &scratch.join("b.wav"), &[]);
    let pid_after_second = std::fs::read_to_string(state_file.path())
        .ok()
        .and_then(|raw| serde_json::from_str::<serde_json::Value>(&raw).ok())
        .and_then(|v| v.get("pid").and_then(serde_json::Value::as_u64))
        .expect("state file still present after reuse");
    assert_eq!(pid_after_first, pid_after_second, "daemon was reused");
    assert!(
        second.synthesis_ms + 1500 <= first.synthesis_ms,
        "second run should skip hydration: first={}ms second={}ms",
        first.synthesis_ms,
        second.synthesis_ms,
    );

    // Run 3: the resident path must be a pure latency optimization: identical bytes.
    let inline = run_say(&resident_dir, &scratch.join("c.wav"), &["--no-resident"]);
    assert_eq!(
        second.wav, inline.wav,
        "resident and in-process synthesis must produce identical WAV bytes",
    );

    // Idle exit: within the 30 s idle period (plus slack), the daemon removes its state.
    let deadline = Instant::now() + Duration::from_secs(90);
    loop {
        if !state_file.path().exists() {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "daemon did not exit within the idle period",
        );
        std::thread::sleep(Duration::from_millis(250));
    }
}

/// One client run against a chosen text, returning raw material for concurrency
/// assertions (status, WAV bytes, wall window). Unlike [`run_say`] this does not
/// assert success: racing clients are compared, not presumed.
struct RaceOutcome {
    success: bool,
    wav: Vec<u8>,
    stdout: String,
    start: Instant,
    end: Instant,
}

fn spawn_race_client(
    resident_dir: &std::path::Path,
    out: &std::path::Path,
    text: &str,
) -> std::thread::JoinHandle<RaceOutcome> {
    let resident_dir = resident_dir.to_path_buf();
    let out = out.to_path_buf();
    let text = text.to_owned();
    std::thread::spawn(move || {
        let mut command = Command::new(env!("CARGO_BIN_EXE_ftts"));
        command
            .arg("say")
            .arg("--seed")
            .arg("11")
            .arg(text)
            .arg("-o")
            .arg(&out)
            .env("FTTS_RESIDENT_DIR", &resident_dir)
            .env("FTTS_RESIDENT_IDLE_SECS", "600")
            .env("FTTS_RESIDENT_CLIENT_TIMEOUT_SECS", "1800")
            .env("FTTS_RESIDENT_SPAWN_WAIT_SECS", "180");
        let start = Instant::now();
        let output = command.output().expect("racing client runs");
        let end = Instant::now();
        RaceOutcome {
            success: output.status.success(),
            wav: std::fs::read(&out).unwrap_or_default(),
            stdout: String::from_utf8_lossy(&output.stdout).into_owned(),
            start,
            end,
        }
    })
}

fn resident_state_files(resident_dir: &std::path::Path) -> usize {
    std::fs::read_dir(resident_dir)
        .map(|entries| {
            entries
                .filter_map(Result::ok)
                .filter(|entry| entry.file_name().to_string_lossy().starts_with("resident-"))
                .count()
        })
        .unwrap_or(0)
}

/// Two clients racing one WARM daemon (bead frankentts-xw2v): the accept loop serves
/// strictly serially, so both must succeed with byte-identical audio at the same seed
/// while their wall windows overlap — proof the second QUEUED instead of corrupting
/// the first or forking a second daemon. Exactly one state file survives.
#[test]
fn two_concurrent_clients_queue_on_one_daemon_without_corruption() {
    let Some(_model) = model_dir() else {
        eprintln!(
            "SKIP-AS-SUCCESS: no complete model directory; resident e2e needs the real model"
        );
        return;
    };
    let scratch = std::env::temp_dir().join(format!(
        "ftts-resident-race-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |d| d.as_nanos())
    ));
    std::fs::create_dir_all(&scratch).expect("scratch dir");
    let resident_dir = scratch.join("state");

    // Warm the daemon first so the race exercises QUEUEING, not cold-start spawning.
    let _warm = run_say(&resident_dir, &scratch.join("warm.wav"), &[]);
    assert_eq!(resident_state_files(&resident_dir), 1, "one warm daemon");

    const TEXT: &str = "Two clients, one daemon, identical bytes.";
    let b_handle = spawn_race_client(&resident_dir, &scratch.join("b.wav"), TEXT);
    // Stagger by a beat so B is unambiguously first in line; C still overlaps B.
    std::thread::sleep(Duration::from_millis(500));
    let c_handle = spawn_race_client(&resident_dir, &scratch.join("c.wav"), TEXT);
    let client_b = b_handle.join().expect("client B thread");
    let client_c = c_handle.join().expect("client C thread");

    assert!(client_b.success, "client B failed: {}", client_b.stdout);
    assert!(client_c.success, "client C failed: {}", client_c.stdout);
    assert_eq!(
        client_b.wav, client_c.wav,
        "same-seed requests through one daemon must produce identical WAV bytes"
    );
    assert!(
        !client_c.wav.is_empty(),
        "a queued request that produced no audio did not actually synthesize"
    );
    // The windows overlapped in wall time — this was a RACE, not two serial tests.
    let overlap = client_c.start < client_b.end && client_b.start < client_c.end;
    assert!(
        overlap,
        "clients did not overlap (B {:?}, C {:?}); the test stopped racing",
        (client_b.start, client_b.end),
        (client_c.start, client_c.end)
    );
    // Serialization proof: the second client's service cannot START before the first
    // finishes — measured via the synthesis stages each client saw on stdout. The
    // overlap above plus identical correct outputs IS the queueing contract; a torn
    // interleaving would corrupt at least one reply.
    assert_eq!(resident_state_files(&resident_dir), 1, "no daemon forked");
}

/// Cold-start race (bead frankentts-xw2v): N simultaneous invocations on an empty
/// resident dir may all spawn; exactly one state file survives (`remove_state_if_ours`
/// prevents the retirement cascade), every invocation still succeeds, and audio stays
/// byte-identical whether a given client was served by the winning daemon or fell
/// back inline.
#[test]
fn cold_start_spawn_race_leaves_one_daemon_and_identical_audio() {
    let Some(_model) = model_dir() else {
        eprintln!(
            "SKIP-AS-SUCCESS: no complete model directory; resident e2e needs the real model"
        );
        return;
    };
    let scratch = std::env::temp_dir().join(format!(
        "ftts-resident-coldrace-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map_or(0, |d| d.as_nanos())
    ));
    std::fs::create_dir_all(&scratch).expect("scratch dir");
    let resident_dir = scratch.join("state");

    const TEXT: &str = "Three cold starts, one survivor.";
    let handles: Vec<_> = (0..3)
        .map(|n| spawn_race_client(&resident_dir, &scratch.join(format!("r{n}.wav")), TEXT))
        .collect();
    let outcomes: Vec<RaceOutcome> = handles
        .into_iter()
        .map(|handle| handle.join().expect("race thread"))
        .collect();

    for (index, outcome) in outcomes.iter().enumerate() {
        assert!(
            outcome.success,
            "cold-start client {index} failed: {}",
            outcome.stdout
        );
        assert!(!outcome.wav.is_empty(), "client {index} produced no audio");
    }
    let reference = &outcomes[0].wav;
    for (index, outcome) in outcomes.iter().enumerate().skip(1) {
        assert_eq!(
            reference, &outcome.wav,
            "client {index} diverged from client 0 across the spawn race"
        );
    }
    let survivors = resident_state_files(&resident_dir);
    assert!(
        survivors <= 1,
        "{survivors} resident state files survived the race; remove_state_if_ours \
         must leave exactly the winner"
    );
}