use parking_lot::Mutex;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::{Duration, Instant, SystemTime};

use notify::{EventKind, RecommendedWatcher as NotifyRecommendedWatcher, RecursiveMode};
use notify_debouncer_full::{
    DebounceEventResult, DebouncedEvent, Debouncer, RecommendedCache, new_debouncer,
};
use tracing::{debug, error, trace, warn};

use crate::domain::{FileClassification, LanguageId};
use crate::live_index::store::{IndexedFile, SharedIndex};
use crate::{hash, parsing};

// Watcher state snapshot types live in the engine-safe `watcher_state` module so
// the engine's health stats can use them in `embed` builds; the notify-based
// runtime below is server-only.
pub use crate::watcher_state::{WatcherInfo, WatcherState};

/// Tracks event bursts to adaptively extend the debounce window.
///
/// Debounce logic:
/// - Base window: 200ms
/// - Burst window: 500ms (when >BURST_THRESHOLD events in a 200ms window)
/// - Resets to 200ms after QUIET_SECS of inactivity
pub struct BurstTracker {
    pub event_count: u32,
    pub window_start: Instant,
    pub last_event_at: Instant,
    pub extended: bool,
}

impl BurstTracker {
    const BURST_THRESHOLD: u32 = 3;
    const BASE_MS: u64 = 200;
    const BURST_MS: u64 = 500;
    const QUIET_SECS: u64 = 5;

    /// Create a new BurstTracker with all counters at zero.
    pub fn new() -> Self {
        let now = Instant::now();
        BurstTracker {
            event_count: 0,
            window_start: now,
            last_event_at: now,
            extended: false,
        }
    }

    /// Record an event at the given instant, updating burst state.
    ///
    /// Window logic: if `now - window_start > BASE_MS`, start a new window
    /// and reset count to 1. Otherwise increment count.
    /// If count exceeds BURST_THRESHOLD, set extended=true.
    /// Always updates last_event_at.
    pub fn update(&mut self, now: Instant) {
        let window_duration = now.duration_since(self.window_start);
        if window_duration > Duration::from_millis(Self::BASE_MS) {
            // Start a new window
            self.window_start = now;
            self.event_count = 1;
            self.extended = false;
        } else {
            self.event_count += 1;
            if self.event_count > Self::BURST_THRESHOLD {
                self.extended = true;
            }
        }
        self.last_event_at = now;
    }

    /// Returns the effective debounce window in milliseconds.
    ///
    /// - If last event was more than QUIET_SECS ago, return BASE_MS (quiet reset)
    /// - If in burst mode (extended=true), return BURST_MS
    /// - Otherwise return BASE_MS
    pub fn effective_debounce_ms(&self) -> u64 {
        let since_last = self.last_event_at.elapsed();
        if since_last > Duration::from_secs(Self::QUIET_SECS) {
            return Self::BASE_MS;
        }
        if self.extended {
            Self::BURST_MS
        } else {
            Self::BASE_MS
        }
    }
}

impl Default for BurstTracker {
    fn default() -> Self {
        Self::new()
    }
}

// ---------------------------------------------------------------------------
// Plan 02: Event processing, path normalization, content hash skip, ENOENT
// ---------------------------------------------------------------------------

/// Result of a single re-index attempt for one file.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum ReindexResult {
    /// Content hash matched existing entry — tree-sitter parse was skipped.
    HashSkip,
    /// File was re-parsed and the index was updated.
    Reindexed,
    /// ENOENT observed by `read_and_index`; caller decides whether to retry or treat as confirmed-absent.
    NotFound,
    /// File was not found (ENOENT) — it has been removed from the index.
    Removed,
    /// File could not be read for a reason other than ENOENT.
    ReadError(String),
}

#[must_use]
pub(crate) enum FreshenResult {
    Fresh,
    StaleReindexed,
    StaleRemoved,
    GenerationMismatch,
}

/// Strip `\\?\` Windows extended-length path prefix and normalize backslashes.
///
/// Returns the relative forward-slash path if `abs_path` is inside `repo_root`,
/// or `None` if it lies outside.
pub(crate) fn normalize_event_path(abs_path: &Path, repo_root: &Path) -> Option<String> {
    let raw_path = abs_path.to_string_lossy();

    // Strip \\?\ prefix (Windows extended-length format)
    let stripped_raw: &str = if let Some(stripped) = raw_path.strip_prefix(r"\\?\") {
        stripped
    } else {
        raw_path.as_ref()
    };

    let clean_abs = Path::new(stripped_raw);

    // Try strip_prefix with the original repo_root first, then with its own \\?\ stripped
    let relative = clean_abs.strip_prefix(repo_root).or_else(|_| {
        let root_raw = repo_root.to_string_lossy();
        let stripped_root: &str = if let Some(stripped) = root_raw.strip_prefix(r"\\?\") {
            stripped
        } else {
            return clean_abs.strip_prefix(repo_root);
        };
        clean_abs.strip_prefix(Path::new(stripped_root))
    });

    relative
        .ok()
        .map(|p| p.to_string_lossy().replace('\\', "/"))
}

/// Return the `LanguageId` for a file path based on its extension.
///
/// Returns `None` for unsupported or missing extensions.
pub(crate) fn supported_language(path: &Path) -> Option<LanguageId> {
    let ext = path.extension()?.to_str()?;
    LanguageId::from_extension(ext)
}

/// Return `true` for Create, Modify, or Remove events; `false` for Access and others.
pub(crate) fn is_relevant_event(event: &DebouncedEvent) -> bool {
    matches!(
        event.kind,
        EventKind::Create(_) | EventKind::Modify(_) | EventKind::Remove(_)
    )
}

/// Content-hash-gated single-file re-index.
///
/// Reads the file, compares its hash against the existing index entry, and
/// skips the expensive tree-sitter parse when the hash matches.
///
/// # Lock discipline
/// The write lock is **never** held during the tree-sitter parse. The sequence is:
/// 1. Read file bytes (no lock)
/// 2. Acquire read lock → compare hash → drop read lock
/// 3. Parse (no lock)
/// 4. Acquire write lock → update_file → drop write lock
pub(crate) fn maybe_reindex(
    relative_path: &str,
    abs_path: &Path,
    shared: &SharedIndex,
    language: LanguageId,
    expected_gen: u64,
) -> ReindexResult {
    match read_and_index(
        relative_path,
        abs_path,
        shared,
        language.clone(),
        expected_gen,
    ) {
        ReindexResult::NotFound => {}
        other => return other,
    }

    let delays_ms = [50u64, 200, 500];
    for delay_ms in delays_ms {
        std::thread::sleep(std::time::Duration::from_millis(delay_ms));
        match read_and_index(
            relative_path,
            abs_path,
            shared,
            language.clone(),
            expected_gen,
        ) {
            ReindexResult::NotFound => continue,
            other => return other,
        }
    }

    if shared.remove_file_at_generation(relative_path, expected_gen) {
        warn!("watcher: file not found after retries, removed from index: {relative_path}");
    } else {
        trace!(
            "watcher: file not found after retries, stale generation rejected remove: {relative_path}"
        );
    }
    ReindexResult::Removed
}

fn read_and_index(
    relative_path: &str,
    abs_path: &Path,
    shared: &SharedIndex,
    language: LanguageId,
    expected_gen: u64,
) -> ReindexResult {
    // 1. Read mtime BEFORE content to avoid TOCTOU: if the file is written
    //    between stat and read, we get an mtime that is older-or-equal to the
    //    content we actually parsed, so a future watcher event will correctly
    //    detect staleness. The reverse order (read then stat) can record a
    //    newer mtime paired with older content, permanently hiding the change.
    let mtime_secs = match std::fs::metadata(abs_path) {
        Ok(metadata) => metadata
            .modified()
            .ok()
            .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
            .map(|d| d.as_secs())
            .unwrap_or(0),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return ReindexResult::NotFound,
        Err(_) => 0,
    };
    let bytes = match std::fs::read(abs_path) {
        Ok(b) => b,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return ReindexResult::NotFound,
        Err(e) => {
            warn!("watcher: failed to read {relative_path}: {e}");
            return ReindexResult::ReadError(e.to_string());
        }
    };

    // 2. Compute hash and check against existing entry (read lock, dropped before parse)
    let new_hash = hash::digest_hex(&bytes);
    {
        let index = shared.read();
        if let Some(existing) = index.get_file(relative_path)
            && existing.content_hash == new_hash
        {
            // Content unchanged but mtime may have drifted (e.g., git rebase, touch).
            // Update the stored mtime so the reconciliation loop doesn't re-check
            // this file on every sweep. Without this, a hash-skip leaves the old
            // mtime in the index, causing an infinite stale → hash-skip → stale loop.
            let needs_mtime_touch = existing.mtime_secs != mtime_secs && mtime_secs != 0;
            drop(index); // release read lock before potential write
            if needs_mtime_touch {
                shared.touch_mtime_at_generation(relative_path, mtime_secs, expected_gen);
            }
            debug!("watcher: hash-skip {relative_path}");
            return ReindexResult::HashSkip;
        }
        // read lock dropped here
    }

    // 3. Parse outside the lock
    let result = parsing::process_file_with_classification(
        relative_path,
        &bytes,
        language,
        FileClassification::for_code_path(relative_path),
    );
    let indexed = IndexedFile::from_parse_result(result, bytes).with_mtime(mtime_secs);

    // 4. Acquire write lock and update
    shared.update_file_at_generation(relative_path, indexed, expected_gen);

    debug!("watcher: re-indexed {relative_path}");
    ReindexResult::Reindexed
}

/// Mtime-based freshness check for a single file.
///
/// Compares the file's current mtime on disk against the value stored in the
/// index. If they differ (or the file is not yet indexed), re-indexes it
/// immediately before the caller proceeds.
///
/// Returns a structured freshness outcome so callers can distinguish a
/// confirmed deletion from a stale project-generation mismatch.
pub(crate) fn freshen_file_if_stale(
    relative_path: &str,
    abs_path: &Path,
    shared: &SharedIndex,
    expected_gen: u64,
) -> FreshenResult {
    if shared.current_project_generation() != expected_gen {
        let _ = shared.remove_file_at_generation(relative_path, expected_gen);
        return FreshenResult::GenerationMismatch;
    }

    // 1. Stat the file on disk
    let disk_mtime = std::fs::metadata(abs_path)
        .and_then(|m| m.modified())
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map(|d| d.as_secs())
        .unwrap_or(0);

    // 2. Compare against indexed mtime (read lock, released immediately)
    let indexed_mtime = {
        let index = shared.read();
        index
            .get_file(relative_path)
            .map(|f| f.mtime_secs)
            .unwrap_or(u64::MAX)
    };

    if disk_mtime == 0 && indexed_mtime == 0 {
        return FreshenResult::Fresh; // both unknown — treat as fresh to avoid churn
    }
    if disk_mtime != 0 && disk_mtime == indexed_mtime {
        return FreshenResult::Fresh; // already fresh
    }

    // 3. Stale — re-index
    let language = match supported_language(abs_path) {
        Some(l) => l,
        None => return FreshenResult::Fresh,
    };

    debug!("freshness guard: stale file detected, re-indexing {relative_path}");
    let result = maybe_reindex(relative_path, abs_path, shared, language, expected_gen);
    if shared.current_project_generation() != expected_gen {
        let _ = shared.remove_file_at_generation(relative_path, expected_gen);
        return FreshenResult::GenerationMismatch;
    }

    match result {
        ReindexResult::HashSkip | ReindexResult::Reindexed | ReindexResult::ReadError(_) => {
            FreshenResult::StaleReindexed
        }
        ReindexResult::NotFound | ReindexResult::Removed => FreshenResult::StaleRemoved,
    }
}

fn freshen_file_if_stale_at_generation(
    relative_path: &str,
    abs_path: &Path,
    shared: &SharedIndex,
    expected_gen: u64,
) -> bool {
    matches!(
        freshen_file_if_stale(relative_path, abs_path, shared, expected_gen),
        FreshenResult::StaleReindexed
            | FreshenResult::StaleRemoved
            | FreshenResult::GenerationMismatch
    )
}

/// Walk all indexed files and re-index any whose on-disk mtime differs from
/// the stored value. Returns the number of stale files re-indexed.
///
/// Called on watcher overflow and by the periodic reconciliation timer.
pub(crate) fn reconcile_stale_files_with_stop(
    repo_root: &Path,
    shared: &SharedIndex,
    should_stop: impl Fn() -> bool,
    expected_gen: u64,
) -> usize {
    let paths: Vec<String> = {
        let index = shared.read();
        index.all_files().map(|(p, _)| p.clone()).collect()
    };

    let mut stale_count = 0usize;
    for relative_path in &paths {
        if should_stop() {
            break;
        }
        let abs_path = repo_root.join(relative_path);
        if freshen_file_if_stale_at_generation(relative_path, &abs_path, shared, expected_gen) {
            stale_count += 1;
        }
    }

    if stale_count > 0 {
        // Collect stale paths for diagnostic logging to help debug reconciliation loops.
        let stale_paths: Vec<&str> = paths
            .iter()
            .filter(|p| {
                let abs = repo_root.join(p.as_str());
                let disk = std::fs::metadata(&abs)
                    .and_then(|m| m.modified())
                    .ok()
                    .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                    .map(|d| d.as_secs())
                    .unwrap_or(0);
                let indexed = {
                    let idx = shared.read();
                    idx.get_file(p).map(|f| f.mtime_secs).unwrap_or(0)
                };
                disk != indexed
            })
            .map(|p| p.as_str())
            .take(5)
            .collect();
        if stale_paths.is_empty() {
            warn!("reconciliation re-indexed {stale_count} file(s) (now fresh)");
        } else {
            warn!(
                "reconciliation found {stale_count} stale file(s), still divergent: {}",
                stale_paths.join(", ")
            );
        }
    }
    stale_count
}

pub(crate) fn reconcile_stale_files(repo_root: &Path, shared: &SharedIndex) -> usize {
    let expected_gen = shared.current_project_generation();
    reconcile_stale_files_with_stop(repo_root, shared, || false, expected_gen)
}

// ---------------------------------------------------------------------------
// Plan 02: Watcher lifecycle — start_watcher, run_watcher, restart-with-backoff
// ---------------------------------------------------------------------------

/// Owns the debouncer and the receiving end of the event channel.
///
/// Dropping this struct stops the OS-level file watcher.
pub struct WatcherHandle {
    /// The debouncer owns the OS watcher thread; dropping it stops watching.
    _debouncer: Debouncer<NotifyRecommendedWatcher, RecommendedCache>,
    /// Receive end of the synchronous channel from the notify callback.
    pub event_rx: std::sync::mpsc::Receiver<DebounceEventResult>,
}

/// Owned together: signal `stop_token`, then bounded-await `task`.
/// See H.1b's `abort_watcher_task` for the canonical shutdown sequence.
pub struct WatcherTaskHandle {
    pub task: tokio::task::JoinHandle<()>,
    pub stop_token: Arc<AtomicBool>,
}

/// Create a new debouncer watching `repo_root` recursively.
///
/// `debounce_ms` controls the debounce window (base 200ms, extended to 500ms during bursts).
/// Uses `std::sync::mpsc` (not tokio) because notify's callback runs on its own OS thread.
pub(crate) fn start_watcher(
    repo_root: &Path,
    debounce_ms: u64,
) -> Result<WatcherHandle, notify::Error> {
    let (tx, rx) = std::sync::mpsc::channel::<DebounceEventResult>();

    let mut debouncer = new_debouncer(
        Duration::from_millis(debounce_ms),
        None,
        move |result: DebounceEventResult| {
            let _ = tx.send(result);
        },
    )?;

    debouncer.watch(repo_root, RecursiveMode::Recursive)?;

    Ok(WatcherHandle {
        _debouncer: debouncer,
        event_rx: rx,
    })
}

pub(crate) fn process_events(
    events: Vec<DebouncedEvent>,
    repo_root: &Path,
    shared: &SharedIndex,
    burst_trackers: &mut HashMap<PathBuf, BurstTracker>,
    watcher_info: &Arc<Mutex<WatcherInfo>>,
    should_stop: &dyn Fn() -> bool,
    expected_gen: u64,
) {
    for event in events {
        if should_stop() {
            break;
        }
        if !is_relevant_event(&event) {
            continue;
        }

        for abs_path in &event.paths {
            if should_stop() {
                break;
            }
            // Normalize path — skip if outside repo_root or can't be normalized
            let relative_path = match normalize_event_path(abs_path, repo_root) {
                Some(r) => r,
                None => continue,
            };

            // Filter to supported languages only
            let language = match supported_language(abs_path) {
                Some(l) => l,
                None => continue,
            };

            // Mirror discovery's gitignore-aware walk: never index paths the
            // initial scan would have pruned. Without this the watcher picks
            // up files created under gitignored directories during a session —
            // most importantly SymForge's own `.symforge/` state dir (e.g.
            // `tee/*.rs` edit snapshots) — polluting search and reference
            // results and growing the index unbounded.
            if shared.read().is_path_gitignored(&relative_path) {
                continue;
            }

            match event.kind {
                EventKind::Remove(_) => {
                    shared.remove_file_at_generation(&relative_path, expected_gen);

                    let mut info = watcher_info.lock();
                    info.events_processed += 1;
                    info.last_event_at = Some(SystemTime::now());
                }
                EventKind::Create(_) | EventKind::Modify(_) => {
                    // Update burst tracker for this path
                    let now = Instant::now();
                    let tracker = burst_trackers.entry(abs_path.clone()).or_default();
                    tracker.update(now);
                    let debounce_ms = tracker.effective_debounce_ms();

                    maybe_reindex(&relative_path, abs_path, shared, language, expected_gen);

                    let mut info = watcher_info.lock();
                    info.events_processed += 1;
                    info.last_event_at = Some(SystemTime::now());
                    info.debounce_window_ms = debounce_ms;
                }
                _ => {}
            }
        }
    }

    // Evict burst trackers that have been idle longer than 2 × QUIET_SECS to
    // prevent the map from growing unbounded over the lifetime of the watcher.
    // NOTE: eviction only runs after file-change events, not during overflow
    // reconciliation, so trackers for paths not recently seen are cleaned up
    // lazily on the next incoming event.
    let evict_threshold = Duration::from_secs(BurstTracker::QUIET_SECS * 2);
    burst_trackers.retain(|_, tracker| tracker.last_event_at.elapsed() < evict_threshold);
}

/// Main watcher supervision loop. Spawned as a background tokio task by `main.rs`.
///
/// Lifecycle:
/// 1. Set state to Active
/// 2. Loop: start_watcher → process events → restart on error with 1s backoff
/// 3. After 3 consecutive failures: set state to Degraded and stop
pub async fn run_watcher_with_stop(
    repo_root: PathBuf,
    shared: SharedIndex,
    watcher_info: Arc<Mutex<WatcherInfo>>,
    stop_token: Arc<AtomicBool>,
) {
    let expected_gen = shared.current_project_generation();
    if stop_token.load(Ordering::Acquire) {
        let mut info = watcher_info.lock();
        info.state = WatcherState::Off;
        return;
    }

    {
        let mut info = watcher_info.lock();
        info.state = WatcherState::Active;
    }

    let mut consecutive_failures: u32 = 0;
    const MAX_FAILURES: u32 = 3;
    let mut cancelled = false;

    'watcher: loop {
        if stop_token.load(Ordering::Acquire) {
            cancelled = true;
            break;
        }

        // Read the current recommended debounce window (updated by the burst tracker).
        let debounce_ms = watcher_info.lock().debounce_window_ms;
        match start_watcher(&repo_root, debounce_ms) {
            Err(e) => {
                consecutive_failures += 1;
                warn!(
                    "watcher: start_watcher failed (attempt {}): {}",
                    consecutive_failures, e
                );
                if consecutive_failures >= MAX_FAILURES {
                    let mut info = watcher_info.lock();
                    info.state = WatcherState::Degraded;
                    error!(
                        "watcher: entering degraded mode after {} consecutive failures",
                        MAX_FAILURES
                    );
                    break;
                }
                tokio::time::sleep(Duration::from_secs(1)).await;
                continue;
            }
            Ok(handle) => {
                consecutive_failures = 0;
                {
                    let mut info = watcher_info.lock();
                    info.state = WatcherState::Active;
                }

                let mut burst_trackers: HashMap<PathBuf, BurstTracker> = HashMap::new();
                let mut session_errors: u32 = 0;
                const MAX_SESSION_ERRORS: u32 = 10;
                // Poll timeout: yield to tokio between checks to avoid blocking the executor.
                const RECV_TIMEOUT_MS: u64 = 50;

                // Reconciliation interval from env (default 30s, 0 to disable).
                let reconcile_interval_secs: u64 = std::env::var("SYMFORGE_RECONCILE_INTERVAL")
                    .ok()
                    .and_then(|v| v.parse().ok())
                    .unwrap_or(30);
                let mut last_reconcile = Instant::now();

                loop {
                    if stop_token.load(Ordering::Acquire) {
                        cancelled = true;
                        break 'watcher;
                    }

                    // Periodic reconciliation sweep (belt-and-suspenders against missed events).
                    if reconcile_interval_secs > 0
                        && last_reconcile.elapsed() >= Duration::from_secs(reconcile_interval_secs)
                    {
                        let shared_clone = shared.clone();
                        let root_clone = repo_root.clone();
                        let watcher_info_clone = watcher_info.clone();
                        let stop_for_reconcile = Arc::clone(&stop_token);
                        let expected_gen_for_reconcile = expected_gen;
                        tokio::task::spawn_blocking(move || {
                            let stale = reconcile_stale_files_with_stop(
                                &root_clone,
                                &shared_clone,
                                || stop_for_reconcile.load(Ordering::Acquire),
                                expected_gen_for_reconcile,
                            );
                            if stop_for_reconcile.load(Ordering::Acquire) {
                                return;
                            }
                            let mut info = watcher_info_clone.lock();
                            info.stale_files_found += stale as u64;
                            info.last_reconcile_at = Some(SystemTime::now());
                        });
                        // Coupling store refresh runs on its own task so a
                        // slow delta never delays stale-file reconciliation.
                        // Gates on SYMFORGE_COUPLING internally and holds a
                        // per-workspace guard against concurrent refreshes.
                        let root_for_coupling = repo_root.clone();
                        let stop_for_coupling = Arc::clone(&stop_token);
                        let expected_gen_for_coupling = expected_gen;
                        let shared_for_coupling = shared.clone();
                        tokio::task::spawn_blocking(move || {
                            if stop_for_coupling.load(Ordering::Acquire) {
                                return;
                            }
                            crate::live_index::coupling::refresh_on_reconcile_tick(
                                &root_for_coupling,
                                expected_gen_for_coupling,
                                &shared_for_coupling,
                            );
                        });
                        last_reconcile = Instant::now();
                    }

                    match handle.event_rx.try_recv() {
                        Ok(Ok(events)) => {
                            // Run process_events in spawn_blocking to avoid
                            // starving tokio worker threads during file I/O
                            // and tree-sitter parsing.
                            let shared_clone = shared.clone();
                            let root_clone = repo_root.clone();
                            let watcher_info_clone = watcher_info.clone();
                            let stop_for_events = Arc::clone(&stop_token);
                            let expected_gen_for_events = expected_gen;
                            let mut trackers = std::mem::take(&mut burst_trackers);
                            match tokio::task::spawn_blocking(move || {
                                process_events(
                                    events,
                                    &root_clone,
                                    &shared_clone,
                                    &mut trackers,
                                    &watcher_info_clone,
                                    &|| stop_for_events.load(Ordering::Acquire),
                                    expected_gen_for_events,
                                );
                                trackers
                            })
                            .await
                            {
                                Ok(t) => burst_trackers = t,
                                // Intentional: on panic the burst trackers reset to empty.
                                // This is acceptable — burst tracking is a performance
                                // optimization, not a correctness requirement.
                                Err(e) => warn!("watcher: process_events panicked: {e}"),
                            }
                        }
                        Ok(Err(errors)) => {
                            let mut overflow_detected = false;
                            for err in &errors {
                                warn!("watcher: notify error: {err}");
                                // Detect watcher buffer overflow / rescan events.
                                if matches!(
                                    err.kind,
                                    notify::ErrorKind::Io(_)
                                        | notify::ErrorKind::Generic(_)
                                        | notify::ErrorKind::MaxFilesWatch
                                ) {
                                    overflow_detected = true;
                                }
                            }
                            if overflow_detected {
                                warn!(
                                    "watcher: buffer overflow detected — running full reconciliation"
                                );
                                let shared_clone = shared.clone();
                                let root_clone = repo_root.clone();
                                let watcher_info_clone = watcher_info.clone();
                                let stop_for_reconcile = Arc::clone(&stop_token);
                                let expected_gen_for_reconcile = expected_gen;
                                tokio::task::spawn_blocking(move || {
                                    let stale = reconcile_stale_files_with_stop(
                                        &root_clone,
                                        &shared_clone,
                                        || stop_for_reconcile.load(Ordering::Acquire),
                                        expected_gen_for_reconcile,
                                    );
                                    if stop_for_reconcile.load(Ordering::Acquire) {
                                        return;
                                    }
                                    let mut info = watcher_info_clone.lock();
                                    info.overflow_count += 1;
                                    info.last_overflow_at = Some(SystemTime::now());
                                    info.stale_files_found += stale as u64;
                                    info.last_reconcile_at = Some(SystemTime::now());
                                });
                            }
                            session_errors += u32::try_from(errors.len()).unwrap_or(u32::MAX);
                            if session_errors >= MAX_SESSION_ERRORS {
                                warn!("watcher: too many session errors, restarting watcher");
                                break;
                            }
                        }
                        Err(std::sync::mpsc::TryRecvError::Empty) => {
                            // No event ready — yield to tokio async executor
                            // instead of blocking the worker thread.
                            tokio::time::sleep(Duration::from_millis(RECV_TIMEOUT_MS)).await;
                        }
                        Err(std::sync::mpsc::TryRecvError::Disconnected) => {
                            // Channel closed — debouncer dropped or OS watcher died
                            warn!("watcher: event channel closed, restarting");
                            break;
                        }
                    }
                }

                // Inner loop exited — count as a failure and try to restart
                consecutive_failures += 1;
                if consecutive_failures >= MAX_FAILURES {
                    let mut info = watcher_info.lock();
                    info.state = WatcherState::Degraded;
                    error!(
                        "watcher: entering degraded mode after {} consecutive failures",
                        MAX_FAILURES
                    );
                    break;
                }
                tokio::time::sleep(Duration::from_secs(1)).await;
            }
        }
    }

    if cancelled {
        let mut info = watcher_info.lock();
        info.state = WatcherState::Off;
    }
}

pub async fn run_watcher(
    repo_root: PathBuf,
    shared: SharedIndex,
    watcher_info: Arc<Mutex<WatcherInfo>>,
) {
    let stop_token = Arc::new(AtomicBool::new(false));
    run_watcher_with_stop(repo_root, shared, watcher_info, stop_token).await;
}

/// Spawn a new watcher task.
///
/// Called by `index_folder` after a full reload to restart the watcher
/// on the new root path.
pub fn restart_watcher(
    repo_root: PathBuf,
    shared: SharedIndex,
    watcher_info: Arc<Mutex<WatcherInfo>>,
    prev: Option<WatcherTaskHandle>,
) -> WatcherTaskHandle {
    {
        let mut info = watcher_info.lock();
        info.state = WatcherState::Off;
    }
    let stop_token = Arc::new(AtomicBool::new(false));
    let stop_for_task = Arc::clone(&stop_token);
    let task = tokio::spawn(async move {
        if let Some(prev) = prev {
            prev.stop_token.store(true, Ordering::Release);
            let mut old_task = prev.task;
            if tokio::time::timeout(Duration::from_secs(2), &mut old_task)
                .await
                .is_err()
            {
                warn!("watcher: previous watcher did not stop within 2s; aborting task");
                old_task.abort();
            }
        }
        run_watcher_with_stop(repo_root, shared, watcher_info, stop_for_task).await;
    });
    WatcherTaskHandle { task, stop_token }
}

// ---------------------------------------------------------------------------
// Unit tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use std::time::Duration;
    use tempfile::TempDir;

    // --- BurstTracker tests from Plan 01 (preserved) ---

    #[test]
    fn test_watcher_state_variants() {
        // All three variants exist and are distinct
        let active = WatcherState::Active;
        let degraded = WatcherState::Degraded;
        let off = WatcherState::Off;
        assert_ne!(active, degraded);
        assert_ne!(active, off);
        assert_ne!(degraded, off);
    }

    #[test]
    fn test_watcher_info_default() {
        let info = WatcherInfo::default();
        assert_eq!(info.state, WatcherState::Off);
        assert_eq!(info.events_processed, 0);
        assert!(info.last_event_at.is_none());
        assert_eq!(info.debounce_window_ms, 200);
    }

    #[test]
    fn test_burst_tracker_new() {
        let tracker = BurstTracker::new();
        assert_eq!(tracker.event_count, 0);
        assert!(!tracker.extended);
    }

    #[test]
    fn test_burst_tracker_extends_window() {
        // 4 events within 200ms -> extended=true, effective=500
        let mut tracker = BurstTracker::new();
        let start = Instant::now();
        // Simulate 4 rapid events within the same 200ms window
        tracker.update(start + Duration::from_millis(10));
        tracker.update(start + Duration::from_millis(20));
        tracker.update(start + Duration::from_millis(30));
        tracker.update(start + Duration::from_millis(40));
        assert!(tracker.extended, "4 events in window should trigger burst");
        assert_eq!(tracker.effective_debounce_ms(), 500);
    }

    #[test]
    fn test_burst_tracker_resets_after_quiet() {
        // After last event > 5s ago, effective should return 200
        let mut tracker = BurstTracker::new();
        let past = Instant::now() - Duration::from_secs(10);
        // We simulate this by forcing extended=true and setting last_event_at in the past
        tracker.extended = true;
        tracker.last_event_at = past;
        assert_eq!(
            tracker.effective_debounce_ms(),
            200,
            "after quiet period, should reset to 200ms"
        );
    }

    #[test]
    fn test_burst_tracker_new_window_resets_count() {
        // An event after >200ms gap should start a fresh window with count=1, extended=false
        let mut tracker = BurstTracker::new();
        let t0 = Instant::now();
        // First burst: 4 events
        tracker.update(t0 + Duration::from_millis(10));
        tracker.update(t0 + Duration::from_millis(20));
        tracker.update(t0 + Duration::from_millis(30));
        tracker.update(t0 + Duration::from_millis(40));
        assert!(tracker.extended, "should be extended after burst");

        // Event after 300ms gap
        tracker.update(t0 + Duration::from_millis(350));
        assert_eq!(tracker.event_count, 1, "count should reset to 1 after gap");
        assert!(!tracker.extended, "extended should reset after new window");
    }

    #[test]
    fn test_burst_tracker_base_debounce_no_burst() {
        // Under threshold: effective should remain 200ms
        let mut tracker = BurstTracker::new();
        let t0 = Instant::now();
        tracker.update(t0 + Duration::from_millis(10));
        tracker.update(t0 + Duration::from_millis(20));
        // Only 2 events, under BURST_THRESHOLD of 3
        assert!(!tracker.extended);
        assert_eq!(tracker.effective_debounce_ms(), 200);
    }

    // --- Plan 02: Path normalization tests ---

    #[test]
    #[cfg(windows)]
    fn test_normalize_event_path_basic() {
        // Windows-style absolute path: strip root prefix, normalize slashes
        let abs = Path::new(r"C:\repo\src\main.rs");
        let root = Path::new(r"C:\repo");
        let result = normalize_event_path(abs, root);
        assert_eq!(result, Some("src/main.rs".to_string()));
    }

    #[test]
    #[cfg(windows)]
    fn test_normalize_event_path_unc_prefix() {
        // Windows extended-length path with \\?\ prefix
        let abs = Path::new(r"\\?\C:\repo\src\main.rs");
        let root = Path::new(r"C:\repo");
        let result = normalize_event_path(abs, root);
        assert_eq!(result, Some("src/main.rs".to_string()));
    }

    #[test]
    #[cfg(windows)]
    fn test_normalize_event_path_outside_repo() {
        // Path is completely outside the repo root — should return None
        let abs = Path::new(r"C:\other\file.rs");
        let root = Path::new(r"C:\repo");
        let result = normalize_event_path(abs, root);
        assert_eq!(result, None);
    }

    #[test]
    fn test_normalize_event_path_forward_slash() {
        // Forward-slash paths (Linux/macOS) should also work
        let abs = Path::new("/home/user/project/src/lib.rs");
        let root = Path::new("/home/user/project");
        let result = normalize_event_path(abs, root);
        assert_eq!(result, Some("src/lib.rs".to_string()));
    }

    #[test]
    fn test_normalize_event_path_nested_subdir() {
        let abs = Path::new("/repo/a/b/c.rs");
        let root = Path::new("/repo");
        let result = normalize_event_path(abs, root);
        assert_eq!(result, Some("a/b/c.rs".to_string()));
    }

    // --- Plan 02: Language filter tests ---

    #[test]
    fn test_supported_language_rs() {
        let path = Path::new("src/main.rs");
        assert_eq!(supported_language(path), Some(LanguageId::Rust));
    }

    #[test]
    fn test_supported_language_py() {
        let path = Path::new("scripts/build.py");
        assert_eq!(supported_language(path), Some(LanguageId::Python));
    }

    #[test]
    fn test_supported_language_ts() {
        let path = Path::new("src/app.ts");
        assert_eq!(supported_language(path), Some(LanguageId::TypeScript));
    }

    #[test]
    fn test_supported_language_go() {
        let path = Path::new("main.go");
        assert_eq!(supported_language(path), Some(LanguageId::Go));
    }

    #[test]
    fn test_supported_language_java() {
        let path = Path::new("Main.java");
        assert_eq!(supported_language(path), Some(LanguageId::Java));
    }

    #[test]
    fn test_supported_language_txt() {
        let path = Path::new("README.txt");
        assert_eq!(supported_language(path), None);
    }

    #[test]
    fn test_supported_language_md() {
        let path = Path::new("README.md");
        assert_eq!(supported_language(path), Some(LanguageId::Markdown));
    }

    #[test]
    fn test_supported_language_no_extension() {
        let path = Path::new("Makefile");
        assert_eq!(supported_language(path), None);
    }

    // --- Plan 04-02: watcher incremental xref update (XREF-08) ---

    /// Proves that after `maybe_reindex` re-parses a file, the reverse_index
    /// reflects the new references and the old references are gone.
    ///
    /// We write a Rust file with an initial function call, confirm the reverse
    /// index contains it, then overwrite the file with a different call, call
    /// maybe_reindex again, and confirm the index now reflects the new call.
    #[test]
    fn test_maybe_reindex_updates_reverse_index_on_change() {
        use crate::domain::LanguageId;
        use crate::live_index::store::IndexedFile;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let rs_path = tmp
            .path()
            .join("tests")
            .join("generated")
            .join("lib.generated.rs");
        std::fs::create_dir_all(rs_path.parent().unwrap()).unwrap();

        // --- Initial content: calls `old_function` ---
        let initial_content = b"fn entry() { old_function(); }";
        std::fs::write(&rs_path, initial_content).unwrap();

        // Build the initial shared index by parsing the file directly.
        let rel_path = "tests/generated/lib.generated.rs";
        let shared: crate::live_index::store::SharedIndex = {
            let result = crate::parsing::process_file(rel_path, initial_content, LanguageId::Rust);
            let indexed = IndexedFile::from_parse_result(result, initial_content.to_vec());
            let mut index = crate::live_index::store::LiveIndex {
                files: std::collections::HashMap::new(),
                loaded_at: std::time::Instant::now(),
                loaded_at_system: std::time::SystemTime::now(),
                load_duration: std::time::Duration::ZERO,
                cb_state: crate::live_index::store::CircuitBreakerState::new(0.20),
                is_empty: false,
                load_source: crate::live_index::store::IndexLoadSource::FreshLoad,
                snapshot_verify_state: crate::live_index::store::SnapshotVerifyState::NotNeeded,
                reverse_index: std::collections::HashMap::new(),
                files_by_basename: std::collections::HashMap::new(),
                files_by_dir_component: std::collections::HashMap::new(),
                trigram_index: crate::live_index::trigram::TrigramIndex::new(),
                gitignore: None,
                skipped_files: Vec::new(),
                coupling_store: None,
                local_empty_reason: std::sync::Arc::new(parking_lot::RwLock::new(None)),
                indexed_root: None,
            };
            index.update_file(rel_path.to_string(), indexed);
            crate::live_index::SharedIndexHandle::shared(index)
        };

        // Confirm the reverse index contains "old_function".
        {
            let idx = shared.read();
            assert!(
                idx.reverse_index.contains_key("old_function"),
                "reverse_index should contain 'old_function' after initial parse"
            );
        }

        // --- Updated content: calls `new_function` instead ---
        let updated_content = b"fn entry() { new_function(); }";
        std::fs::write(&rs_path, updated_content).unwrap();

        // maybe_reindex detects a hash change and re-parses.
        let expected_gen = shared.current_project_generation();
        let result = maybe_reindex(rel_path, &rs_path, &shared, LanguageId::Rust, expected_gen);
        assert_eq!(
            result,
            ReindexResult::Reindexed,
            "file should be re-parsed on content change"
        );

        // Confirm reverse index now has "new_function" and not "old_function".
        {
            let idx = shared.read();
            assert!(
                idx.reverse_index.contains_key("new_function"),
                "reverse_index should contain 'new_function' after re-index"
            );
            assert!(
                !idx.reverse_index.contains_key("old_function"),
                "reverse_index should NOT contain 'old_function' after re-index"
            );
            let file = idx
                .get_file(rel_path)
                .expect("reindexed file should still exist");
            assert!(file.classification.is_code());
            assert!(file.classification.is_test);
            assert!(file.classification.is_generated);
        }
    }

    #[test]
    fn process_events_predicate_skips_gitignored_state_dir() {
        use crate::live_index::store::LiveIndex;
        use ignore::gitignore::GitignoreBuilder;

        // Reproduce SymForge's own root ignore rules: ignore every root-level dot
        // directory but explicitly re-include `.github` (as the repo's .gitignore
        // does via `/.*/` + `!/.github/`).
        let mut builder = GitignoreBuilder::new("/repo");
        builder.add_line(None, "/.*/").unwrap();
        builder.add_line(None, "!/.github/").unwrap();
        let gitignore = builder.build().unwrap();

        let index = LiveIndex {
            files: std::collections::HashMap::new(),
            loaded_at: std::time::Instant::now(),
            loaded_at_system: std::time::SystemTime::now(),
            load_duration: std::time::Duration::ZERO,
            cb_state: crate::live_index::store::CircuitBreakerState::new(0.20),
            is_empty: false,
            load_source: crate::live_index::store::IndexLoadSource::FreshLoad,
            snapshot_verify_state: crate::live_index::store::SnapshotVerifyState::NotNeeded,
            reverse_index: std::collections::HashMap::new(),
            files_by_basename: std::collections::HashMap::new(),
            files_by_dir_component: std::collections::HashMap::new(),
            trigram_index: crate::live_index::trigram::TrigramIndex::new(),
            gitignore: Some(gitignore),
            skipped_files: Vec::new(),
            coupling_store: None,
            local_empty_reason: std::sync::Arc::new(parking_lot::RwLock::new(None)),
            indexed_root: None,
        };

        // SymForge's own gitignored state dir must never be indexed, even though
        // tee snapshots are `.rs` files with a supported language.
        assert!(index.is_path_gitignored(".symforge/tee/1780038581944-000040-handlers.rs"));
        assert!(index.is_path_gitignored(".claude/settings.local.json"));
        // Real source, whitelisted `.github`, and committed `vendor/` stay indexable.
        assert!(!index.is_path_gitignored("src/sidecar/handlers.rs"));
        assert!(!index.is_path_gitignored(".github/workflows/ci.yml"));
        assert!(!index.is_path_gitignored("vendor/tree-sitter-scss/src/parser.c"));
        // Absolute paths are rejected defensively (the `ignore` crate requires
        // relative paths).
        assert!(!index.is_path_gitignored("/abs/path.rs"));
    }

    /// Confirms that maybe_reindex returns HashSkip when content has not changed.
    #[test]
    fn test_maybe_reindex_hash_skip_on_unchanged_content() {
        use crate::domain::LanguageId;
        use crate::live_index::store::IndexedFile;
        use tempfile::TempDir;

        let tmp = TempDir::new().unwrap();
        let rs_path = tmp.path().join("a.rs");
        let content = b"fn foo() {}";
        std::fs::write(&rs_path, content).unwrap();

        let rel_path = "a.rs";
        let shared: crate::live_index::store::SharedIndex = {
            let result = crate::parsing::process_file(rel_path, content, LanguageId::Rust);
            let indexed = IndexedFile::from_parse_result(result, content.to_vec());
            let mut index = crate::live_index::store::LiveIndex {
                files: std::collections::HashMap::new(),
                loaded_at: std::time::Instant::now(),
                loaded_at_system: std::time::SystemTime::now(),
                load_duration: std::time::Duration::ZERO,
                cb_state: crate::live_index::store::CircuitBreakerState::new(0.20),
                is_empty: false,
                load_source: crate::live_index::store::IndexLoadSource::FreshLoad,
                snapshot_verify_state: crate::live_index::store::SnapshotVerifyState::NotNeeded,
                reverse_index: std::collections::HashMap::new(),
                files_by_basename: std::collections::HashMap::new(),
                files_by_dir_component: std::collections::HashMap::new(),
                trigram_index: crate::live_index::trigram::TrigramIndex::new(),
                gitignore: None,
                skipped_files: Vec::new(),
                coupling_store: None,
                local_empty_reason: std::sync::Arc::new(parking_lot::RwLock::new(None)),
                indexed_root: None,
            };
            index.update_file(rel_path.to_string(), indexed);
            crate::live_index::SharedIndexHandle::shared(index)
        };

        // File content unchanged — expect HashSkip.
        let expected_gen = shared.current_project_generation();
        let result = maybe_reindex(rel_path, &rs_path, &shared, LanguageId::Rust, expected_gen);
        assert_eq!(
            result,
            ReindexResult::HashSkip,
            "unchanged content should produce HashSkip"
        );
    }

    #[test]
    fn test_read_and_index_preserves_crlf_bytes_and_hash() {
        use crate::domain::LanguageId;

        let tmp = TempDir::new().unwrap();
        let rs_path = tmp.path().join("src").join("lib.rs");
        std::fs::create_dir_all(rs_path.parent().unwrap()).unwrap();
        let content = b"fn entry() {\r\n    watched_call();\r\n}\r\n";
        std::fs::write(&rs_path, content).unwrap();

        let rel_path = "src/lib.rs";
        let shared: crate::live_index::store::SharedIndex = {
            let index = crate::live_index::store::LiveIndex {
                files: std::collections::HashMap::new(),
                loaded_at: std::time::Instant::now(),
                loaded_at_system: std::time::SystemTime::now(),
                load_duration: std::time::Duration::ZERO,
                cb_state: crate::live_index::store::CircuitBreakerState::new(0.20),
                is_empty: false,
                load_source: crate::live_index::store::IndexLoadSource::FreshLoad,
                snapshot_verify_state: crate::live_index::store::SnapshotVerifyState::NotNeeded,
                reverse_index: std::collections::HashMap::new(),
                files_by_basename: std::collections::HashMap::new(),
                files_by_dir_component: std::collections::HashMap::new(),
                trigram_index: crate::live_index::trigram::TrigramIndex::new(),
                gitignore: None,
                skipped_files: Vec::new(),
                coupling_store: None,
                local_empty_reason: std::sync::Arc::new(parking_lot::RwLock::new(None)),
                indexed_root: None,
            };
            crate::live_index::SharedIndexHandle::shared(index)
        };

        let expected_gen = shared.current_project_generation();
        let result = read_and_index(rel_path, &rs_path, &shared, LanguageId::Rust, expected_gen);
        assert_eq!(result, ReindexResult::Reindexed);

        let idx = shared.read();
        let file = idx
            .get_file(rel_path)
            .expect("watcher should index the CRLF file");
        assert_eq!(file.content, content);
        assert_eq!(file.byte_len, content.len() as u64);
        assert_eq!(file.content_hash, crate::hash::digest_hex(content));
    }

    #[test]
    fn test_maybe_reindex_retries_transient_not_found() {
        let tmp = TempDir::new().unwrap();
        let src_dir = tmp.path().join("src");
        std::fs::create_dir_all(&src_dir).unwrap();
        let rel_path = "src/flaky.rs";
        let abs_path = tmp.path().join(rel_path);
        let content = b"fn flaky() -> usize { 1 }";
        std::fs::write(&abs_path, content).unwrap();

        let shared = crate::live_index::LiveIndex::load(tmp.path()).unwrap();
        let expected_gen = shared.current_project_generation();
        std::fs::remove_file(&abs_path).unwrap();

        let restore_path = abs_path.clone();
        let restore = std::thread::spawn(move || {
            std::thread::sleep(std::time::Duration::from_millis(20));
            std::fs::write(&restore_path, content).unwrap();
        });

        let result = maybe_reindex(rel_path, &abs_path, &shared, LanguageId::Rust, expected_gen);
        restore.join().unwrap();

        assert_ne!(
            result,
            ReindexResult::Removed,
            "transient NotFound should be retried instead of removed immediately"
        );
        let index = shared.read();
        assert!(
            index.get_file(rel_path).is_some(),
            "transiently missing file should remain indexed after retry succeeds"
        );
    }

    #[test]
    fn test_maybe_reindex_removes_persistent_not_found() {
        let tmp = TempDir::new().unwrap();
        let src_dir = tmp.path().join("src");
        std::fs::create_dir_all(&src_dir).unwrap();
        let rel_path = "src/deleted.rs";
        let abs_path = tmp.path().join(rel_path);
        std::fs::write(&abs_path, b"fn deleted() {}").unwrap();

        let shared = crate::live_index::LiveIndex::load(tmp.path()).unwrap();
        let expected_gen = shared.current_project_generation();
        std::fs::remove_file(&abs_path).unwrap();

        let result = maybe_reindex(rel_path, &abs_path, &shared, LanguageId::Rust, expected_gen);

        assert_eq!(
            result,
            ReindexResult::Removed,
            "persistent NotFound should remove after bounded retries"
        );
        let index = shared.read();
        assert!(
            index.get_file(rel_path).is_none(),
            "persistently missing file should be removed from the index"
        );
    }

    #[test]
    fn slipped_past_cancellation_fence_increments_counter() {
        let project_a = TempDir::new().unwrap();
        let project_b = TempDir::new().unwrap();
        let a_src = project_a.path().join("src");
        let b_src = project_b.path().join("src");
        std::fs::create_dir_all(&a_src).unwrap();
        std::fs::create_dir_all(&b_src).unwrap();
        std::fs::write(a_src.join("a.rs"), b"fn a() {}").unwrap();
        std::fs::write(b_src.join("b.rs"), b"fn b() {}").unwrap();

        let shared = crate::live_index::LiveIndex::load(project_a.path()).unwrap();
        let stale_gen = shared.current_project_generation();
        let rejected_before = shared.current_rejected_stale_mutations();
        shared.reload(project_b.path()).unwrap();

        let stale = reconcile_stale_files_with_stop(project_a.path(), &shared, || false, stale_gen);

        assert_eq!(
            stale, 1,
            "stale doomed reconcile should visit B's path once"
        );
        assert!(
            shared.current_rejected_stale_mutations() > rejected_before,
            "stale-generation watcher reconcile should be rejected by the fence"
        );
        let index = shared.read();
        assert!(
            index.get_file("src/b.rs").is_some(),
            "B file should survive stale-generation reconcile"
        );
    }
}
