//! Watch module unit tests.

use super::*;
use notify::EventKind;
use serial_test::serial;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::LazyLock;

/// RAII guard that snapshots a process-global env var on construction and
/// restores it (set back to the prior value, or removed if it was unset) on
/// drop. Pairs with `#[serial(group)]`: serial prevents concurrent readers
/// from observing a half-set var, and the guard ensures a panicking test
/// leaves the env exactly as it found it instead of leaking the mutation into
/// later tests. Use one per env var that the watch paths read.
struct EnvVarGuard {
    key: &'static str,
    prev: Option<String>,
}

impl EnvVarGuard {
    /// Snapshot `key`, then set it to `value`.
    fn set(key: &'static str, value: &str) -> Self {
        let prev = std::env::var(key).ok();
        std::env::set_var(key, value);
        Self { key, prev }
    }

    /// Snapshot `key`, then remove it.
    fn remove(key: &'static str) -> Self {
        let prev = std::env::var(key).ok();
        std::env::remove_var(key);
        Self { key, prev }
    }
}

impl Drop for EnvVarGuard {
    fn drop(&mut self) {
        match &self.prev {
            Some(v) => std::env::set_var(self.key, v),
            None => std::env::remove_var(self.key),
        }
    }
}

// Shared test fixtures. These four fields are identical across every
// `test_watch_config*` call, so we keep exactly one `&'static` copy per type
// instead of leaking a fresh heap allocation on each call. The
// `test_watch_config_with_gitignore` helper still leaks its per-call matcher
// (each caller passes a distinct `Gitignore`).
static TEST_PARSER: LazyLock<CqParser> = LazyLock::new(|| CqParser::new().unwrap());
static TEST_EMBEDDER: LazyLock<std::sync::OnceLock<std::sync::Arc<Embedder>>> =
    LazyLock::new(std::sync::OnceLock::new);
static TEST_MODEL_CONFIG: LazyLock<ModelConfig> = LazyLock::new(ModelConfig::default_model);
static TEST_GITIGNORE_NONE: LazyLock<std::sync::RwLock<Option<ignore::gitignore::Gitignore>>> =
    LazyLock::new(|| std::sync::RwLock::new(None));

fn make_event(paths: Vec<PathBuf>, kind: EventKind) -> notify::Event {
    notify::Event {
        kind,
        paths,
        attrs: Default::default(),
    }
}

/// Helper to build a minimal WatchConfig for testing collect_events.
fn test_watch_config<'a>(
    root: &'a Path,
    cqs_dir: &'a Path,
    notes_path: &'a Path,
    supported_ext: &'a HashSet<&'a str>,
) -> WatchConfig<'a> {
    // These fields are unused by collect_events but required by the
    // struct. The four fixtures are shared `LazyLock` statics so
    // tests reference a single `&'static` copy instead of leaking a
    // fresh heap allocation on every call.
    WatchConfig {
        root,
        cqs_dir,
        notes_path,
        supported_ext,
        parser: &TEST_PARSER,
        embedder: &TEST_EMBEDDER,
        quiet: true,
        model_config: &TEST_MODEL_CONFIG,
        gitignore: &TEST_GITIGNORE_NONE,
        splade_encoder: None,
        global_cache: None,
    }
}

/// Variant that installs a gitignore matcher for .gitignore-specific tests.
fn test_watch_config_with_gitignore<'a>(
    root: &'a Path,
    cqs_dir: &'a Path,
    notes_path: &'a Path,
    supported_ext: &'a HashSet<&'a str>,
    matcher: ignore::gitignore::Gitignore,
) -> WatchConfig<'a> {
    // `parser` / `embedder` / `model_config` are shared statics (see
    // comment above); the per-call `matcher` still needs a distinct
    // `&'static RwLock`, so we leak that one field only.
    let gitignore = Box::leak(Box::new(std::sync::RwLock::new(Some(matcher))));
    WatchConfig {
        root,
        cqs_dir,
        notes_path,
        supported_ext,
        parser: &TEST_PARSER,
        embedder: &TEST_EMBEDDER,
        quiet: true,
        model_config: &TEST_MODEL_CONFIG,
        gitignore,
        splade_encoder: None,
        global_cache: None,
    }
}

fn test_watch_state() -> WatchState {
    WatchState {
        embedder_backoff: EmbedderBackoff::new(),
        pending_files: HashSet::new(),
        pending_notes: false,
        last_event: std::time::Instant::now(),
        first_pending_event: None,
        last_indexed_mtime: HashMap::new(),
        hnsw_index: None,
        incremental_count: 0,
        dropped_this_cycle: 0,
        pending_rebuild: None,
        // Throttle seed — tests that drive `publish_watch_snapshot` directly
        // want the very first call to re-stat (the cache starts empty). Tests
        // that don't touch the publish path don't care about these fields'
        // specific values.
        last_metadata_check: std::time::Instant::now()
            .checked_sub(std::time::Duration::from_secs(60))
            .unwrap_or_else(std::time::Instant::now),
        cached_last_synced_at: None,
        active_slot: cqs::slot::DEFAULT_SLOT.to_string(),
        last_reindex: None,
        last_error: None,
        observed_stamp: None,
    }
}

// ===== EmbedderBackoff tests =====

#[test]
fn backoff_initial_state_allows_retry() {
    let backoff = EmbedderBackoff::new();
    assert!(backoff.should_retry(), "Fresh backoff should allow retry");
}

#[test]
fn backoff_after_failure_delays_retry() {
    let mut backoff = EmbedderBackoff::new();
    backoff.record_failure();
    // After 1 failure, delay is 2^1 = 2 seconds
    assert!(
        !backoff.should_retry(),
        "Should not retry immediately after failure"
    );
    assert_eq!(backoff.failures, 1);
}

#[test]
fn backoff_reset_clears_failures() {
    let mut backoff = EmbedderBackoff::new();
    backoff.record_failure();
    backoff.record_failure();
    backoff.reset();
    assert_eq!(backoff.failures, 0);
    assert!(backoff.should_retry());
}

#[test]
fn backoff_caps_at_300s() {
    let mut backoff = EmbedderBackoff::new();
    // 2^9 = 512 > 300, so it should be capped
    for _ in 0..9 {
        backoff.record_failure();
    }
    // Verify it doesn't panic or overflow
    assert_eq!(backoff.failures, 9);
}

#[test]
fn backoff_saturating_add_no_overflow() {
    let mut backoff = EmbedderBackoff::new();
    backoff.failures = u32::MAX;
    backoff.record_failure();
    assert_eq!(backoff.failures, u32::MAX, "Should saturate, not overflow");
}

// ===== collect_events tests =====

#[test]
fn collect_events_filters_unsupported_extensions() {
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs", "py", "js"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();

    // .txt is not supported
    let event = make_event(
        vec![PathBuf::from("/tmp/test_project/readme.txt")],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );

    collect_events(&event, &cfg, &mut state);

    assert!(
        state.pending_files.is_empty(),
        "Unsupported extension should not be added"
    );
    assert!(!state.pending_notes);
}

#[test]
fn collect_events_skips_cqs_dir() {
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs", "db"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();

    let event = make_event(
        vec![PathBuf::from("/tmp/test_project/.cqs/index.db")],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );

    collect_events(&event, &cfg, &mut state);

    assert!(
        state.pending_files.is_empty(),
        ".cqs dir events should be skipped"
    );
}

/// Helper: build a `Gitignore` matcher in-memory from lines (no file IO).
fn gitignore_from_lines(root: &Path, lines: &[&str]) -> ignore::gitignore::Gitignore {
    let mut b = ignore::gitignore::GitignoreBuilder::new(root);
    for line in lines {
        b.add_line(None, line).expect("add_line");
    }
    b.build().expect("build gitignore")
}

#[test]
fn collect_events_skips_gitignore_matched_paths() {
    // `.claude/worktrees/` is a representative pollution case from
    // parallel-agent work. Verify that a path matched by .gitignore is
    // skipped.
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let matcher = gitignore_from_lines(&root, &[".claude/", "target/"]);
    let cfg = test_watch_config_with_gitignore(&root, &cqs_dir, &notes_path, &supported, matcher);

    let mut state = test_watch_state();
    let event = make_event(
        vec![PathBuf::from(
            "/tmp/test_project/.claude/worktrees/agent-a1b2c3d4/src/lib.rs",
        )],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );
    collect_events(&event, &cfg, &mut state);
    assert!(
        state.pending_files.is_empty(),
        ".gitignore-matched path .claude/worktrees/... should be skipped"
    );
}

#[test]
fn collect_events_skips_target_dir_via_gitignore() {
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let matcher = gitignore_from_lines(&root, &["target/"]);
    let cfg = test_watch_config_with_gitignore(&root, &cqs_dir, &notes_path, &supported, matcher);

    let mut state = test_watch_state();
    let event = make_event(
        vec![PathBuf::from("/tmp/test_project/target/debug/foo.rs")],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );
    collect_events(&event, &cfg, &mut state);
    assert!(
        state.pending_files.is_empty(),
        "target/ ignored by .gitignore should be skipped"
    );
}

#[test]
fn collect_events_does_not_skip_unrelated_paths_when_gitignore_present() {
    // False-positive guard: files under a directory not in .gitignore
    // must still be indexed even when a matcher is installed.
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let matcher = gitignore_from_lines(&root, &[".claude/", "target/"]);
    let cfg = test_watch_config_with_gitignore(&root, &cqs_dir, &notes_path, &supported, matcher);

    let mut state = test_watch_state();
    let event = make_event(
        vec![PathBuf::from("/tmp/test_project/src/foo.rs")],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );
    collect_events(&event, &cfg, &mut state);
    assert!(
        !state.pending_files.is_empty(),
        "src/foo.rs is not in .gitignore and must not be skipped"
    );
}

#[test]
fn collect_events_negations_include_path() {
    // `.gitignore` negations (`!foo`) keep the file indexed even
    // if a broader pattern excludes its parent.
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let matcher = gitignore_from_lines(&root, &["vendor/", "!vendor/keep/"]);
    let cfg = test_watch_config_with_gitignore(&root, &cqs_dir, &notes_path, &supported, matcher);

    let mut state = test_watch_state();
    let event = make_event(
        vec![PathBuf::from("/tmp/test_project/vendor/keep/lib.rs")],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );
    collect_events(&event, &cfg, &mut state);
    assert!(
        !state.pending_files.is_empty(),
        "negation `!vendor/keep/` must keep the file indexed"
    );
}

#[test]
fn collect_events_honors_none_matcher() {
    // With no matcher (--no-ignore or no .gitignore present), the
    // watch loop indexes every supported-extension path. Verifies
    // the `Option<_>` in `WatchConfig.gitignore` behaves as
    // documented.
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    // Default test_watch_config → gitignore is None.
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);

    let mut state = test_watch_state();
    let event = make_event(
        vec![PathBuf::from(
            "/tmp/test_project/.claude/worktrees/agent-x/src/lib.rs",
        )],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );
    collect_events(&event, &cfg, &mut state);
    assert!(
        !state.pending_files.is_empty(),
        "with matcher=None, all supported-ext paths must be accepted"
    );
}

#[test]
fn collect_events_cqs_dir_skip_survives_gitignore_allowlist() {
    // Even if a user accidentally or deliberately adds `!.cqs/` to
    // .gitignore, the hardcoded `.cqs/` skip keeps the system's own
    // files out of the index.
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs", "db"].iter().cloned().collect();
    // Negation allowing .cqs/ — should still be filtered by the
    // hardcoded .cqs/ skip in collect_events.
    let matcher = gitignore_from_lines(&root, &["*.tmp", "!.cqs/"]);
    let cfg = test_watch_config_with_gitignore(&root, &cqs_dir, &notes_path, &supported, matcher);

    let mut state = test_watch_state();
    let event = make_event(
        vec![PathBuf::from("/tmp/test_project/.cqs/index.db")],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );
    collect_events(&event, &cfg, &mut state);
    assert!(
        state.pending_files.is_empty(),
        ".cqs/ must always be skipped (belt-and-suspenders vs gitignore allowlist)"
    );
}

// All `build_gitignore_matcher_*` tests share the `watch_gitignore_env` serial
// group: `build_gitignore_matcher` reads `CQS_WATCH_RESPECT_GITIGNORE`, and the
// kill-switch test sets it to "0" (which forces a `None` matcher). Without
// serialization a parallel sibling observes that "0" mid-window and its
// `.expect("matcher should build")` panics on the unexpected `None`.
#[test]
#[serial(watch_gitignore_env)]
fn build_gitignore_matcher_missing_returns_none() {
    // A project with neither .gitignore nor .cqsignore should produce
    // a `None` matcher — the watch loop indexes everything.
    let tmp = tempfile::TempDir::new().unwrap();
    assert!(
        build_gitignore_matcher(tmp.path()).is_none(),
        "missing .gitignore + .cqsignore should yield None matcher"
    );
}

#[test]
#[serial(watch_gitignore_env)]
fn build_gitignore_matcher_env_kill_switch() {
    // CQS_WATCH_RESPECT_GITIGNORE=0 forces None even if .gitignore exists.
    let tmp = tempfile::TempDir::new().unwrap();
    std::fs::write(tmp.path().join(".gitignore"), "target/\n").unwrap();

    // RAII guard restores the prior value even if the assertion panics, so the
    // "0" never leaks into the next test in the serial group.
    let _guard = EnvVarGuard::set("CQS_WATCH_RESPECT_GITIGNORE", "0");
    let result = build_gitignore_matcher(tmp.path());

    assert!(
        result.is_none(),
        "CQS_WATCH_RESPECT_GITIGNORE=0 must disable the matcher"
    );
}

#[test]
#[serial(watch_gitignore_env)]
fn build_gitignore_matcher_real_file_loads_rules() {
    let tmp = tempfile::TempDir::new().unwrap();
    std::fs::write(
        tmp.path().join(".gitignore"),
        "target/\n.claude/\nnode_modules/\n",
    )
    .unwrap();

    let matcher =
        build_gitignore_matcher(tmp.path()).expect("matcher should build for real gitignore");
    assert!(matcher.num_ignores() >= 3, "expected ≥3 rules loaded");

    // Sanity: matcher returns is_ignore for a target/ path via
    // parent-walk (file inside a directory-ignore rule).
    let hit = matcher
        .matched_path_or_any_parents(tmp.path().join("target/debug/foo.rs"), false)
        .is_ignore();
    assert!(hit, "target/ should match");
}

#[test]
#[serial(watch_gitignore_env)]
fn build_gitignore_matcher_loads_cqsignore() {
    // The watch matcher must layer .cqsignore on top of .gitignore so
    // cqs-specific exclusions (vendor bundles etc.) are respected at
    // event time, mirroring the indexer behaviour.
    let tmp = tempfile::TempDir::new().unwrap();
    std::fs::write(tmp.path().join(".gitignore"), "target/\n").unwrap();
    std::fs::write(tmp.path().join(".cqsignore"), "**/*.min.js\n").unwrap();

    let matcher = build_gitignore_matcher(tmp.path()).expect("matcher should build with cqsignore");
    assert!(matcher.num_ignores() >= 2, "expected rules from both files");

    let vendor_hit = matcher
        .matched_path_or_any_parents(
            tmp.path().join("src/serve/assets/vendor/three.min.js"),
            false,
        )
        .is_ignore();
    assert!(
        vendor_hit,
        ".cqsignore *.min.js rule should match vendor JS"
    );

    let regular_miss = matcher
        .matched_path_or_any_parents(tmp.path().join("src/main.rs"), false)
        .is_ignore();
    assert!(!regular_miss, "regular source files must not match");
}

#[test]
#[serial(watch_gitignore_env)]
fn build_gitignore_matcher_cqsignore_only() {
    // .cqsignore alone (no .gitignore) should still build the matcher.
    let tmp = tempfile::TempDir::new().unwrap();
    std::fs::write(tmp.path().join(".cqsignore"), "secret.txt\n").unwrap();

    let matcher =
        build_gitignore_matcher(tmp.path()).expect("matcher should build with cqsignore alone");
    let hit = matcher
        .matched_path_or_any_parents(tmp.path().join("secret.txt"), false)
        .is_ignore();
    assert!(hit, "cqsignore-only rule should match");
}

// ===== SPLADE builder / batch-size tests =====
//
// Every test in this section reads or writes the process-global
// `CQS_SPLADE_BATCH` env var (`splade_batch_size` / `splade_batch_size_for`
// both consult it). They share the `splade_batch_env` serial group so a
// writer's value never bleeds into a concurrent reader, and each uses an
// `EnvVarGuard` so a panicking assertion restores the prior value instead of
// leaking it to the next test in the group.

#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_env_override() {
    let _guard = EnvVarGuard::set("CQS_SPLADE_BATCH", "16");
    let got = splade_batch_size();
    assert_eq!(got, 16);
}

#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_default_is_32() {
    let _guard = EnvVarGuard::remove("CQS_SPLADE_BATCH");
    let got = splade_batch_size();
    assert_eq!(got, 32);
}

#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_invalid_falls_back_to_default() {
    let _guard = EnvVarGuard::set("CQS_SPLADE_BATCH", "not-a-number");
    let got = splade_batch_size();
    assert_eq!(got, 32, "unparseable value falls back to default");
}

#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_zero_falls_back_to_default() {
    let _guard = EnvVarGuard::set("CQS_SPLADE_BATCH", "0");
    let got = splade_batch_size();
    assert_eq!(got, 32, "0 is not a valid batch size, falls back");
}

// ===== splade_batch_size_for(hidden, max_length) =====
//
// Pin the baseline shape's batch, then verify scaling cases for wider
// hidden / longer seq / compound / env-override.

use super::reindex::splade_batch_size_for;

/// Reference shape (hidden=768, max_length=256, the SPLADE-base /
/// ensembledistil values) reproduces baseline 32.
#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_for_baseline_returns_default() {
    let _guard = EnvVarGuard::remove("CQS_SPLADE_BATCH");
    let got = splade_batch_size_for(768, 256);
    assert_eq!(got, 32, "baseline (768, 256) → 32");
}

/// SPLADE-Code 0.6B class (hidden ≈ 1024, seq=512). hidden_factor =
/// 768/1024 = 0.75; seq_factor = 256/512 = 0.5; scaled = 32 * 0.75 *
/// 0.5 = 12; rounded = 16 (next power of two).
#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_for_splade_code_class_halves() {
    let _guard = EnvVarGuard::remove("CQS_SPLADE_BATCH");
    let got = splade_batch_size_for(1024, 512);
    assert_eq!(got, 16, "1024/512 → 16");
}

/// Wider hidden (1024) at baseline seq. hidden_factor = 0.75; scaled
/// = 32 * 0.75 = 24; rounded = 32. (next_power_of_two(24) = 32.)
#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_for_wider_hidden_at_baseline_seq_holds_32() {
    let _guard = EnvVarGuard::remove("CQS_SPLADE_BATCH");
    let got = splade_batch_size_for(1024, 256);
    assert_eq!(got, 32, "1024/256 → 32 (rounded up)");
}

/// Long seq (1024) at baseline hidden. seq_factor = 256/1024 = 0.25;
/// scaled = 32 * 0.25 = 8.
#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_for_long_seq_quarters_batch() {
    let _guard = EnvVarGuard::remove("CQS_SPLADE_BATCH");
    let got = splade_batch_size_for(768, 1024);
    assert_eq!(got, 8, "768/1024 → 8");
}

/// Env override wins regardless of (hidden, max_length).
#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_for_env_override_wins() {
    let _guard = EnvVarGuard::set("CQS_SPLADE_BATCH", "100");
    let got = splade_batch_size_for(1024, 1024);
    let got2 = splade_batch_size_for(384, 128);
    assert_eq!(got, 100, "env override wins for wide+long shape");
    assert_eq!(got2, 100, "env override wins for narrow+short shape");
}

/// Degenerate dims (zeros) don't panic. The `.max(1)` floors ensure
/// the formula stays bounded.
#[test]
#[serial(splade_batch_env)]
fn splade_batch_size_for_zero_dims_do_not_panic() {
    let _guard = EnvVarGuard::remove("CQS_SPLADE_BATCH");
    let _ = splade_batch_size_for(0, 256);
    let _ = splade_batch_size_for(768, 0);
    let _ = splade_batch_size_for(0, 0);
}

#[test]
#[serial(watch_incremental_splade_env)]
fn build_splade_encoder_env_kill_switch_returns_none() {
    // CQS_WATCH_INCREMENTAL_SPLADE=0 must return None regardless of
    // whether a SPLADE model is configured. Verifies the feature-flag
    // kill-switch fires before any model-load work.
    let _guard = EnvVarGuard::set("CQS_WATCH_INCREMENTAL_SPLADE", "0");
    let got = build_splade_encoder_for_watch();
    assert!(
        got.is_none(),
        "CQS_WATCH_INCREMENTAL_SPLADE=0 must disable the encoder"
    );
}

#[test]
fn splade_origin_key_normalizes_backslashes() {
    // `encode_splade_for_changed_files` builds the DB lookup key via
    // `cqs::normalize_path(file)`. A `PathBuf`
    // carrying backslashes (as any Windows-canonicalized path does)
    // must normalize to the forward-slash form stored at ingest, or
    // `get_chunks_by_origin` returns Ok(vec![]) and SPLADE silently
    // no-ops for the file.
    let p = std::path::PathBuf::from(r"src\cli\watch.rs");
    let origin = cqs::normalize_path(&p);
    assert_eq!(
        origin, "src/cli/watch.rs",
        "origin key must use forward slashes to match DB origins"
    );

    // UNC verbatim prefix must be stripped too (dunce::canonicalize
    // may leave `\\?\C:\…` on Windows). On Unix this just asserts
    // the helper doesn't mangle a plain relative path.
    let p2 = std::path::PathBuf::from(r"\\?\C:\repo\src\cli\watch.rs");
    let origin2 = cqs::normalize_path(&p2);
    assert!(
        !origin2.contains('\\') && !origin2.starts_with(r"\\?\"),
        "normalize_path must strip the verbatim UNC prefix: got {origin2}"
    );
}

#[test]
fn collect_events_detects_notes_path() {
    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path().to_path_buf();
    let cqs_dir = root.join(".cqs");
    let notes_dir = root.join("docs");
    std::fs::create_dir_all(&notes_dir).unwrap();
    let notes_path = notes_dir.join("notes.toml");
    std::fs::write(&notes_path, "# notes").unwrap();

    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();

    let event = make_event(
        vec![notes_path.clone()],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );

    collect_events(&event, &cfg, &mut state);

    assert!(state.pending_notes, "Notes path should set pending_notes");
    assert!(
        state.pending_files.is_empty(),
        "Notes should not be added to pending_files"
    );
}

#[test]
fn collect_events_respects_max_pending_files() {
    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path().to_path_buf();
    let cqs_dir = root.join(".cqs");
    let notes_path = root.join("docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();

    // Pre-fill pending_files to max_pending_files()
    for i in 0..max_pending_files() {
        state
            .pending_files
            .insert(PathBuf::from(format!("f{}.rs", i)));
    }

    // Create a real file so mtime check passes
    let new_file = root.join("overflow.rs");
    std::fs::write(&new_file, "fn main() {}").unwrap();

    let event = make_event(
        vec![new_file],
        EventKind::Create(notify::event::CreateKind::File),
    );

    collect_events(&event, &cfg, &mut state);

    assert_eq!(
        state.pending_files.len(),
        max_pending_files(),
        "Should not exceed max_pending_files()"
    );
}

#[test]
fn collect_events_skips_unchanged_mtime() {
    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path().to_path_buf();
    let cqs_dir = root.join(".cqs");
    let notes_path = root.join("docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();

    // Create a file and record its mtime as already indexed
    let file = root.join("src/lib.rs");
    std::fs::create_dir_all(root.join("src")).unwrap();
    std::fs::write(&file, "fn main() {}").unwrap();
    let mtime = std::fs::metadata(&file).unwrap().modified().unwrap();
    state
        .last_indexed_mtime
        .insert(PathBuf::from("src/lib.rs"), mtime);

    let event = make_event(
        vec![file],
        EventKind::Modify(notify::event::ModifyKind::Data(
            notify::event::DataChange::Content,
        )),
    );

    collect_events(&event, &cfg, &mut state);

    assert!(
        state.pending_files.is_empty(),
        "Unchanged mtime should be skipped"
    );
}

// ===== last_indexed_mtime prune tests =====

/// Recency prune drops entries older than `LAST_INDEXED_PRUNE_AGE_SECS`,
/// keeps fresh entries, and only triggers once the map exceeds
/// `LAST_INDEXED_PRUNE_SIZE_THRESHOLD_DEFAULT`.
#[test]
fn test_last_indexed_mtime_recency_prune() {
    let now = SystemTime::now();
    let two_days = Duration::from_secs(2 * LAST_INDEXED_PRUNE_AGE_SECS);
    let one_minute = Duration::from_secs(60);
    let old = now.checked_sub(two_days).unwrap();
    let fresh = now.checked_sub(one_minute).unwrap();

    // (1) Small map — below the size threshold — must not prune at all,
    // even if every entry is ancient. The threshold is a cache-size
    // safety valve, not a TTL for the whole session.
    let mut small: HashMap<PathBuf, SystemTime> = HashMap::new();
    small.insert(PathBuf::from("a.rs"), old);
    small.insert(PathBuf::from("b.rs"), fresh);
    let pruned_small = prune_last_indexed_mtime(&mut small);
    assert_eq!(
        pruned_small, 0,
        "Prune must not run below size threshold (got {} entries removed from 2-entry map)",
        pruned_small
    );
    assert_eq!(
        small.len(),
        2,
        "Small map should be untouched when below threshold"
    );

    // (2) Large map — above the size threshold — prunes old entries
    // and keeps fresh ones. Build a map with SIZE_THRESHOLD + 1 old
    // entries plus a handful of fresh sentinels so we can check both
    // that old entries are removed and fresh ones survive.
    let mut large: HashMap<PathBuf, SystemTime> = HashMap::new();
    for i in 0..=LAST_INDEXED_PRUNE_SIZE_THRESHOLD_DEFAULT {
        large.insert(PathBuf::from(format!("old_{}.rs", i)), old);
    }
    large.insert(PathBuf::from("fresh_1.rs"), fresh);
    large.insert(PathBuf::from("fresh_2.rs"), now);
    let total_before = large.len();
    let pruned_large = prune_last_indexed_mtime(&mut large);

    // Every "old" entry (two days stale) should be gone.
    assert_eq!(
        pruned_large,
        LAST_INDEXED_PRUNE_SIZE_THRESHOLD_DEFAULT + 1,
        "Expected all old entries pruned (total_before={}, remaining={})",
        total_before,
        large.len()
    );
    assert!(
        large.contains_key(&PathBuf::from("fresh_1.rs")),
        "Fresh entry from 1 minute ago must survive prune"
    );
    assert!(
        large.contains_key(&PathBuf::from("fresh_2.rs")),
        "Entry at `now` must survive prune"
    );
    assert_eq!(
        large.len(),
        2,
        "Only the 2 fresh entries should remain after prune"
    );

    // (3) Entry just inside the cutoff window survives. We use a 1-second
    // margin rather than exactly `now - PRUNE_AGE` because `prune_*` calls
    // `SystemTime::now()` internally — its clock ticks a few microseconds
    // past the test's clock, so an entry pinned to the test's computed
    // cutoff would be classified as older and pruned. 1 second is
    // comfortably more than the inter-call drift while still well under
    // the 1-day window.
    let just_inside = now
        .checked_sub(Duration::from_secs(LAST_INDEXED_PRUNE_AGE_SECS - 1))
        .unwrap();
    let mut boundary: HashMap<PathBuf, SystemTime> = HashMap::new();
    for i in 0..=LAST_INDEXED_PRUNE_SIZE_THRESHOLD_DEFAULT {
        boundary.insert(PathBuf::from(format!("stale_{}.rs", i)), old);
    }
    boundary.insert(PathBuf::from("just_inside.rs"), just_inside);
    prune_last_indexed_mtime(&mut boundary);
    assert!(
        boundary.contains_key(&PathBuf::from("just_inside.rs")),
        "Entry 1 second inside the cutoff window must survive"
    );
}

#[test]
fn prune_last_indexed_mtime_by_age_ignores_size_threshold() {
    // Age-only prune fires regardless of map size, so a daemon below the
    // threshold still ages out stale entries.
    let now = SystemTime::now();
    let two_days = Duration::from_secs(LAST_INDEXED_PRUNE_AGE_SECS + 86_400);
    let one_minute = Duration::from_secs(60);
    let old = now.checked_sub(two_days).unwrap();
    let fresh = now.checked_sub(one_minute).unwrap();

    let mut tiny: HashMap<PathBuf, SystemTime> = HashMap::new();
    tiny.insert(PathBuf::from("old_a.rs"), old);
    tiny.insert(PathBuf::from("old_b.rs"), old);
    tiny.insert(PathBuf::from("fresh.rs"), fresh);
    // Size = 3, well below LAST_INDEXED_PRUNE_SIZE_THRESHOLD_DEFAULT (5_000).
    // Size-gated `prune_last_indexed_mtime` would be a no-op here; the
    // age-only variant still drops the two stale entries.
    let pruned = super::gc::prune_last_indexed_mtime_by_age(&mut tiny);
    assert_eq!(pruned, 2, "age-only prune must drop entries past cutoff");
    assert_eq!(tiny.len(), 1, "fresh entry survives");
    assert!(tiny.contains_key(&PathBuf::from("fresh.rs")));
}

// ===== Constants tests =====

#[test]
fn hnsw_rebuild_threshold_is_reasonable() {
    assert!(hnsw_rebuild_threshold() > 0);
    assert!(hnsw_rebuild_threshold() <= 1000);
}

#[test]
fn max_pending_files_is_bounded() {
    assert!(max_pending_files() > 0);
    assert!(max_pending_files() <= 100_000);
}

// ===== trim_trailing_newline tests =====

#[cfg(unix)]
#[test]
fn trim_newline_strips_lf() {
    assert_eq!(socket::trim_trailing_newline(b"hello\n"), b"hello");
}

#[cfg(unix)]
#[test]
fn trim_newline_strips_crlf() {
    assert_eq!(socket::trim_trailing_newline(b"hello\r\n"), b"hello");
}

#[cfg(unix)]
#[test]
fn trim_newline_no_op_when_absent() {
    assert_eq!(socket::trim_trailing_newline(b"hello"), b"hello");
}

#[cfg(unix)]
#[test]
fn trim_newline_handles_empty() {
    assert_eq!(socket::trim_trailing_newline(b""), b"");
}

#[cfg(unix)]
#[test]
fn trim_newline_only_strips_one_lf() {
    // Two trailing newlines → only the last is stripped (callers that
    // wrote two newlines deliberately are uncommon, but we don't want
    // to silently consume more than one terminator).
    assert_eq!(socket::trim_trailing_newline(b"hello\n\n"), b"hello\n");
}

// ===== chunk.id prefix-strip uses normalize_path =====

/// Exercises the same strip-and-rewrite shape used by `reindex_files`. The
/// direct function isn't extracted, but the logic is small and identical —
/// this test documents the contract so a regression back to
/// `abs_path.display()` is caught by a targeted unit test instead of the
/// next Windows CI run.
fn normalize_strip_and_rewrite(abs_path: &Path, rel_path: &Path, chunk_id: &str) -> Option<String> {
    let abs_norm = cqs::normalize_path(abs_path);
    let rel_norm = cqs::normalize_path(rel_path);
    chunk_id
        .strip_prefix(abs_norm.as_str())
        .map(|rest| format!("{}{}", rel_norm, rest))
}

#[test]
fn prefix_strip_normalizes_backslash_verbatim_prefix() {
    // Simulates the Windows shape that the bug regressed on:
    //   abs_path   = \\?\C:\Projects\cqs\src\foo.rs
    //   chunk.id   = C:/Projects/cqs/src/foo.rs:10:abcd  (parser output)
    //   rel_path   = src\foo.rs  (after strip_prefix on the root)
    // Before the fix: `abs_path.display()` emits the verbatim `\\?\` +
    // backslashes, so the prefix-strip fails and chunk.id keeps its
    // absolute prefix. After the fix: both sides normalize.
    let abs = Path::new(r"\\?\C:\Projects\cqs\src\foo.rs");
    let rel = Path::new(r"src\foo.rs");
    let chunk_id = "C:/Projects/cqs/src/foo.rs:10:abcd";
    let rewritten =
        normalize_strip_and_rewrite(abs, rel, chunk_id).expect("prefix-strip must match");
    assert!(
        rewritten.starts_with("src/foo.rs"),
        "expected rewritten id to start with forward-slash rel path, got {rewritten}"
    );
    assert_eq!(rewritten, "src/foo.rs:10:abcd");
}

#[test]
fn prefix_strip_unix_path_round_trip() {
    // Baseline: Unix path with forward slashes on both sides still works.
    let abs = Path::new("/home/user/proj/src/foo.rs");
    let rel = Path::new("src/foo.rs");
    let chunk_id = "/home/user/proj/src/foo.rs:42:deadbeef";
    let rewritten =
        normalize_strip_and_rewrite(abs, rel, chunk_id).expect("prefix-strip must match");
    assert_eq!(rewritten, "src/foo.rs:42:deadbeef");
}

// ===== gitignore RwLock poison recovery =====

#[test]
fn gitignore_rwlock_poison_still_yields_matcher() {
    // A writer that panics while holding the write lock leaves the inner
    // value valid but the lock poisoned; the `match gitignore.read()` arm
    // must recover via `poisoned.into_inner()` instead of silently dropping
    // to "no matcher".
    use std::sync::{Arc, RwLock};

    let matcher_builder = ignore::gitignore::GitignoreBuilder::new(std::path::Path::new("."));
    let (matcher, _errs) = matcher_builder.build_global();
    let lock: Arc<RwLock<Option<ignore::gitignore::Gitignore>>> =
        Arc::new(RwLock::new(Some(matcher)));

    // Poison the lock by panicking inside a write guard on a helper
    // thread — the panic propagates, leaves the RwLock poisoned, and
    // joins.
    let poisoner = Arc::clone(&lock);
    let _ = std::thread::spawn(move || {
        let _guard = poisoner.write().expect("initial write must succeed");
        panic!("intentional poison for EH-V1.29-8 test");
    })
    .join();

    // Post-poison: the read must still yield `Some(_)` by recovering the
    // inner value via `into_inner()`, not silently return `None`.
    let matcher_guard = match lock.read() {
        Ok(g) => Some(g),
        Err(poisoned) => Some(poisoned.into_inner()),
    };
    assert!(
        matcher_guard.is_some(),
        "poison-recovery must still surface the previously-written matcher"
    );
    assert!(
        matcher_guard.as_ref().unwrap().is_some(),
        "inner Option<Gitignore> must still be Some after poison recovery"
    );
}

// ── background rebuild + atomic swap ──────────────────────────────

/// Build a tiny `Owned` HnswIndex from N synthetic vectors. Stand-in for a
/// thread-built index in the `drain_pending_rebuild` tests below.
fn synthetic_owned_index(n: usize, dim: usize) -> cqs::hnsw::HnswIndex {
    // Non-zero, distinct vectors per id — hnsw_rs's HNSW can collapse
    // zero vectors (undefined cosine sim) so the first entry needs a
    // non-trivial value or the index ends up under-populated.
    let batch: Vec<(String, cqs::Embedding)> = (0..n)
        .map(|i| {
            let mut v = vec![0.1_f32; dim];
            v[i % dim] = (i as f32 + 1.0) * 0.5;
            (format!("c{i}"), cqs::Embedding::new(v))
        })
        .collect();
    let iter = std::iter::once(Ok::<_, cqs::store::StoreError>(batch));
    cqs::hnsw::HnswIndex::build_batched_with_dim(iter, n, dim).expect("build synthetic index")
}

/// Make a Store + WatchConfig pair for a fresh tempdir, init'd to `dim`.
/// Returns owned bindings so each caller can pass long-lived references
/// to `test_watch_config`.
struct DrainFixture {
    tmp: tempfile::TempDir,
    store: Store,
    supported_ext: HashSet<&'static str>,
    notes_path: PathBuf,
}

fn drain_test_fixture(dim: usize) -> DrainFixture {
    let tmp = tempfile::TempDir::new().unwrap();
    let store_path = tmp.path().join("index.db");
    let mut store = Store::open(&store_path).unwrap();
    store
        .init(&cqs::store::ModelInfo::new("test/m", dim))
        .unwrap();
    store.set_dim(dim);
    let notes_path = tmp.path().join("docs/notes.toml");
    DrainFixture {
        tmp,
        store,
        supported_ext: HashSet::new(),
        notes_path,
    }
}

#[test]
fn drain_pending_rebuild_replays_delta_into_new_index() {
    let dim = 4;
    let new_idx = synthetic_owned_index(3, dim);
    assert_eq!(new_idx.len(), 3);

    let (tx, rx) = std::sync::mpsc::channel();
    tx.send(Ok(Some(RebuildResult {
        index: new_idx,
        // No overlap between delta ids and snapshot — all replay.
        // HashSet<u64> of (id, hash) fingerprints.
        snapshot_keys: std::collections::HashSet::new(),
    })))
    .unwrap();
    drop(tx);

    let mut state = test_watch_state();
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: vec![
            (
                "delta_a".to_string(),
                cqs::Embedding::new(vec![1.0; dim]),
                "h_delta_a".to_string(),
            ),
            (
                "delta_b".to_string(),
                cqs::Embedding::new(vec![0.5; dim]),
                "h_delta_b".to_string(),
            ),
        ],
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(dim);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );
    let store = &fix.store;

    drain_pending_rebuild(&cfg, store, &mut state);

    let idx = state.hnsw_index.expect("rebuild was swapped in");
    assert_eq!(idx.len(), 5, "3 from new_idx + 2 from delta");
    // id_map is `Vec<Box<str>>`; comparing `Box<str>` against `&str`
    // requires deref both sides via `&**id == "..."`.
    assert!(idx.ids().iter().any(|id| &**id == "delta_a"));
    assert!(idx.ids().iter().any(|id| &**id == "delta_b"));
    assert_eq!(state.incremental_count, 0);
    assert!(state.pending_rebuild.is_none());
}

/// Ordering contract for same-cycle pairs: chunks committed in the same
/// cycle a rebuild completes are queued into the pending delta BEFORE the
/// drain runs (`PendingRebuild::queue_delta`, called by
/// `process_file_changes` ahead of `drain_pending_rebuild`). The drain's
/// sidecar save is stamped with the loop's observed store state — which
/// includes those chunks — and clears the dirty flag, so the saved index
/// must contain them: a crash right after the drain would otherwise leave
/// a stamped-current sidecar that is missing the cycle's chunks, and the
/// dirty-path self-heal would trust it.
#[test]
fn same_cycle_pairs_queued_before_drain_land_in_drained_and_saved_index() {
    // dim = 64 (not the usual 4): the final assertion round-trips the saved
    // sidecar through `load_with_dim`, whose data-file size sanity cap is
    // `len * dim * 4 * 2` — at dim=4 that (128 B) is below the hnsw_rs file
    // header overhead and the load is rejected as oversized.
    let dim = 64;
    let new_idx = synthetic_owned_index(3, dim); // ids: c0, c1, c2

    let (tx, rx) = std::sync::mpsc::channel();
    tx.send(Ok(Some(RebuildResult {
        index: new_idx,
        snapshot_keys: std::collections::HashSet::new(),
    })))
    .unwrap();
    drop(tx);

    let mut state = test_watch_state();
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: Vec::new(),
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(dim);
    fix.store
        .set_hnsw_dirty(cqs::HnswKind::Enriched, true)
        .unwrap();
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );

    // The same-cycle pair: committed this cycle, rebuild thread's snapshot
    // never saw it. Queued BEFORE the drain — the seam under test.
    let pending = state.pending_rebuild.as_mut().expect("pending set above");
    pending.queue_delta(
        vec![(
            "same_cycle".to_string(),
            cqs::Embedding::new(vec![0.9; dim]),
            "h_same_cycle".to_string(),
        )],
        dim,
        true,
    );

    let outcome = drain_pending_rebuild(&cfg, &fix.store, &mut state);
    assert_eq!(outcome, DrainOutcome::Swapped);

    // In-memory index contains the same-cycle pair.
    let idx = state.hnsw_index.expect("rebuild was swapped in");
    assert!(idx.ids().iter().any(|id| &**id == "same_cycle"));

    // The stamped on-disk save contains it too — and the dirty flag was
    // cleared, so this sidecar is exactly what a post-crash self-heal
    // would trust.
    assert!(
        !fix.store.is_hnsw_dirty(cqs::HnswKind::Enriched).unwrap(),
        "drain save landed → dirty flag cleared"
    );
    let loaded = cqs::HnswIndex::load_with_dim(fix.tmp.path(), "index", dim)
        .expect("drain must have saved a loadable index");
    assert!(
        loaded.ids().iter().any(|id| &**id == "same_cycle"),
        "the saved sidecar must contain the same-cycle pair its stamp claims"
    );
}

/// Lost-update guard: when the live store stamp no longer matches the stamp
/// the watch loop last observed (a concurrent `cqs index` committed chunks
/// the rebuild snapshot + delta never saw), the drain must discard the
/// rebuilt index — no disk save, no in-memory swap, dirty flag untouched.
/// Swapping or saving would serve an index that silently lacks the foreign
/// writer's chunks while the cleared flag claims it's current.
#[test]
fn drain_pending_rebuild_discards_when_store_moved_past_observed_stamp() {
    let dim = 4;
    let new_idx = synthetic_owned_index(3, dim);

    let (tx, rx) = std::sync::mpsc::channel();
    tx.send(Ok(Some(RebuildResult {
        index: new_idx,
        snapshot_keys: std::collections::HashSet::new(),
    })))
    .unwrap();
    drop(tx);

    let mut state = test_watch_state();
    state.incremental_count = 7;
    // The seam: the watch loop's observed stamp is from a store state the
    // live store has moved past (fixture store is fresh: count 0, gen 0 —
    // any non-zero observed stamp models "store changed under us" with the
    // same inequality the real interleaving produces).
    state.observed_stamp = Some(cqs::hnsw::StoreStamp {
        chunk_count: 999,
        splade_generation: 999,
    });
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: Vec::new(),
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(dim);
    fix.store
        .set_hnsw_dirty(cqs::HnswKind::Enriched, true)
        .unwrap();
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );

    drain_pending_rebuild(&cfg, &fix.store, &mut state);

    assert!(
        state.hnsw_index.is_none(),
        "stale rebuild must not be swapped into memory"
    );
    assert!(
        state.pending_rebuild.is_none(),
        "pending must be cleared so the next threshold rebuild can retry"
    );
    assert_eq!(
        state.incremental_count, 7,
        "discard must not reset the incremental counter"
    );
    assert!(
        !cqs::HnswIndex::exists(fix.tmp.path(), "index"),
        "discarded rebuild must not write sidecars"
    );
    assert!(
        fix.store.is_hnsw_dirty(cqs::HnswKind::Enriched).unwrap(),
        "dirty flag must stay set after a discarded rebuild save"
    );
}

/// When a chunk is re-embedded mid-rebuild, the snapshot has the OLD vector
/// under the same id while delta has the NEW vector + new content_hash. The
/// drain must REPLAY the delta entry so the fresh embedding lands in the
/// swapped HNSW. Dedup is (id, content_hash)-aware, so a same-id-new-hash
/// entry is not dropped.
///
/// We can't query hnsw_rs for "give me the embedding stored under id X"
/// (it's a graph, not a kv store) and there's no deletion API, so we
/// assert the side-effect: the swapped index contains MORE entries
/// than the snapshot alone (orphan + replayed vector both present),
/// and a search by the FRESH embedding returns id "a" with cosine ≈ 1.0.
#[test]
fn test_rebuild_window_re_embedding_replays_fresh_vector() {
    let dim = 4;

    // Snapshot has id "a" baked in with hash h_v1 (and an unrelated id "z"
    // so the index isn't trivially empty).
    let snapshot_batch: Vec<(String, cqs::Embedding)> = vec![
        (
            "a".to_string(),
            cqs::Embedding::new(vec![1.0, 0.0, 0.0, 0.0]),
        ),
        (
            "z".to_string(),
            cqs::Embedding::new(vec![0.0, 0.0, 0.0, 1.0]),
        ),
    ];
    let snapshot_iter = std::iter::once(Ok::<_, cqs::store::StoreError>(snapshot_batch));
    let new_idx = cqs::hnsw::HnswIndex::build_batched_with_dim(snapshot_iter, 2, dim)
        .expect("build snapshot index");
    assert_eq!(new_idx.len(), 2, "snapshot starts with 2 entries");

    // HashSet<u64> of (id, hash) fingerprints.
    let mut snapshot_keys = std::collections::HashSet::new();
    snapshot_keys.insert(crate::cli::commands::snapshot_fingerprint("a", "h_v1"));
    snapshot_keys.insert(crate::cli::commands::snapshot_fingerprint("z", "h_z"));

    let (tx, rx) = std::sync::mpsc::channel();
    tx.send(Ok(Some(RebuildResult {
        index: new_idx,
        snapshot_keys,
    })))
    .unwrap();
    drop(tx);

    // Delta has "a" again, but with a NEW embedding and a NEW content_hash —
    // i.e. the file was re-embedded between the snapshot and the swap.
    // The fresh vector points along axis 1, distinct from the snapshot's
    // axis-0 vector, so we can tell them apart by search.
    let fresh_embedding = cqs::Embedding::new(vec![0.0, 1.0, 0.0, 0.0]);

    let mut state = test_watch_state();
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: vec![(
            "a".to_string(),
            fresh_embedding.clone(),
            "h_v2".to_string(), // hash differs from snapshot's "h_v1"
        )],
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(dim);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );

    drain_pending_rebuild(&cfg, &fix.store, &mut state);

    let idx = state.hnsw_index.expect("rebuild was swapped in");
    // The fresh vector was REPLAYED — index now contains 3 nodes
    // (snapshot's "a" + "z" + replayed "a"). hnsw_rs has no deletion,
    // so both vectors for "a" coexist as duplicate-id orphans; that's
    // the same trade-off as the fast-incremental path. Search
    // post-filters via SQLite in production, which collapses the
    // duplicates into one logical hit.
    assert_eq!(
        idx.len(),
        3,
        "fresh re-embedding must be replayed (snapshot 2 + 1 replay)"
    );

    // Crucial assertion: searching by the FRESH embedding returns id "a"
    // with cosine ≈ 1.0 — the axis-1 vector is in the index under "a"
    // because the replay landed it there.
    let hits = idx.search(&fresh_embedding, 1);
    assert!(!hits.is_empty(), "search must return at least one hit");
    let top = &hits[0];
    assert_eq!(
        top.id, "a",
        "top hit for fresh embedding must be the re-embedded chunk \"a\""
    );
    assert!(
        top.score > 0.99,
        "top hit cosine must be near 1.0 (fresh vector is in the index); got {}",
        top.score
    );

    assert!(state.pending_rebuild.is_none());
}

#[test]
fn drain_pending_rebuild_dedups_against_known_ids() {
    // Dedup is (id, content_hash)-aware. The rebuild thread snapshotted
    // c0/c1/c2 with hashes h0/h1/h2. Delta replays c0 with the SAME hash h0
    // (true duplicate — must be skipped), c1 with the same hash h1 (skipped),
    // and c_new with a brand-new id (must replay). The dedup uses the
    // snapshot hashes the rebuild produced.
    let dim = 4;
    let new_idx = synthetic_owned_index(3, dim); // ids: c0, c1, c2

    // HashSet<u64> of (id, hash) fingerprints.
    let mut snapshot_keys = std::collections::HashSet::new();
    snapshot_keys.insert(crate::cli::commands::snapshot_fingerprint("c0", "h0"));
    snapshot_keys.insert(crate::cli::commands::snapshot_fingerprint("c1", "h1"));
    snapshot_keys.insert(crate::cli::commands::snapshot_fingerprint("c2", "h2"));

    let (tx, rx) = std::sync::mpsc::channel();
    tx.send(Ok(Some(RebuildResult {
        index: new_idx,
        snapshot_keys,
    })))
    .unwrap();
    drop(tx);

    let mut state = test_watch_state();
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: vec![
            // Same id + same hash → genuine duplicate, skip.
            (
                "c0".to_string(),
                cqs::Embedding::new(vec![9.0; dim]),
                "h0".to_string(),
            ),
            (
                "c1".to_string(),
                cqs::Embedding::new(vec![9.0; dim]),
                "h1".to_string(),
            ),
            // Brand-new id → snapshot didn't see it, must replay.
            (
                "c_new".to_string(),
                cqs::Embedding::new(vec![9.0; dim]),
                "h_new".to_string(),
            ),
        ],
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(dim);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );
    let store = &fix.store;

    drain_pending_rebuild(&cfg, store, &mut state);

    let idx = state.hnsw_index.expect("rebuild was swapped in");
    assert_eq!(
        idx.len(),
        4,
        "3 from new_idx + 1 genuinely-new delta entry — same-hash duplicates skipped"
    );
    // id_map is `Vec<Box<str>>`; deref both sides.
    assert!(idx.ids().iter().any(|id| &**id == "c_new"));
}

#[test]
fn drain_pending_rebuild_clears_pending_on_thread_error() {
    let (tx, rx) = std::sync::mpsc::channel();
    tx.send(Err(anyhow::anyhow!("simulated rebuild failure")))
        .unwrap();
    drop(tx);

    let mut state = test_watch_state();
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: Vec::new(),
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(4);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );
    let store = &fix.store;

    drain_pending_rebuild(&cfg, store, &mut state);
    assert!(state.pending_rebuild.is_none());
    assert!(state.hnsw_index.is_none());
}

// spawn_hnsw_rebuild adversarial coverage for the dim-mismatch and
// store-open-fail paths — both realistic failure modes (model-swap
// mid-flight, slot dir deleted under the daemon).
//
// We invoke `spawn_hnsw_rebuild` directly, then join the worker thread
// and inspect what landed on the receive channel. The contract is:
//   - dim mismatch  → channel carries Err, pending must clear on drain
//   - missing index → channel carries Err, ditto
// Both paths must NOT panic and must NOT leak the pending entry forever.

/// A dim mismatch between the store and the caller's `expected_dim` must
/// surface as `Err` on the channel, not a panic. The on-disk store is
/// dim=4; we ask for dim=8.
#[test]
fn spawn_hnsw_rebuild_dim_mismatch_returns_error_outcome() {
    let dim = 4;
    let expected_dim = 8;
    let fix = drain_test_fixture(dim);
    let cqs_dir = fix.tmp.path().to_path_buf();
    let index_path = fix.tmp.path().join("index.db");

    let pending = spawn_hnsw_rebuild(cqs_dir, index_path, expected_dim, "p2_29_dim");
    // Wait for the worker thread to finish, bounded.
    let outcome = pending
        .rx
        .recv_timeout(std::time::Duration::from_secs(10))
        .expect("rebuild thread must report within 10s");
    // `RebuildOutcome` is `Result<Option<HnswIndex>, anyhow::Error>` and
    // `HnswIndex` is not `Debug`, so we can't call `unwrap_err` directly.
    // Pattern-match instead.
    let err = match outcome {
        Ok(_) => panic!("dim mismatch must surface as an Err on the rebuild channel"),
        Err(e) => e,
    };
    let msg = format!("{}", err);
    assert!(
        msg.contains("does not match expected") || msg.contains("dim"),
        "error must mention the dim mismatch (got: {msg})"
    );
    // Drain the worker handle so the OS thread is reaped.
    if let Some(h) = pending.handle {
        let _ = h.join();
    }
}

/// Pointing at a non-existent index path (e.g. slot dir removed mid-flight)
/// must surface as `Err` on the channel — never panic, never hang.
/// `Store::open_readonly_pooled` returns an Err immediately and the closure
/// propagates it via `?`.
#[test]
fn spawn_hnsw_rebuild_missing_index_path_returns_error_outcome() {
    let tmp = tempfile::TempDir::new().unwrap();
    let cqs_dir = tmp.path().to_path_buf();
    let bogus = tmp.path().join("does_not_exist.db");

    let pending = spawn_hnsw_rebuild(cqs_dir, bogus, 4, "p2_29_missing");
    let outcome = pending
        .rx
        .recv_timeout(std::time::Duration::from_secs(10))
        .expect("rebuild thread must report within 10s");
    assert!(
        outcome.is_err(),
        "missing index must surface as an Err on the rebuild channel"
    );
    if let Some(h) = pending.handle {
        let _ = h.join();
    }
}

/// Drain path must clear `pending_rebuild` when the worker thread reported
/// an error. The rebuild thread can fail for many reasons (dim mismatch,
/// store gone, save failure); the drain must always reset state so the next
/// threshold trigger can retry — otherwise the pending slot leaks forever
/// and no further rebuilds run.
#[test]
fn drain_clears_pending_when_spawned_rebuild_errors() {
    // Drive the full spawn+drain cycle through a guaranteed-failing
    // path (missing index) so the drain sees a real Err rather than
    // a hand-crafted `tx.send(Err(_))`.
    let tmp = tempfile::TempDir::new().unwrap();
    let pending = spawn_hnsw_rebuild(
        tmp.path().to_path_buf(),
        tmp.path().join("nope.db"),
        4,
        "p2_29_drain",
    );

    // Block until the worker thread has signalled — the drain uses
    // try_recv so we want the message already enqueued.
    if let Some(h) = pending.handle.as_ref() {
        // Best-effort wait: rebuild thread writes to channel before
        // exiting. Up to 10s tolerance for slow CI.
        for _ in 0..100 {
            if h.is_finished() {
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(100));
        }
    }

    // Now build a state with this PendingRebuild and drive the drain.
    let mut state = test_watch_state();
    state.pending_rebuild = Some(pending);

    let fix = drain_test_fixture(4);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );
    drain_pending_rebuild(&cfg, &fix.store, &mut state);
    assert!(
        state.pending_rebuild.is_none(),
        "drain must clear pending_rebuild when the rebuild thread errored"
    );
    assert!(
        state.hnsw_index.is_none(),
        "no index must be swapped in when the rebuild errored"
    );
}

#[test]
fn drain_pending_rebuild_leaves_pending_when_still_running() {
    let (_tx, rx) = std::sync::mpsc::channel::<RebuildOutcome>();
    let mut state = test_watch_state();
    state.pending_rebuild = Some(PendingRebuild {
        rx,
        delta: Vec::new(),
        started_at: std::time::Instant::now(),
        handle: None,
        delta_saturated: false,
        shutdown: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
    });

    let fix = drain_test_fixture(4);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );
    let store = &fix.store;

    drain_pending_rebuild(&cfg, store, &mut state);
    assert!(
        state.pending_rebuild.is_some(),
        "pending should remain in flight when channel has no message"
    );
}

// ── reindex_files consults the global EmbeddingCache ─────────────

/// `reindex_files` must read from `global_cache` before calling the
/// embedder. We prime the cache with a known embedding for the chunk's
/// content_hash, then ensure the chunk written to the store has THAT
/// vector — proof the embedder was bypassed entirely.
///
/// `#[ignore]` because building a real `Embedder` (CPU) loads ONNX
/// weights and is too heavy for the default test pass. The test still
/// exercises the cache wiring; running it gated catches the regression
/// when the watch path drops the cache check.
#[test]
#[ignore = "Requires loading the BGE-large model (heavy)"]
fn test_reindex_files_hits_global_cache_skipping_embedder() {
    use cqs::cache::{CachePurpose, EmbeddingCache};
    use cqs::embedder::ModelConfig;
    use std::io::Write;

    // 1) Tempdir with a tiny rust file we can parse.
    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path();
    let cqs_dir = root.join(".cqs");
    std::fs::create_dir_all(&cqs_dir).unwrap();
    let rs_file = root.join("hit.rs");
    let source = "pub fn hits_cache() { let _ = 42; }";
    let mut f = std::fs::File::create(&rs_file).unwrap();
    f.write_all(source.as_bytes()).unwrap();
    drop(f);

    // 2) Build a Store and an Embedder. Both required by reindex_files.
    let model_cfg = ModelConfig::resolve(None, None);
    let embedder = Embedder::new_cpu(model_cfg).expect("init CPU embedder");
    let dim = embedder.embedding_dim();
    let store_path = cqs_dir.join(cqs::INDEX_DB_FILENAME);
    let mut store = Store::open(&store_path).unwrap();
    store
        .init(&cqs::store::ModelInfo::new(
            &embedder.model_config().repo,
            dim,
        ))
        .unwrap();
    store.set_dim(dim);

    // 3) Parse the file once to learn the chunk's content_hash. Only
    //    deterministic way to know what to prime — the parser's hash
    //    is computed from chunk metadata + bytes.
    let parser = CqParser::new().unwrap();
    let chunks = parser
        .parse_file_all_with_chunk_calls(&rs_file)
        .map(|(c, _, _, _, _)| c)
        .expect("parse hit.rs");
    assert!(!chunks.is_empty(), "parser must yield at least one chunk");
    let target_hash = chunks[0].content_hash.clone();

    // 4) Prime the global cache with a SENTINEL embedding for the
    //    chunk's content_hash. Sentinel = first lane large, others zero,
    //    then unit-normalized — distinguishes it from anything the
    //    embedder would produce on this content.
    let cache_path = EmbeddingCache::project_default_path(&cqs_dir);
    let cache = EmbeddingCache::open(&cache_path).expect("open cache");
    let mut sentinel = vec![0.0_f32; dim];
    sentinel[0] = 7.7;
    let norm: f32 = sentinel.iter().map(|x| x * x).sum::<f32>().sqrt();
    for x in &mut sentinel {
        *x /= norm;
    }
    let sentinel_clone = sentinel.clone();
    cache
        .write_batch_owned(
            &[(target_hash.clone(), sentinel_clone)],
            &embedder.model_fingerprint(),
            CachePurpose::Embedding,
            dim,
        )
        .unwrap();

    // 5) Run reindex_files with the cache wired in.
    let files = vec![PathBuf::from("hit.rs")];
    let (count, _) = reindex_files(root, &store, &files, &parser, &embedder, Some(&cache), true)
        .expect("reindex_files");
    assert!(count >= 1, "at least one chunk indexed");

    // 6) The chunk in the store must hold the SENTINEL — proof that
    //    the global cache served the read instead of the embedder.
    let stored = store
        .get_embeddings_by_hashes(&[target_hash.as_str()])
        .expect("store lookup");
    let stored_emb = stored
        .get(&target_hash)
        .expect("chunk written under the same content_hash");
    let stored_slice = stored_emb.as_slice();
    assert_eq!(stored_slice.len(), dim);
    for (i, (&got, &want)) in stored_slice.iter().zip(sentinel.iter()).enumerate() {
        assert!(
            (got - want).abs() < 1e-5,
            "lane {i}: got {got} want {want} — embedder was called instead of cache hit"
        );
    }
}

/// `reindex_files` with `global_cache: None` uses the store-only path.
/// Lighter assertion: just confirm the function runs to completion and
/// writes chunks. Pins the degrade path so `CQS_CACHE_ENABLED=0` doesn't
/// break watch.
#[test]
#[ignore = "Requires loading the BGE-large model (heavy)"]
fn test_reindex_files_no_global_cache_still_works() {
    use cqs::embedder::ModelConfig;
    use std::io::Write;

    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path();
    let cqs_dir = root.join(".cqs");
    std::fs::create_dir_all(&cqs_dir).unwrap();
    let rs_file = root.join("nocache.rs");
    let mut f = std::fs::File::create(&rs_file).unwrap();
    f.write_all(b"pub fn no_cache_path() { let _ = 0; }")
        .unwrap();
    drop(f);

    let model_cfg = ModelConfig::resolve(None, None);
    let embedder = Embedder::new_cpu(model_cfg).expect("init CPU embedder");
    let dim = embedder.embedding_dim();
    let store_path = cqs_dir.join(cqs::INDEX_DB_FILENAME);
    let mut store = Store::open(&store_path).unwrap();
    store
        .init(&cqs::store::ModelInfo::new(
            &embedder.model_config().repo,
            dim,
        ))
        .unwrap();
    store.set_dim(dim);

    let parser = CqParser::new().unwrap();
    let files = vec![PathBuf::from("nocache.rs")];
    let (count, _) = reindex_files(root, &store, &files, &parser, &embedder, None, true)
        .expect("reindex_files without global_cache");
    assert!(count >= 1, "no-cache path must still index");
}

#[test]
fn env_snapshot_redacts_api_key() {
    // CQS_LLM_API_KEY mustn't land in journald. The daemon-startup snapshot
    // in `mod.rs` builds the same redaction logic; this test pins the
    // redaction shape against a fixture so a regression that drops the suffix
    // list flips this assertion.
    const SECRET_SUFFIXES: &[&str] = &["_API_KEY", "_TOKEN", "_PASSWORD", "_SECRET"];
    let pairs = vec![
        ("CQS_LLM_API_KEY".to_string(), "sk-real-secret".to_string()),
        ("CQS_TELEMETRY".to_string(), "1".to_string()),
    ];
    let redacted: Vec<(String, String)> = pairs
        .into_iter()
        .map(|(k, v)| {
            let is_secret = SECRET_SUFFIXES.iter().any(|suffix| k.ends_with(suffix));
            let value = if is_secret {
                format!("<redacted len={}>", v.len())
            } else {
                v
            };
            (k, value)
        })
        .collect();
    assert_eq!(
        redacted[0].1, "<redacted len=14>",
        "CQS_LLM_API_KEY value must not appear in plaintext"
    );
    assert_eq!(
        redacted[1].1, "1",
        "CQS_TELEMETRY is non-secret, kept verbatim"
    );
}

// ===== process_file_changes ordering tests =====

/// `dropped_this_cycle` must NOT be reset before the embedder-init check.
/// When `try_init_embedder` early-returns (init failure or backoff), the
/// counter must survive so the outer loop's `publish_watch_snapshot` can
/// observe it as a Stale signal. Zeroing it on every path would let
/// `cqs eval --require-fresh` accept indexes whose only-witness drops had
/// been wiped.
///
/// We force the embedder-init early-return by recording an
/// `EmbedderBackoff` failure (so `should_retry()` blocks for ~2 s) on a
/// state whose `TEST_EMBEDDER` `OnceLock` has never been populated.
#[test]
fn dropped_this_cycle_survives_embedder_init_early_return() {
    let fix = drain_test_fixture(4);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );

    let mut state = test_watch_state();
    // Seed the conditions of a saturated debounce cycle: one queued
    // file plus a non-zero dropped count from prior cap-overflow events.
    state.pending_files.insert(PathBuf::from("queued.rs"));
    state.dropped_this_cycle = 5;
    // Force `try_init_embedder` → None without loading any model: the
    // shared `TEST_EMBEDDER` OnceLock is empty (set up in this file as
    // `LazyLock::new(OnceLock::new)`), so we just need backoff to block
    // the retry path. One recorded failure is enough — `record_failure`
    // sets `next_retry = now + 2 s`.
    state.embedder_backoff.record_failure();
    assert!(
        !state.embedder_backoff.should_retry(),
        "test setup: backoff must block retry so try_init_embedder returns None"
    );

    process_file_changes(&cfg, &fix.store, &mut state, &mut SiblingSet::empty());

    // Pin: the early-return path must NOT have zeroed the counter.
    assert_eq!(
        state.dropped_this_cycle, 5,
        "dropped_this_cycle must survive embedder-init early-return so the \
         next publish_watch_snapshot reports state=Stale"
    );
    // Sanity: `pending_files` is drained at the top of the function (by
    // design — collected events are taken before any work), so this is
    // not what shields the snapshot. The drop counter is.
    assert!(state.pending_files.is_empty(), "pending_files drains first");
}

/// A *successful* drain must reset `dropped_this_cycle` to 0. The reset runs
/// after `Ok((count, ...))`, so this exercises the post-drain branch and pins
/// the contract for future refactors. Uses `drain_test_fixture` (no embedder
/// needed) and confirms the function reaches the success arm by passing an
/// empty `pending_files` set after seeding the dropped counter — the function
/// processes "0 files changed" as a successful (count=0) drain.
///
/// Implementation note: even with no files, `reindex_files` returns
/// `Ok((0, vec![]))` and the success arm runs — that's the behavior we
/// want, since "we got through cleanly" is the right time to reset. A
/// CPU embedder is required to reach `reindex_files`. We use the same
/// trick as `dropped_this_cycle_survives_embedder_init_early_return`
/// inverted: this time the embedder is needed, so we don't want the
/// backoff path. We `#[ignore]` this test because it requires loading a
/// real CPU embedder model — the early-return test above carries the
/// load-bearing regression assertion; this one just documents the
/// success-arm reset contract for human readers.
#[test]
#[ignore = "Requires loading a real CPU embedder; survives_embedder_init_early_return is the load-bearing regression test"]
fn dropped_this_cycle_resets_after_successful_drain() {
    use cqs::embedder::ModelConfig;

    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path();
    let cqs_dir = root.join(".cqs");
    std::fs::create_dir_all(&cqs_dir).unwrap();

    let model_cfg = ModelConfig::resolve(None, None);
    let embedder = Embedder::new_cpu(model_cfg).expect("init CPU embedder");
    let dim = embedder.embedding_dim();
    let store_path = cqs_dir.join(cqs::INDEX_DB_FILENAME);
    let mut store = Store::open(&store_path).unwrap();
    store
        .init(&cqs::store::ModelInfo::new(
            &embedder.model_config().repo,
            dim,
        ))
        .unwrap();
    store.set_dim(dim);
    let notes_path = cqs_dir.join("docs/notes.toml");
    let supported: HashSet<&str> = HashSet::new();

    // Embedder slot pre-populated so try_init_embedder returns Some.
    let embedder_slot = std::sync::OnceLock::new();
    let _ = embedder_slot.set(std::sync::Arc::new(embedder));
    let model_cfg2 = ModelConfig::default_model();
    let parser = CqParser::new().unwrap();
    let gitignore = std::sync::RwLock::new(None);

    let cfg = WatchConfig {
        root,
        cqs_dir: &cqs_dir,
        notes_path: &notes_path,
        supported_ext: &supported,
        parser: &parser,
        embedder: &embedder_slot,
        quiet: true,
        model_config: &model_cfg2,
        gitignore: &gitignore,
        splade_encoder: None,
        global_cache: None,
    };

    let mut state = test_watch_state();
    state.dropped_this_cycle = 5;
    // No files queued → reindex_files returns Ok((0, vec![])).

    process_file_changes(&cfg, &store, &mut state, &mut SiblingSet::empty());

    assert_eq!(
        state.dropped_this_cycle, 0,
        "successful drain must reset the counter so the next cycle's \
         snapshot reflects fresh state"
    );
}

/// Empty `pending_files` plus a blocked embedder backoff is a no-op — no
/// panic, no state mutation, no embedder init.
/// This is the smallest possible exercise of `process_file_changes` and
/// pins the contract that the function tolerates the "nothing to do"
/// case even when the embedder is currently unavailable. Companion to
/// `dropped_this_cycle_survives_embedder_init_early_return` (which seeds
/// `pending_files` and `dropped_this_cycle` to test the early-return
/// path's preservation guarantee).
#[test]
fn process_file_changes_zero_files_is_noop_when_embedder_blocked() {
    let fix = drain_test_fixture(4);
    let cfg = test_watch_config(
        fix.tmp.path(),
        fix.tmp.path(),
        &fix.notes_path,
        &fix.supported_ext,
    );

    let mut state = test_watch_state();
    // Block embedder retry without loading any model. Mirrors the trick
    // from `dropped_this_cycle_survives_embedder_init_early_return`:
    // `record_failure()` advances `next_retry` to ~2 s in the future,
    // which is well past this test's runtime.
    state.embedder_backoff.record_failure();
    assert!(
        !state.embedder_backoff.should_retry(),
        "test setup: backoff must block retry so try_init_embedder returns None"
    );
    assert!(state.pending_files.is_empty());
    assert_eq!(state.dropped_this_cycle, 0);

    // Should not panic, should not mutate state.
    process_file_changes(&cfg, &fix.store, &mut state, &mut SiblingSet::empty());

    assert!(state.pending_files.is_empty(), "still empty after no-op");
    assert_eq!(
        state.dropped_this_cycle, 0,
        "no work was attempted, no counter to clear or preserve"
    );
}

// ===== touch_mtime_or_warn fail-loud helper =====

/// The parse-failure mtime-touch chain returns false and logs a distinct
/// warn at each failure step rather than silently abandoning the touch
/// (which would leave `cqs status --watch-fresh` reporting fresh while the
/// touch never landed). This test exercises the metadata() failure branch
/// (most common — file was deleted between the parse attempt and the touch).
#[test]
fn touch_mtime_or_warn_returns_false_on_missing_file() {
    use super::reindex::touch_mtime_or_warn;

    let fix = drain_test_fixture(4);
    let rel = PathBuf::from("nonexistent.rs");
    let abs = fix.tmp.path().join(&rel);
    // Deliberately do NOT create the file. metadata() will fail.
    assert!(!abs.exists(), "test setup: file must not exist");

    let ok = touch_mtime_or_warn(&fix.store, &rel, &abs);
    assert!(
        !ok,
        "metadata() failure must short-circuit and return false, not silently succeed"
    );
}

/// The success path — when the file exists, the helper reads its mtime and
/// calls `touch_source_mtime` on the store. Returns true even if no rows were
/// affected (origin format mismatch with stored chunks would produce 0 rows
/// but still indicate the FS chain succeeded). Pins the happy path so a
/// future refactor that breaks the success branch is caught.
#[test]
fn touch_mtime_or_warn_returns_true_on_extant_file() {
    use super::reindex::touch_mtime_or_warn;
    use std::io::Write;

    let fix = drain_test_fixture(4);
    let abs = fix.tmp.path().join("present.rs");
    let mut f = std::fs::File::create(&abs).unwrap();
    f.write_all(b"fn x() {}").unwrap();
    drop(f);
    let rel = PathBuf::from("present.rs");

    let ok = touch_mtime_or_warn(&fix.store, &rel, &abs);
    assert!(
        ok,
        "extant file with readable mtime must succeed; FS chain returned false unexpectedly"
    );
}

// ============================================================================
// daemon GC happy-path tests
//
// `run_daemon_startup_gc` and `run_daemon_periodic_gc` are pub(super) helpers
// in `super::gc`. These tests exercise the missing-file prune branch end-to-
// end: real on-disk files + a phantom chunk for a file that never existed.
// The gitignore-prune branch is exercised by passing `None` for the matcher
// (covered by the dedicated `prune_gitignored` unit tests).
// ============================================================================

fn make_gc_test_chunk(file: &Path, name: &str, line_start: u32) -> cqs::parser::Chunk {
    let content = format!("fn {name}() {{ }}");
    let content_hash = blake3::hash(content.as_bytes()).to_hex().to_string();
    cqs::parser::Chunk {
        id: format!("{}:{line_start}:{name}", file.display()),
        file: file.to_path_buf(),
        language: cqs::parser::Language::Rust,
        chunk_type: cqs::parser::ChunkType::Function,
        name: name.to_string(),
        signature: format!("fn {name}()"),
        content,
        doc: None,
        line_start,
        line_end: line_start + 4,
        byte_start: 0,
        content_hash,
        canonical_hash: String::new(),
        parent_id: None,
        window_idx: None,
        parent_type_name: None,
        parser_version: 0,
    }
}

/// Startup GC prunes phantom chunks for files no longer on disk.
///
/// Shares the `daemon_startup_gc_env` serial group with
/// `run_daemon_startup_gc_disabled_by_env_keeps_phantoms`: both read/write
/// `CQS_DAEMON_STARTUP_GC` (consulted by `run_daemon_startup_gc`), and the
/// disabled-test's "0" would skip this test's expected prune if they overlap.
#[test]
#[serial(daemon_startup_gc_env)]
fn run_daemon_startup_gc_prunes_phantom_chunks() {
    use super::gc::run_daemon_startup_gc;
    use std::io::Write;

    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path();

    // Create one real on-disk file.
    let live_abs = root.join("src").join("live.rs");
    std::fs::create_dir_all(live_abs.parent().unwrap()).unwrap();
    std::fs::File::create(&live_abs)
        .unwrap()
        .write_all(b"fn live_fn() {}")
        .unwrap();

    // Open Store and seed with chunks keyed by paths matching what
    // `enumerate_files` returns under this root: project-relative paths.
    let store_path = root.join(cqs::INDEX_DB_FILENAME);
    let store = Store::open(&store_path).unwrap();
    store.init(&cqs::store::ModelInfo::default()).unwrap();

    let dim = cqs::EMBEDDING_DIM;
    let mut emb = vec![0.0_f32; dim];
    emb[0] = 1.0;
    let embedding = cqs::Embedding::new(emb);

    let live_rel = PathBuf::from("src/live.rs");
    let phantom_rel = PathBuf::from("src/phantom.rs");
    let chunks = vec![
        (
            make_gc_test_chunk(&live_rel, "live_fn", 1),
            embedding.clone(),
        ),
        (make_gc_test_chunk(&phantom_rel, "phantom_fn", 1), embedding),
    ];
    store.upsert_chunks_batch(&chunks, Some(0)).unwrap();

    let before = store.stats().unwrap().total_chunks;
    assert_eq!(before, 2, "fixture must seed 2 chunks");

    let parser = CqParser::new().unwrap();

    // No env override → startup GC runs.
    let _guard = EnvVarGuard::remove("CQS_DAEMON_STARTUP_GC");
    run_daemon_startup_gc(&store, root, &parser, None);

    let after = store.stats().unwrap().total_chunks;
    assert_eq!(
        after, 1,
        "phantom chunk for nonexistent file must be pruned (before={before}, after={after})"
    );

    let live_origin = cqs::normalize_path(&live_rel);
    let live_chunks = store.get_chunks_by_origin(&live_origin).unwrap();
    assert_eq!(live_chunks.len(), 1);
    assert_eq!(live_chunks[0].name, "live_fn");
}

/// `CQS_DAEMON_STARTUP_GC=0` short-circuits the prune. Pinning it keeps the
/// operator-visible escape hatch honest.
#[test]
#[serial(daemon_startup_gc_env)]
fn run_daemon_startup_gc_disabled_by_env_keeps_phantoms() {
    use super::gc::run_daemon_startup_gc;

    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path();
    let store_path = root.join(cqs::INDEX_DB_FILENAME);
    let store = Store::open(&store_path).unwrap();
    store.init(&cqs::store::ModelInfo::default()).unwrap();

    let dim = cqs::EMBEDDING_DIM;
    let mut emb = vec![0.0_f32; dim];
    emb[0] = 1.0;
    let chunks = vec![(
        make_gc_test_chunk(&PathBuf::from("src/phantom.rs"), "phantom_fn", 1),
        cqs::Embedding::new(emb),
    )];
    store.upsert_chunks_batch(&chunks, Some(0)).unwrap();

    let parser = CqParser::new().unwrap();

    // `#[serial(daemon_startup_gc_env)]` serializes this against the sibling
    // prune test; the guard restores the prior value on drop so a panic can't
    // leak "0" into the next test.
    let _guard = EnvVarGuard::set("CQS_DAEMON_STARTUP_GC", "0");
    run_daemon_startup_gc(&store, root, &parser, None);

    let after = store.stats().unwrap().total_chunks;
    assert_eq!(
        after, 1,
        "CQS_DAEMON_STARTUP_GC=0 must skip prune; phantom must survive"
    );
}

/// Periodic GC also prunes missing-file chunks. The bounded version paths
/// through `prune_missing` the same way startup does; this test pins the
/// missing-file branch (gitignore branch is `None`).
#[test]
fn run_daemon_periodic_gc_prunes_missing_files() {
    use super::gc::run_daemon_periodic_gc;
    use std::io::Write;

    let tmp = tempfile::TempDir::new().unwrap();
    let root = tmp.path();

    let live_abs = root.join("src").join("live.rs");
    std::fs::create_dir_all(live_abs.parent().unwrap()).unwrap();
    std::fs::File::create(&live_abs)
        .unwrap()
        .write_all(b"fn live_fn() {}")
        .unwrap();

    let store_path = root.join(cqs::INDEX_DB_FILENAME);
    let store = Store::open(&store_path).unwrap();
    store.init(&cqs::store::ModelInfo::default()).unwrap();

    let dim = cqs::EMBEDDING_DIM;
    let mut emb = vec![0.0_f32; dim];
    emb[0] = 1.0;
    let embedding = cqs::Embedding::new(emb);

    let chunks = vec![
        (
            make_gc_test_chunk(&PathBuf::from("src/live.rs"), "live_fn", 1),
            embedding.clone(),
        ),
        (
            make_gc_test_chunk(&PathBuf::from("src/ghost.rs"), "ghost_fn", 1),
            embedding,
        ),
    ];
    store.upsert_chunks_batch(&chunks, Some(0)).unwrap();
    assert_eq!(store.stats().unwrap().total_chunks, 2);

    let parser = CqParser::new().unwrap();
    // disk_files: None → function does its own enumerate_files walk.
    run_daemon_periodic_gc(&store, root, &parser, None, None);

    assert_eq!(
        store.stats().unwrap().total_chunks,
        1,
        "periodic GC must prune missing-file chunks (ghost.rs)"
    );
}

// ===== publish_watch_snapshot ops block =====

/// The watch loop's snapshot publisher must carry the operational stats
/// the daemon's in-flight counter, the undrained reconcile
/// signal, and the state-recorded last-reindex/last-error pair — plus a
/// per-slot vec populated with the active slot. This drives the real
/// producer (`publish_watch_snapshot`) end-to-end into the shared
/// `Arc<RwLock<WatchSnapshot>>` the daemon's `dispatch_status` reads.
#[test]
fn publish_watch_snapshot_carries_ops_block() {
    let tmp = tempfile::tempdir().unwrap();
    let handle = cqs::watch_status::shared_unknown();
    let notifier = cqs::watch_status::shared_fresh_notifier();
    let in_flight = std::sync::atomic::AtomicUsize::new(2);
    let reconcile = std::sync::atomic::AtomicBool::new(true);

    let mut state = test_watch_state();
    state.last_reindex = Some(cqs::watch_status::ReindexLatency {
        at_unix_secs: 1_750_000_100,
        duration_ms: 1234,
        files: 3,
    });
    state.last_error = Some(cqs::watch_status::WatchErrorInfo {
        at_unix_secs: 1_749_999_000,
        message: "reindex failed: synthetic".to_string(),
    });

    publish_watch_snapshot(
        &handle,
        &notifier,
        &mut state,
        &tmp.path().join("missing-index.db"),
        &in_flight,
        &reconcile,
        &SiblingSet::empty(),
    );

    let snap = handle.read().unwrap().clone();
    let ops = snap.ops.as_ref().expect("publish must populate ops");
    assert_eq!(ops.in_flight_clients, 2);
    assert!(ops.reconcile_pending, "undrained signal must surface");
    assert_eq!(
        ops.last_reindex.as_ref().map(|l| (l.duration_ms, l.files)),
        Some((1234, 3))
    );
    assert_eq!(
        ops.last_error.as_ref().map(|e| e.message.as_str()),
        Some("reindex failed: synthetic")
    );
    // Per-slot vec: exactly the active slot, named, with matching state.
    assert_eq!(ops.slots.len(), 1);
    assert_eq!(ops.slots[0].name, cqs::slot::DEFAULT_SLOT);
    assert_eq!(ops.slots[0].state, snap.state);
    assert_eq!(
        ops.slots[0].last_reindex.as_ref().map(|l| l.duration_ms),
        Some(1234)
    );
    // Sampling must not drain the reconcile signal — the loop body owns
    // the swap-to-false.
    assert!(reconcile.load(std::sync::atomic::Ordering::Acquire));
}

/// Without a daemon (`--serve` off) the in-flight counter stays 0 and the
/// reconcile signal false — the ops block reports real zeros, not `None`,
/// so `cqs status --watch` against a non-serve watch session still
/// renders a complete block.
#[test]
fn publish_watch_snapshot_ops_zeros_without_daemon() {
    let tmp = tempfile::tempdir().unwrap();
    let handle = cqs::watch_status::shared_unknown();
    let notifier = cqs::watch_status::shared_fresh_notifier();
    let in_flight = std::sync::atomic::AtomicUsize::new(0);
    let reconcile = std::sync::atomic::AtomicBool::new(false);

    let mut state = test_watch_state();
    publish_watch_snapshot(
        &handle,
        &notifier,
        &mut state,
        &tmp.path().join("missing-index.db"),
        &in_flight,
        &reconcile,
        &SiblingSet::empty(),
    );

    let snap = handle.read().unwrap().clone();
    let ops = snap.ops.expect("ops present even without --serve");
    assert_eq!(ops.in_flight_clients, 0);
    assert!(!ops.reconcile_pending);
    assert!(ops.last_reindex.is_none());
    assert!(ops.last_error.is_none());
    assert_eq!(ops.slots.len(), 1, "active slot entry is always present");
}

// ===== Idle-flush debounce tests =====
//
// The watch loop flushes the pending sets when EITHER the quiet gap
// elapses since the last accepted event OR the oldest pending event has
// waited out the max-latency cap. These tests pin the pure decision
// (`flush_due`) and the knob resolution (`resolve_debounce`) without
// touching the filesystem or sleeping.

fn dbc(gap_ms: u64, max_ms: u64) -> DebounceConfig {
    DebounceConfig {
        quiet_gap: Duration::from_millis(gap_ms),
        max_latency: Duration::from_millis(max_ms),
    }
}

fn backdate(ms: u64) -> std::time::Instant {
    std::time::Instant::now()
        .checked_sub(Duration::from_millis(ms))
        .expect("test instant backdate")
}

#[test]
fn resolve_debounce_defaults() {
    let cfg = resolve_debounce(500, false, None, None);
    assert_eq!(cfg.quiet_gap, Duration::from_millis(500));
    assert_eq!(
        cfg.max_latency,
        Duration::from_millis(500 * u64::from(MAX_DEBOUNCE_FACTOR))
    );
}

#[test]
fn resolve_debounce_wsl_poll_auto_bump_applies_to_quiet_gap() {
    let cfg = resolve_debounce(500, true, None, None);
    assert_eq!(cfg.quiet_gap, Duration::from_millis(1500));
    assert_eq!(
        cfg.max_latency,
        Duration::from_millis(1500 * u64::from(MAX_DEBOUNCE_FACTOR))
    );
}

#[test]
fn resolve_debounce_env_gap_overrides_flag_and_suppresses_bump() {
    let cfg = resolve_debounce(500, true, Some(800), None);
    assert_eq!(cfg.quiet_gap, Duration::from_millis(800));
    assert_eq!(
        cfg.max_latency,
        Duration::from_millis(800 * u64::from(MAX_DEBOUNCE_FACTOR))
    );
}

#[test]
fn resolve_debounce_explicit_flag_suppresses_bump() {
    let cfg = resolve_debounce(1000, true, None, None);
    assert_eq!(cfg.quiet_gap, Duration::from_millis(1000));
}

#[test]
fn resolve_debounce_env_max_override() {
    let cfg = resolve_debounce(500, false, None, Some(2000));
    assert_eq!(cfg.max_latency, Duration::from_millis(2000));
}

#[test]
fn resolve_debounce_max_clamped_to_at_least_quiet_gap() {
    let cfg = resolve_debounce(500, false, None, Some(100));
    assert_eq!(
        cfg.max_latency,
        Duration::from_millis(500),
        "a cap below the quiet gap would invert the two knobs' contract"
    );
}

#[test]
fn flush_due_no_pending_never_flushes() {
    let mut state = test_watch_state();
    // Even with a long-stale last event and an armed burst clock,
    // empty pending sets never flush.
    state.last_event = backdate(60_000);
    state.first_pending_event = Some(backdate(60_000));
    assert!(!flush_due(&state, &dbc(500, 3000)));
}

#[test]
fn flush_due_single_save_flushes_at_quiet_gap() {
    // Gate: single-save latency no worse than the old fixed window —
    // one event flushes exactly one quiet gap after it arrived.
    let mut state = test_watch_state();
    state.pending_files.insert(PathBuf::from("src/lib.rs"));
    state.last_event = backdate(499);
    state.first_pending_event = Some(backdate(499));
    assert!(
        !flush_due(&state, &dbc(500, 3000)),
        "not yet — quiet gap has not elapsed"
    );
    state.last_event = backdate(500);
    state.first_pending_event = Some(backdate(500));
    assert!(flush_due(&state, &dbc(500, 3000)));
}

#[test]
fn flush_due_burst_fires_quiet_gap_after_last_event_not_first() {
    // Gate: a multi-file burst produces one flush fired ~quiet-gap
    // after the LAST event. Simulated burst: first event 600 ms ago
    // (already past one full quiet gap), last event 100 ms ago.
    let mut state = test_watch_state();
    for i in 0..200 {
        state
            .pending_files
            .insert(PathBuf::from(format!("f{i}.rs")));
    }
    state.first_pending_event = Some(backdate(600));
    state.last_event = backdate(100);
    assert!(
        !flush_due(&state, &dbc(500, 3000)),
        "burst still hot — quiet gap restarts on every event"
    );
    state.last_event = backdate(500);
    assert!(
        flush_due(&state, &dbc(500, 3000)),
        "quiet gap elapsed since the last event of the burst"
    );
}

#[test]
fn flush_due_disarmed_after_flush_does_not_refire() {
    // Gate: ONE flush per burst. After the flush block drains the
    // pending sets and clears `first_pending_event`, a stale
    // `last_event` alone must not refire.
    let mut state = test_watch_state();
    state.last_event = backdate(10_000);
    state.first_pending_event = None;
    state.pending_files.clear();
    state.pending_notes = false;
    assert!(!flush_due(&state, &dbc(500, 3000)));
}

#[test]
fn flush_due_continuous_stream_capped_by_max_latency() {
    // Gate: a stream that never goes quiet still flushes within the
    // cap. The quiet-gap timer keeps restarting (last_event is fresh)
    // but the burst clock has hit the cap.
    let mut state = test_watch_state();
    state.pending_files.insert(PathBuf::from("gen.rs"));
    state.last_event = backdate(50);
    state.first_pending_event = Some(backdate(3000));
    assert!(flush_due(&state, &dbc(500, 3000)));
}

#[test]
fn flush_due_continuous_stream_below_cap_waits() {
    let mut state = test_watch_state();
    state.pending_files.insert(PathBuf::from("gen.rs"));
    state.last_event = backdate(50);
    state.first_pending_event = Some(backdate(2000));
    assert!(!flush_due(&state, &dbc(500, 3000)));
}

#[test]
fn flush_due_notes_only_pending_flushes() {
    let mut state = test_watch_state();
    state.pending_notes = true;
    state.last_event = backdate(500);
    state.first_pending_event = Some(backdate(500));
    assert!(flush_due(&state, &dbc(500, 3000)));
}

#[test]
fn collect_events_arms_burst_clock_once_per_burst() {
    // `first_pending_event` is sticky: the first accepted event arms
    // it, later events in the same burst leave it untouched (only
    // `last_event` keeps advancing). The flush block disarms it.
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();
    assert!(state.first_pending_event.is_none());

    let kind = EventKind::Modify(notify::event::ModifyKind::Data(
        notify::event::DataChange::Content,
    ));
    collect_events(
        &make_event(vec![PathBuf::from("/tmp/test_project/a.rs")], kind),
        &cfg,
        &mut state,
    );
    let armed = state
        .first_pending_event
        .expect("first accepted event arms the burst clock");

    collect_events(
        &make_event(vec![PathBuf::from("/tmp/test_project/b.rs")], kind),
        &cfg,
        &mut state,
    );
    assert_eq!(
        state.first_pending_event,
        Some(armed),
        "second event of the burst must not restart the burst clock"
    );
    assert_eq!(state.pending_files.len(), 2);
}

#[test]
fn collect_events_filtered_event_does_not_arm_burst_clock() {
    let root = PathBuf::from("/tmp/test_project");
    let cqs_dir = PathBuf::from("/tmp/test_project/.cqs");
    let notes_path = PathBuf::from("/tmp/test_project/docs/notes.toml");
    let supported: HashSet<&str> = ["rs"].iter().cloned().collect();
    let cfg = test_watch_config(&root, &cqs_dir, &notes_path, &supported);
    let mut state = test_watch_state();

    collect_events(
        &make_event(
            vec![PathBuf::from("/tmp/test_project/readme.txt")],
            EventKind::Modify(notify::event::ModifyKind::Data(
                notify::event::DataChange::Content,
            )),
        ),
        &cfg,
        &mut state,
    );
    assert!(
        state.first_pending_event.is_none(),
        "filtered events queue nothing and must not arm the burst clock"
    );
}
