"""
File system watcher for manual file additions/modifications.

Watches output and custom root directories for files added outside of ComfyUI
(e.g., manual copies, external tools). Does NOT conflict with entry.js which
handles ComfyUI executed events.
"""
import asyncio
import os
import time
from collections import deque
from collections.abc import Awaitable, Callable, Coroutine
from pathlib import Path
from threading import Lock, RLock
from typing import (
    Any,
    cast,
)

from watchdog.events import (
    DirCreatedEvent,
    FileCreatedEvent,
    FileDeletedEvent,
    FileMovedEvent,
    FileSystemEventHandler,
)
from watchdog.observers import Observer

from ...config import (
    WATCHER_DEBOUNCE_MS,
    WATCHER_DEDUPE_TTL_MS,
    WATCHER_FLUSH_MAX_FILES,
    WATCHER_GENERATED_GRACE_SECONDS,
    WATCHER_MAX_FILE_SIZE_BYTES,
    WATCHER_MAX_FLUSH_CONCURRENCY,
    WATCHER_MIN_FILE_SIZE_BYTES,
    WATCHER_PENDING_MAX,
    WATCHER_STREAM_ALERT_COOLDOWN_SECONDS,
    WATCHER_STREAM_ALERT_THRESHOLD,
    WATCHER_STREAM_ALERT_WINDOW_SECONDS,
)
from ...shared import EXTENSIONS, get_logger
from ..watcher_settings import get_watcher_settings

logger = get_logger(__name__)

# Extensions explicitly excluded even if they ever appear in EXTENSIONS
EXCLUDED_EXTENSIONS: set[str] = {".psd", ".json", ".txt", ".csv", ".db", ".sqlite", ".log"}

# Supported extensions (flattened)
SUPPORTED_EXTENSIONS: set[str] = set()
try:
    for kind in ("image", "video", "audio", "model3d"):
        for ext in (EXTENSIONS or {}).get(kind, []):
            ext_lower = str(ext).lower()
            if ext_lower not in EXCLUDED_EXTENSIONS:
                SUPPORTED_EXTENSIONS.add(ext_lower)
except Exception:
    SUPPORTED_EXTENSIONS = {
        ".png",
        ".jpg",
        ".jpeg",
        ".webp",
        ".gif",
        ".avif",
        ".jxl",
        ".mp4",
        ".mov",
        ".webm",
        ".mkv",
        ".wav",
        ".mp3",
        ".flac",
        ".ogg",
        ".m4a",
        ".aac",
        ".obj",
        ".fbx",
        ".glb",
        ".gltf",
        ".stl",
        ".ply",
        ".splat",
        ".ksplat",
        ".spz",
    }


def _is_supported_extension(ext: str) -> bool:
    if ext == ".jxl" and str(os.getenv("MAJOOR_ENABLE_JXL", "")).strip().lower() not in {"1", "true", "yes", "on"}:
        return False
    return ext in SUPPORTED_EXTENSIONS

# Directories to always ignore (case-insensitive on Windows)
IGNORED_DIRS = {
    "_mjr_index",
    ".majoor",
    "__pycache__",
    ".git",
    "node_modules",
    ".thumbs",
    ".cache",
}

# Minimum file size (bytes) to avoid indexing partial/temp writes
MIN_FILE_SIZE = max(0, int(WATCHER_MIN_FILE_SIZE_BYTES))
# Maximum file size (bytes) to avoid indexing oversized files
MAX_FILE_SIZE = max(0, int(WATCHER_MAX_FILE_SIZE_BYTES))
# Timing assumption (HIGH-005):
#
# The _RECENT_GENERATED dict prevents the watcher from re-indexing files that
# ComfyUI just generated (which are already indexed via the executed-event path).
#
# Sequence:
#   T=0ms    entry.js receives WS "executed" event
#   T~50ms   POST /index-files → scan.py calls mark_recent_generated(paths) BEFORE index_paths()
#   T~3000ms watcher debounce fires → _flush() → _is_recent_generated() returns True → skip
#
# The 3000ms default debounce (WATCHER_DEFAULT_DEBOUNCE_MS) guarantees that the
# mark always precedes the flush check.  If the debounce is ever reduced below
# ~100ms this assumption may break and a lock-based coordination would be needed.
# The TTL of 30s provides ample margin for slow index_paths() runs.
_RECENT_GENERATED_TTL_S = 30.0
_RECENT_GENERATED_LOCK = RLock()  # RLock allows re-entrant acquisition if callers are ever composed
_RECENT_GENERATED: dict[str, float] = {}
_MAX_PENDING_FILES = max(0, WATCHER_PENDING_MAX)
_STREAM_EVENTS: deque[tuple[float, int]] = deque()
_STREAM_TOTAL_FILES = 0
_STREAM_LOCK = Lock()
_LAST_STREAM_ALERT_TIME = 0.0
_PENDING_LIMIT_WARN_INTERVAL = 30.0


class DebouncedWatchHandler(FileSystemEventHandler):
    """
    Handles file system events with deduplication and debouncing.

    - Ignores non-media files and internal directories
    - Deduplicates rapid events for the same file
    - Batches files before triggering indexing
    - Only reacts to file creation (not modification noise)
    """

    def __init__(
        self,
        on_files_ready: Callable[[list], Awaitable[None]],
        on_files_removed: Callable[[list], Awaitable[None]] | None,
        on_files_moved: Callable[[list], Awaitable[None]] | None,
        loop: asyncio.AbstractEventLoop,
        debounce_ms: int = 500,
        dedupe_ttl_ms: int = 3000,
        flush_concurrency: int = 1,
    ):
        super().__init__()
        self._on_files_ready = on_files_ready
        self._on_files_removed = on_files_removed
        self._on_files_moved = on_files_moved
        self._loop = loop
        self._debounce_s = debounce_ms / 1000.0
        self._dedupe_ttl_s = dedupe_ttl_ms / 1000.0

        self._lock = Lock()
        self._pending: dict[str, float] = {}  # filepath -> timestamp
        self._overflow: dict[str, float] = {}  # overflow queue when pending cap is reached
        self._recent: dict[str, float] = {}   # filepath -> last processed timestamp
        self._flush_timer: asyncio.TimerHandle | None = None
        concurrency = max(1, flush_concurrency or 1)
        self._flush_semaphore = asyncio.Semaphore(concurrency)
        self._last_pending_warning = 0.0
        self._last_overflow_drop_warning = 0.0
        self._dropped_count = 0
        self._refresh_runtime_settings()

    def _is_ignored_path(self, path: str) -> bool:
        """Check if path is inside an ignored directory or is a hidden file."""
        try:
            parts = Path(path).parts
            for part in parts:
                if part.lower() in IGNORED_DIRS:
                    return True
            # Apply the dot-prefix check only to the filename itself, not to parent
            # directory components.  A blanket check on all parts would silently ignore
            # files whose configured output root contains a dot-segment (e.g.
            # ~/.config/ComfyUI/outputs/ on Linux/macOS) — MED-04.
            if parts and parts[-1].startswith("."):
                return True
        except Exception:
            pass
        return False

    def _is_supported(self, path: str) -> bool:
        """Check if file has a supported extension."""
        try:
            ext = Path(path).suffix.lower()
            return _is_supported_extension(ext)
        except Exception:
            return False

    def _normalize_path(self, path: str) -> str:
        """Normalize path for deduplication."""
        try:
            return os.path.normcase(os.path.normpath(path))
        except Exception:
            return path

    def _refresh_runtime_settings(self) -> None:
        try:
            settings = get_watcher_settings()
            self._debounce_s = max(0.05, settings.debounce_ms / 1000.0)
            self._dedupe_ttl_s = max(0.1, settings.dedupe_ttl_ms / 1000.0)
        except Exception:
            pass

    def refresh_runtime_settings(self) -> None:
        """Expose runtime tuning for external callers."""
        self._refresh_runtime_settings()

    def flush_pending(self) -> bool:
        """
        Trigger an immediate flush of pending files.

        Returns True when a flush task was successfully scheduled.
        """
        try:
            if self._flush_timer:
                self._flush_timer.cancel()
                self._flush_timer = None
            asyncio.run_coroutine_threadsafe(self._flush(), self._loop)
            return True
        except Exception:
            return False

    def get_pending_count(self) -> int:
        """Return pending file count in debounce queue."""
        try:
            with self._lock:
                return int(len(self._pending) + len(self._overflow))
        except Exception:
            return 0

    def _maybe_log_pending_limit(self, path: str) -> None:
        now = time.time()
        if _MAX_PENDING_FILES <= 0:
            return
        if (now - self._last_pending_warning) < _PENDING_LIMIT_WARN_INTERVAL:
            return
        try:
            logger.warning(
                "Watcher pending queue capped at %d entries; deferring %s",
                _MAX_PENDING_FILES,
                path,
            )
        except Exception:
            pass
        self._last_pending_warning = now

    def _maybe_log_overflow_drop(self, path: str) -> None:
        now = time.time()
        if (now - self._last_overflow_drop_warning) < _PENDING_LIMIT_WARN_INTERVAL:
            return
        try:
            logger.warning(
                "Watcher overflow queue full (%d); file dropped: %s (total dropped: %d). "
                "Files will be picked up on next scan.",
                _MAX_PENDING_FILES,
                path,
                self._dropped_count,
            )
        except Exception:
            pass
        self._last_overflow_drop_warning = now

    def on_created(self, event):
        if isinstance(event, DirCreatedEvent):
            return
        if not isinstance(event, FileCreatedEvent):
            return
        self._handle_file(str(event.src_path))

    def on_modified(self, event):
        # Modified events are needed for slow/network writes where the create event
        # happens before file content is fully available.
        try:
            if getattr(event, "is_directory", False):
                return
        except Exception:
            return
        self._handle_file(str(getattr(event, "src_path", "")))

    def on_deleted(self, event):
        if isinstance(event, FileDeletedEvent):
            self._handle_deleted_file(str(event.src_path))

    def on_moved(self, event):
        if isinstance(event, FileMovedEvent):
            self._handle_moved_file(str(event.src_path), str(event.dest_path))

    def _handle_file(self, path: str):
        """Queue a file for indexing with deduplication."""
        if not path:
            return

        # Fast rejections — settings refresh happens inside _schedule_flush (MED-03: was double-called)
        if self._is_ignored_path(path):
            return
        if not self._is_supported(path):
            return
        # NOTE: _is_recent_generated is checked at flush time (not here) to avoid
        # a race condition where the watcher detects a file before the frontend
        # executed-event handler has called mark_recent_generated().

        key = self._normalize_path(path)
        now = time.time()

        with self._lock:
            # Skip if recently processed
            last = self._recent.get(key, 0)
            if last and (now - last) < self._dedupe_ttl_s:
                return

            # Prevent unbounded memory growth under bursts.
            if _MAX_PENDING_FILES > 0 and len(self._pending) >= _MAX_PENDING_FILES:
                # Keep overflow candidates for later flush instead of dropping permanently.
                # Cap overflow at the same limit so it cannot grow without bound when
                # pending stays full and events arrive faster than they are flushed.
                if key not in self._overflow and len(self._overflow) < _MAX_PENDING_FILES:
                    self._overflow[key] = now
                elif key not in self._overflow:
                    self._dropped_count += 1
                    self._maybe_log_overflow_drop(path)
                self._maybe_log_pending_limit(path)
                self._schedule_flush()
                return

            # Add to pending batch and mark in _recent immediately to prevent
            # duplicate events from bypassing dedup between pending-add and flush.
            self._pending[key] = now
            self._recent[key] = now

            # Schedule flush
            self._schedule_flush()

    def _schedule_flush(self):
        """Schedule a debounced flush of pending files.

        This method is called from the watchdog background thread, so it must
        not touch the event loop directly.  ``call_soon_threadsafe`` safely
        hands off to the event loop thread, which then arms the debounce timer
        via ``call_later`` (safe because it runs inside the loop).
        """
        try:
            self._refresh_runtime_settings()
            debounce_s = self._debounce_s

            def _do_schedule() -> None:
                if self._flush_timer:
                    self._flush_timer.cancel()
                self._flush_timer = self._loop.call_later(
                    debounce_s,
                    lambda: self._loop.create_task(self._flush()),
                )

            self._loop.call_soon_threadsafe(_do_schedule)
        except Exception:
            pass

    def _handle_deleted_file(self, path: str):
        if not path:
            return
        if self._is_ignored_path(path):
            return
        if not self._is_supported(path):
            return
        if _is_recent_generated(path):
            return
        if not self._on_files_removed:
            return
        key = self._normalize_path(path)
        try:
            coro = cast(Coroutine[Any, Any, None], self._on_files_removed([key]))
            asyncio.run_coroutine_threadsafe(coro, self._loop)
        except Exception:
            return

    def _handle_moved_file(self, src_path: str, dest_path: str):
        if not src_path or not dest_path:
            return
        if self._is_ignored_path(src_path) or self._is_ignored_path(dest_path):
            return
        src_ok = self._is_supported(src_path)
        dst_ok = self._is_supported(dest_path)

        src_norm = self._normalize_path(src_path)
        dst_norm = self._normalize_path(dest_path)
        mode = self._moved_file_mode(src_ok=src_ok, dst_ok=dst_ok)
        if mode == "ignore":
            return
        if mode == "delete":
            self._emit_removed_files([src_norm])
            return
        if mode == "create":
            self._handle_file(dst_norm)
            return
        if mode == "move" and self._emit_moved_files([(src_norm, dst_norm)]):
            return

        # Fallback: index new path and remove old path as separate operations.
        self._handle_file(dst_norm)
        self._emit_removed_files([src_norm])

    @staticmethod
    def _moved_file_mode(*, src_ok: bool, dst_ok: bool) -> str:
        if not src_ok and not dst_ok:
            return "ignore"
        if src_ok and not dst_ok:
            return "delete"
        if dst_ok and not src_ok:
            return "create"
        return "move"

    def _emit_removed_files(self, paths: list[str]) -> None:
        if not self._on_files_removed:
            return
        try:
            coro = cast(Coroutine[Any, Any, None], self._on_files_removed(paths))
            asyncio.run_coroutine_threadsafe(coro, self._loop)
        except Exception:
            return

    def _emit_moved_files(self, moves: list[tuple[str, str]]) -> bool:
        if not self._on_files_moved:
            return False
        try:
            coro = cast(Coroutine[Any, Any, None], self._on_files_moved(moves))
            asyncio.run_coroutine_threadsafe(coro, self._loop)
            return True
        except Exception:
            return False

    async def _flush(self):
        """Flush pending files to the indexer."""
        self._refresh_runtime_settings()
        candidates = self._drain_pending_candidates()
        if not candidates:
            return
        # NOTE: We deliberately do NOT defer the entire flush during ComfyUI
        # generation. Manual file additions, external tool drops, and copy
        # operations must be indexed in real time even while a prompt is
        # running. ComfyUI-generated files are still skipped per-file via
        # `_is_recent_generated()` inside `_filter_flush_candidates`, so we
        # cannot double-index. The previous global early-return blocked all
        # external/manual files and broke the user's expectation that the
        # watcher remains live during execution.

        async with self._flush_semaphore:
            _record_flush_volume(len(candidates))
            files = self._filter_flush_candidates(candidates)
            if not files:
                return

            files = self._apply_flush_backpressure(files)
            self._mark_recent_files(files)
            try:
                await self._on_files_ready(files)
            except Exception as e:
                logger.warning(
                    "Watcher index callback failed — %d file(s) re-queued for retry: %s",
                    len(files), e,
                )
                self._requeue_failed_files(files)

    def _requeue_failed_files(self, files: list[str]) -> None:
        """Re-queue files that the index callback failed to process into overflow."""
        now = time.time()
        dropped = 0
        with self._lock:
            for f in files:
                if _MAX_PENDING_FILES <= 0 or len(self._overflow) < _MAX_PENDING_FILES:
                    self._overflow[f] = now
                else:
                    dropped += 1
        if dropped:
            logger.warning(
                "Watcher: %d failed files could not be re-queued (overflow full at %d)",
                dropped,
                _MAX_PENDING_FILES,
            )

    def _requeue_runtime_deferred_files(self, files: list[str]) -> None:
        now = time.time()
        with self._lock:
            for f in files:
                if not f:
                    continue
                if f not in self._pending and f not in self._overflow:
                    self._overflow[f] = now
            self._schedule_flush()

    def _drain_pending_candidates(self) -> list[str]:
        with self._lock:
            if not self._pending and not self._overflow:
                return []
            candidates = list(self._pending.keys())
            if self._overflow:
                candidates.extend(list(self._overflow.keys()))
            self._pending.clear()
            self._overflow.clear()
            self._flush_timer = None
            return candidates

    def _filter_flush_candidates(self, candidates: list[str]) -> list[str]:
        files: list[str] = []
        deferred_recent: list[str] = []
        for f in candidates:
            if _is_recent_generated(f):
                continue
            if not self._is_supported_flush_file(f):
                continue
            if self._should_defer_recent_file(f):
                deferred_recent.append(f)
                continue
            size = self._flush_file_size_if_eligible(f)
            if size is None:
                continue
            max_size = MAX_FILE_SIZE
            if max_size > 0 and size > max_size:
                logger.debug("Watcher skipping oversized file %s (%d bytes > %d)", f, size, max_size)
                continue
            files.append(f)
        if deferred_recent:
            self._requeue_deferred_recent_files(deferred_recent)
        return files

    @staticmethod
    def _should_defer_recent_file(filepath: str) -> bool:
        grace_s = max(0.0, float(WATCHER_GENERATED_GRACE_SECONDS or 0.0))
        if grace_s <= 0.0:
            return False
        try:
            stat = os.stat(filepath)
        except OSError:
            return False
        now = time.time()
        latest_ts = max(
            float(getattr(stat, "st_mtime", 0.0) or 0.0),
            float(getattr(stat, "st_ctime", 0.0) or 0.0),
        )
        if latest_ts <= 0.0:
            return False
        return (now - latest_ts) < grace_s

    def _requeue_deferred_recent_files(self, files: list[str]) -> None:
        now = time.time()
        with self._lock:
            for f in files:
                if f not in self._pending and f not in self._overflow:
                    self._overflow[f] = now
            self._schedule_flush()

    @staticmethod
    def _is_supported_flush_file(filepath: str) -> bool:
        try:
            ext = Path(filepath).suffix.lower()
            return ext not in EXCLUDED_EXTENSIONS and _is_supported_extension(ext)
        except Exception:
            return False

    @staticmethod
    def _flush_file_size_if_eligible(filepath: str) -> int | None:
        try:
            size = os.path.getsize(filepath)
        except OSError:
            return None
        if size < MIN_FILE_SIZE:
            return None
        return size

    def _apply_flush_backpressure(self, files: list[str]) -> list[str]:
        if WATCHER_FLUSH_MAX_FILES <= 0 or len(files) <= WATCHER_FLUSH_MAX_FILES:
            return files
        now = time.time()
        deferred = files[WATCHER_FLUSH_MAX_FILES:]
        files = files[:WATCHER_FLUSH_MAX_FILES]
        with self._lock:
            for f in deferred:
                if f not in self._pending and f not in self._overflow:
                    self._overflow[f] = now
            self._schedule_flush()
        try:
            logger.warning(
                "Watcher flush capped to %d files (requested %d); deferring %d for next flush.",
                WATCHER_FLUSH_MAX_FILES,
                len(files) + len(deferred),
                len(deferred),
            )
        except Exception:
            pass
        return files

    def _mark_recent_files(self, files: list[str]) -> None:
        with self._lock:
            now = time.time()
            for f in files:
                self._recent[f] = now
            to_remove = [k for k, v in self._recent.items() if (now - v) > self._dedupe_ttl_s * 2]
            for k in to_remove:
                self._recent.pop(k, None)


def _apply_flush_limit(files: list[str]) -> list[str]:
    if WATCHER_FLUSH_MAX_FILES > 0:
        return files[:WATCHER_FLUSH_MAX_FILES]
    return files


def _normalize_recent_key(path: str) -> str:
    try:
        return os.path.normcase(os.path.normpath(path))
    except Exception:
        return str(path or "")


def _is_recent_generated(path: str) -> bool:
    key = _normalize_recent_key(path)
    if not key:
        return False
    try:
        with _RECENT_GENERATED_LOCK:
            now = time.time()
            _prune_recent_generated(now)
            return _is_recent_generated_key_locked(key, now)
    except Exception:
        return False


def _prune_recent_generated(now: float) -> None:
    if not _RECENT_GENERATED:
        return
    to_remove = [k for k, ts in _RECENT_GENERATED.items() if (now - float(ts or 0)) > _RECENT_GENERATED_TTL_S]
    for key in to_remove:
        _RECENT_GENERATED.pop(key, None)


def _is_recent_generated_key_locked(key: str, now: float) -> bool:
    ts = _RECENT_GENERATED.get(key)
    if not ts:
        return False
    return (now - float(ts or 0)) <= _RECENT_GENERATED_TTL_S


def is_recent_generated(path: str) -> bool:
    return _is_recent_generated(path)


def _prune_expired_generated(now: float) -> None:
    cutoff = now - _RECENT_GENERATED_TTL_S
    expired = [k for k, ts in _RECENT_GENERATED.items() if ts < cutoff]
    for k in expired:
        _RECENT_GENERATED.pop(k, None)


def _prune_oldest_generated() -> None:
    for k, _ in sorted(_RECENT_GENERATED.items(), key=lambda kv: kv[1])[:1000]:
        _RECENT_GENERATED.pop(k, None)


_RECENT_GENERATED_HARD_CAP = 2000


def mark_recent_generated(paths: list[str]) -> None:
    if not paths:
        return
    now = time.time()
    try:
        with _RECENT_GENERATED_LOCK:
            # Prune *before* inserting so the cache never exceeds the hard cap.
            if len(_RECENT_GENERATED) >= _RECENT_GENERATED_HARD_CAP:
                _prune_expired_generated(now)
            if len(_RECENT_GENERATED) >= _RECENT_GENERATED_HARD_CAP:
                _prune_oldest_generated()
            for p in paths:
                key = _normalize_recent_key(p)
                if key:
                    _RECENT_GENERATED[key] = now
    except Exception:
        return


def _record_flush_volume(count: int) -> None:
    global _STREAM_TOTAL_FILES, _LAST_STREAM_ALERT_TIME
    if count <= 0 or WATCHER_STREAM_ALERT_THRESHOLD <= 0 or WATCHER_STREAM_ALERT_WINDOW_SECONDS <= 0:
        return
    now = time.time()
    window = WATCHER_STREAM_ALERT_WINDOW_SECONDS
    with _STREAM_LOCK:
        if _MAX_PENDING_FILES > 0 and len(_STREAM_EVENTS) >= _MAX_PENDING_FILES:
            _prune_stream_events(now, window)
            if len(_STREAM_EVENTS) >= _MAX_PENDING_FILES:
                return
        _STREAM_EVENTS.append((now, count))
        _STREAM_TOTAL_FILES += count
        _prune_stream_events(now, window)
        if not _should_emit_stream_alert(now, _LAST_STREAM_ALERT_TIME):
            return
        _LAST_STREAM_ALERT_TIME = now
        try:
            logger.warning(
                "Watcher ingest rate high: %d files in the last %.1f seconds (threshold=%d).",
                _STREAM_TOTAL_FILES,
                window,
                WATCHER_STREAM_ALERT_THRESHOLD,
            )
        except Exception:
            pass


def _prune_stream_events(now: float, window: float) -> None:
    global _STREAM_TOTAL_FILES
    cutoff_time = now - window
    while _STREAM_EVENTS and _STREAM_EVENTS[0][0] < cutoff_time:
        _, stale = _STREAM_EVENTS.popleft()
        _STREAM_TOTAL_FILES -= stale
    if _STREAM_TOTAL_FILES < 0:
        _STREAM_TOTAL_FILES = 0


def _should_emit_stream_alert(now: float, last_alert_time: float) -> bool:
    if _STREAM_TOTAL_FILES < WATCHER_STREAM_ALERT_THRESHOLD:
        return False
    cooldown = max(WATCHER_STREAM_ALERT_COOLDOWN_SECONDS, 0.0)
    return cooldown <= 0 or (now - last_alert_time) >= cooldown


class OutputWatcher:
    """
    Watches output and custom root directories for manual file additions.

    Usage:
        watcher = OutputWatcher(index_callback)
        await watcher.start([{"path": "/path/to/output", "source": "output"}])
        ...
        await watcher.stop()
    """

    def __init__(
        self,
        index_callback: Callable[[list, str, str | None, str | None], Awaitable[None]],
        remove_callback: Callable[[list, str, str | None, str | None], Awaitable[None]] | None = None,
        move_callback: Callable[[list, str, str | None, str | None], Awaitable[None]] | None = None,
    ):
        """
        Args:
            index_callback: async function(filepaths, base_dir, source, root_id) to index files
        """
        self._index_callback = index_callback
        self._remove_callback = remove_callback
        self._move_callback = move_callback
        self._observer: Any | None = None
        self._handler: DebouncedWatchHandler | None = None
        self._watched_paths: dict[str, dict] = {}  # watch_key -> {path, source, root_id, watch}
        self._lock = Lock()
        self._start_lock = asyncio.Lock()  # prevents concurrent start() calls
        self._running = False
        self._allowed_sources: set[str] = set()

    async def start(self, paths: list, loop: asyncio.AbstractEventLoop | None = None):
        """Start watching the given directories.

        Protected by ``_start_lock`` so concurrent ``await watcher.start()``
        calls are serialised — only the first one does any work.
        """
        async with self._start_lock:
            await self._start_locked(paths, loop)

    async def _start_locked(self, paths: list, loop: asyncio.AbstractEventLoop | None = None) -> None:
        if self._running:
            return

        loop = loop or asyncio.get_running_loop()
        self._allowed_sources = set()
        on_files_ready = self._handle_ready_files
        on_files_removed = self._handle_removed_files
        on_files_moved = self._handle_moved_files

        self._handler = DebouncedWatchHandler(
            on_files_ready,
            on_files_removed,
            on_files_moved,
            loop,
            debounce_ms=WATCHER_DEBOUNCE_MS,
            dedupe_ttl_ms=WATCHER_DEDUPE_TTL_MS,
            flush_concurrency=WATCHER_MAX_FLUSH_CONCURRENCY,
        )
        self._observer = Observer()

        for path in paths:
            try:
                raw_path, source, root_id = self._resolve_watch_path(path)
                self._register_watch_path(raw_path, source=source, root_id=root_id, log_label="started")
            except Exception as e:
                logger.warning("Failed to watch %s: %s", path, e)

        if self._watched_paths:
            self._observer.start()
            self._running = True
            # Log which OS-level backend watchdog selected (inotify on Linux,
            # ReadDirectoryChangesW on Windows, FSEvents/kqueue on macOS).
            # watchdog.observers.Observer auto-selects the best native backend.
            try:
                backend = type(self._observer).__name__
                logger.info(
                    "File watcher started for %d directories (backend: %s)",
                    len(self._watched_paths),
                    backend,
                )
            except Exception:
                logger.info("File watcher started for %d directories", len(self._watched_paths))

    async def _handle_ready_files(self, files: list) -> None:
        if not files:
            return
        grouped = self._group_files_by_watched_root(files)
        await self._dispatch_file_groups(grouped, self._index_callback, "index")

    async def _handle_removed_files(self, files: list) -> None:
        if not files or not callable(self._remove_callback):
            return
        grouped = self._group_files_by_watched_root(files)
        await self._dispatch_file_groups(grouped, self._remove_callback, "remove")

    async def _handle_moved_files(self, moves: list) -> None:
        if not moves:
            return
        if callable(self._move_callback):
            grouped = self._group_moves_by_watched_root(moves)
            await self._dispatch_move_groups(grouped)
            await self._emit_removes_for_outbound_moves(moves, grouped)
            return
        await self._handle_move_fallback(moves)

    @staticmethod
    def _get_unmatched_move_src(move: Any, matched_srcs: set[str]) -> str | None:
        """Return the src path of a move whose destination is outside all watched roots."""
        try:
            src, _ = move
        except Exception:
            return None
        src_str = str(src)
        return src_str if src_str not in matched_srcs else None

    def _collect_outbound_move_removes(
        self, moves: list, matched_srcs: set[str]
    ) -> dict[str, dict]:
        """Build a remove-groups dict for move sources that left all watched roots."""
        removes: dict[str, dict] = {}
        for move in moves:
            src_str = self._get_unmatched_move_src(move, matched_srcs)
            if not src_str:
                continue
            entry = self._best_watched_entry_for_path(src_str)
            if not entry:
                continue
            base = str(entry.get("path") or "")
            if not base:
                continue
            removes.setdefault(
                base,
                {"files": [], "source": entry.get("source"), "root_id": entry.get("root_id")},
            )
            removes[base]["files"].append(src_str)
        return removes

    async def _emit_removes_for_outbound_moves(
        self, moves: list, grouped: dict[str, dict]
    ) -> None:
        """Emit remove events for move sources whose destination left all watched roots.

        Without this, a file moved *out* of a watched directory would leave a
        stale record in the index indefinitely.
        """
        if not self._remove_callback:
            return
        matched_srcs = {str(s) for p in grouped.values() for s, _ in p.get("moves", [])}
        removes = self._collect_outbound_move_removes(moves, matched_srcs)
        if removes:
            await self._dispatch_file_groups(removes, self._remove_callback, "remove-outbound-move")

    async def _dispatch_move_groups(self, grouped_moves: dict[str, dict]) -> None:
        if not callable(self._move_callback):
            return
        for base_dir, payload in grouped_moves.items():
            try:
                await self._move_callback(
                    payload.get("moves") or [],
                    base_dir,
                    payload.get("source"),
                    payload.get("root_id"),
                )
            except Exception as exc:
                logger.debug("Watcher move error: %s", exc)

    async def _handle_move_fallback(self, moves: list) -> None:
        # Fallback if no move callback: reindex destination and remove source.
        try:
            src_files, dst_files = self._split_move_pairs(moves)
            if dst_files:
                await self._handle_ready_files(dst_files)
            if src_files:
                await self._handle_removed_files(src_files)
        except Exception:
            return

    @staticmethod
    def _resolve_watch_path(path: Any) -> tuple[Any, str | None, str | None]:
        if isinstance(path, dict):
            return path.get("path"), path.get("source"), path.get("root_id")
        return path, None, None

    def _register_watch_path(
        self,
        raw_path: Any,
        *,
        source: str | None,
        root_id: str | None,
        log_label: str,
    ) -> bool:
        if not self._observer or not self._handler or not raw_path:
            return False
        normalized = os.path.normpath(raw_path)
        if not os.path.isdir(normalized):
            return False
        watch = self._observer.schedule(self._handler, normalized, recursive=True)
        source_norm = self._normalize_source(source)
        if source_norm:
            self._allowed_sources.add(source_norm)
        self._watched_paths[str(id(watch))] = {
            "path": normalized,
            "source": source_norm,
            "root_id": root_id,
            "watch": watch,
        }
        logger.info("Watcher %s for: %s", log_label, normalized)
        return True

    async def stop(self):
        """Stop watching all directories."""
        if not self._running:
            return

        try:
            if self._observer:
                self._observer.stop()
                self._observer.join(timeout=2)
                self._observer = None
        except Exception as e:
            logger.debug("Watcher stop error: %s", e)

        self._watched_paths.clear()
        self._handler = None
        self._running = False
        logger.info("File watcher stopped")

    def add_path(self, path: str, *, source: str | None = None, root_id: str | None = None):
        """Add a new path to watch (e.g., when custom root is added).

        Uses ``self._lock`` to prevent TOCTOU races when two callers check
        ``_is_already_watched`` concurrently and both proceed to ``schedule``.
        """
        if not self._running or not self._observer or not self._handler:
            return

        try:
            normalized = os.path.normpath(path)
            if not os.path.isdir(normalized):
                return
            with self._lock:
                if self._is_already_watched(normalized):
                    return
                source_norm = self._normalize_source(source)
                if not self._allows_source(source_norm):
                    logger.debug(
                        "Watcher ignored path (scope mismatch): %s (source=%s)",
                        normalized, source_norm,
                    )
                    return
                watch = self._observer.schedule(self._handler, normalized, recursive=True)
                self._watched_paths[str(id(watch))] = {
                    "path": normalized,
                    "source": source_norm,
                    "root_id": root_id,
                    "watch": watch,
                }
            logger.info("Watcher added: %s", normalized)
        except Exception as e:
            logger.warning("Failed to add watch for %s: %s", path, e)

    def _normalize_source(self, source: str | None) -> str | None:
        return str(source or "").strip().lower() if source is not None else None

    def _is_already_watched(self, normalized: str) -> bool:
        for entry in self._watched_paths.values():
            if entry.get("path") == normalized:
                return True
        return False

    def _best_watched_entry_for_path(self, path_value: str) -> dict | None:
        normalized_path = os.path.normcase(os.path.normpath(path_value))
        # Snapshot under lock to avoid RuntimeError if add_path/remove_path
        # modifies _watched_paths concurrently while we iterate.
        with self._lock:
            entries = list(self._watched_paths.values())
        best = None
        best_len = -1
        for entry in entries:
            watched = entry.get("path")
            if not watched:
                continue
            normalized_watched = os.path.normcase(os.path.normpath(watched))
            if _is_under_path(normalized_path, normalized_watched) and len(normalized_watched) > best_len:
                best = entry
                best_len = len(normalized_watched)
        return best

    def _group_files_by_watched_root(self, files: list[str]) -> dict[str, dict]:
        by_dir: dict[str, dict] = {}
        for file_path in files:
            try:
                best = self._best_watched_entry_for_path(file_path)
                if not best:
                    continue
                base = best.get("path")
                if not base:
                    continue
                key = str(base)
                if key not in by_dir:
                    by_dir[key] = {"files": [], "source": best.get("source"), "root_id": best.get("root_id")}
                by_dir[key]["files"].append(file_path)
            except Exception:
                continue
        return by_dir

    def _group_moves_by_watched_root(self, moves: list[Any]) -> dict[str, dict]:
        by_dir: dict[str, dict] = {}
        for move in moves:
            try:
                src, dst = move
            except Exception:
                continue
            best = self._best_watched_entry_for_path(str(dst))
            if not best:
                continue
            base = best.get("path")
            if not base:
                continue
            key = str(base)
            if key not in by_dir:
                by_dir[key] = {"moves": [], "source": best.get("source"), "root_id": best.get("root_id")}
            by_dir[key]["moves"].append((src, dst))
        return by_dir

    async def _dispatch_file_groups(
        self,
        grouped: dict[str, dict],
        callback: Callable[[list, str, str | None, str | None], Awaitable[None]],
        label: str,
    ) -> None:
        for base_dir, payload in grouped.items():
            try:
                await callback(
                    payload.get("files") or [],
                    base_dir,
                    payload.get("source"),
                    payload.get("root_id"),
                )
            except Exception as exc:
                logger.debug("Watcher %s error: %s", label, exc)

    @staticmethod
    def _split_move_pairs(moves: list[Any]) -> tuple[list[str], list[str]]:
        src_files: list[str] = []
        dst_files: list[str] = []
        for move in moves:
            if not isinstance(move, (list, tuple)) or len(move) != 2:
                continue
            src_files.append(str(move[0]))
            dst_files.append(str(move[1]))
        return src_files, dst_files

    def remove_path(self, path: str):
        """Remove a path from watching (e.g., when custom root is removed)."""
        if not self._running or not self._observer:
            return

        try:
            normalized = os.path.normpath(path)
            # Find and remove the watch
            to_remove = None
            for key, entry in self._watched_paths.items():
                if entry.get("path") == normalized:
                    to_remove = key
                    break

            if to_remove:
                entry = self._watched_paths.pop(to_remove, {"path": None, "source": None, "root_id": None, "watch": None})
                try:
                    if self._observer and entry.get("watch"):
                        self._observer.unschedule(entry.get("watch"))
                except Exception:
                    pass
                logger.info("Watcher removed: %s", normalized)
        except Exception as e:
            logger.debug("Failed to remove watch for %s: %s", path, e)

    @property
    def is_running(self) -> bool:
        return self._running

    @property
    def watched_directories(self) -> list:
        return [entry.get("path") for entry in self._watched_paths.values() if entry.get("path")]

    def _allows_source(self, source: str | None) -> bool:
        if not self._allowed_sources:
            return True
        if not source:
            return False
        return str(source).strip().lower() in self._allowed_sources

    def refresh_runtime_settings(self) -> None:
        """Refresh handler timing parameters without restarting."""
        if not self._handler:
            return
        try:
            self._handler.refresh_runtime_settings()
        except Exception:
            pass

    def flush_pending(self) -> bool:
        """Flush watcher pending queue immediately (best-effort)."""
        if not self._handler:
            return False
        try:
            return bool(self._handler.flush_pending())
        except Exception:
            return False

    def get_pending_count(self) -> int:
        """Return pending watcher files waiting for flush."""
        if not self._handler:
            return 0
        try:
            return int(self._handler.get_pending_count())
        except Exception:
            return 0


def _is_under_path(candidate: str, root: str) -> bool:
    if not candidate or not root:
        return False
    try:
        common = os.path.commonpath([candidate, root])
        if common == root:
            return True
    except Exception:
        pass
    try:
        root_sep = root if root.endswith(os.sep) else root + os.sep
        return candidate.startswith(root_sep) or candidate == root
    except Exception:
        return False