clawft-core 0.8.3

//! File-system watcher for automatic skill hot-reload.
//!
//! Watches skill directories for changes and triggers [`SkillRegistry`] rebuilds
//! with debouncing to avoid redundant work during rapid file operations.
//!
//! # Architecture
//!
//! The watcher runs as a background `tokio::spawn` task that:
//! 1. Receives events from the [`notify`] crate via an `mpsc` channel.
//! 2. Debounces rapid changes (default: 500ms).
//! 3. Acquires a write lock on [`SharedSkillRegistry`] only during rebuild.
//! 4. Releases the write lock immediately after rebuild completes.
//!
//! # Concurrency
//!
//! - Reads: Every agent loop iteration reads skills (high frequency).
//! - Writes: Only on file-system changes (low frequency).
//! - [`tokio::sync::RwLock`] is appropriate: many concurrent readers, rare
//!   exclusive writers. Write starvation is not a concern given low write
//!   frequency.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use tokio::sync::mpsc;
use tracing::{debug, info, warn};

use clawft_types::skill::SkillDefinition;

use super::skills_v2::SharedSkillRegistry;

/// Configuration for the skill file-system watcher.
#[derive(Debug, Clone)]
pub struct SkillWatcherConfig {
    /// Workspace skill directory (highest priority).
    pub workspace_dir: Option<PathBuf>,
    /// User skill directory (medium priority).
    pub user_dir: Option<PathBuf>,
    /// Additional directories to watch (e.g. plugin-shipped skill dirs).
    pub extra_dirs: Vec<PathBuf>,
    /// Debounce duration for rapid file changes.
    pub debounce: Duration,
    /// Built-in skills (lowest priority, compiled into binary).
    pub builtin_skills: Vec<SkillDefinition>,
    /// Whether workspace skills are trusted.
    pub trust_workspace: bool,
}

impl Default for SkillWatcherConfig {
    fn default() -> Self {
        Self {
            workspace_dir: None,
            user_dir: None,
            extra_dirs: Vec::new(),
            debounce: Duration::from_millis(500),
            builtin_skills: Vec::new(),
            trust_workspace: false,
        }
    }
}

impl SkillWatcherConfig {
    /// Collect all directories that should be watched.
    fn watch_dirs(&self) -> Vec<&PathBuf> {
        let mut dirs = Vec::new();
        if let Some(ref d) = self.workspace_dir {
            dirs.push(d);
        }
        if let Some(ref d) = self.user_dir {
            dirs.push(d);
        }
        for d in &self.extra_dirs {
            dirs.push(d);
        }
        dirs
    }
}

/// Handle to a running skill watcher. Drop to stop watching.
pub struct SkillWatcherHandle {
    /// Sends a shutdown signal to the watcher task.
    shutdown_tx: Option<tokio::sync::oneshot::Sender<()>>,
    /// `true` while the watcher task is considered active (WEFT-76).
    ///
    /// [`refresh_registry`] no-ops when this is true — the notify path
    /// already reloads the registry on filesystem changes.
    active: Arc<std::sync::atomic::AtomicBool>,
}

impl SkillWatcherHandle {
    /// Whether the watcher is still active (not stopped / dropped).
    pub fn is_active(&self) -> bool {
        self.active.load(std::sync::atomic::Ordering::SeqCst)
    }

    /// Stop the watcher gracefully.
    pub fn stop(mut self) {
        self.active
            .store(false, std::sync::atomic::Ordering::SeqCst);
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
    }
}

impl Drop for SkillWatcherHandle {
    fn drop(&mut self) {
        self.active
            .store(false, std::sync::atomic::Ordering::SeqCst);
        if let Some(tx) = self.shutdown_tx.take() {
            let _ = tx.send(());
        }
    }
}

/// Outcome of a manual skill-registry refresh (WEFT-76).
///
/// Headless / CI hosts often skip [`start_watching`]. Call
/// [`refresh_registry`] after installing or editing skills so the
/// in-process registry matches disk.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SkillRefreshOutcome {
    /// File watcher is active; FS events already drive reloads — no rebuild.
    NoOpWatcherEnabled {
        /// Skills currently held in the registry.
        count: usize,
    },
    /// Registry was rebuilt from disk.
    Reloaded {
        /// Number of skills after rebuild.
        count: usize,
        /// Sorted skill names after rebuild.
        names: Vec<String>,
    },
}

/// Manually reload the skill registry when the notify watcher is disabled.
///
/// # Behaviour
///
/// - When `watcher_active` is `true` (or a live [`SkillWatcherHandle`]
///   reports [`SkillWatcherHandle::is_active`]), returns
///   [`SkillRefreshOutcome::NoOpWatcherEnabled`] without rebuilding.
///   Watcher-enabled hosts already hot-reload on FS changes.
/// - When `watcher_active` is `false`, acquires a write lock, rebuilds
///   via [`SkillRegistry::rebuild`](super::skills_v2::SkillRegistry::rebuild),
///   and returns [`SkillRefreshOutcome::Reloaded`].
///
/// # Errors
///
/// Propagates discovery / I/O errors from the rebuild pass.
pub async fn refresh_registry(
    registry: &SharedSkillRegistry,
    config: &SkillWatcherConfig,
    watcher_active: bool,
) -> clawft_types::Result<SkillRefreshOutcome> {
    if watcher_active {
        let reg = registry.read().await;
        return Ok(SkillRefreshOutcome::NoOpWatcherEnabled { count: reg.len() });
    }

    let mut reg = registry.write().await;
    reg.rebuild(
        config.workspace_dir.as_deref(),
        config.user_dir.as_deref(),
        config.builtin_skills.clone(),
        config.trust_workspace,
    )
    .await?;

    let names: Vec<String> = reg.names().into_iter().map(|s| s.to_string()).collect();
    let count = reg.len();
    info!(count, "skill registry manually refreshed (watcher disabled)");
    Ok(SkillRefreshOutcome::Reloaded { count, names })
}

/// Start watching skill directories for changes.
///
/// Returns a [`SkillWatcherHandle`] that can be used to stop watching.
/// The watcher runs in a background tokio task and will automatically
/// rebuild the registry when files change.
///
/// # Errors
///
/// Returns an error if the file-system watcher cannot be created.
pub fn start_watching(
    config: SkillWatcherConfig,
    registry: SharedSkillRegistry,
) -> Result<SkillWatcherHandle, notify::Error> {
    let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
    let (event_tx, mut event_rx) = mpsc::channel::<Event>(100);
    let active = Arc::new(std::sync::atomic::AtomicBool::new(true));
    let active_for_task = Arc::clone(&active);

    // Create OS file watcher.
    let mut watcher = RecommendedWatcher::new(
        move |res: notify::Result<Event>| {
            if let Ok(event) = res {
                let _ = event_tx.blocking_send(event);
            }
        },
        notify::Config::default(),
    )?;

    // Watch all configured directories that exist.
    for dir in config.watch_dirs() {
        if dir.exists() {
            if let Err(e) = watcher.watch(dir, RecursiveMode::Recursive) {
                warn!(
                    path = %dir.display(),
                    error = %e,
                    "failed to watch skill directory"
                );
            } else {
                debug!(path = %dir.display(), "watching skill directory");
            }
        } else {
            debug!(path = %dir.display(), "skill directory does not exist, skipping watch");
        }
    }

    let debounce = config.debounce;
    let ws_dir = config.workspace_dir.clone();
    let user_dir = config.user_dir.clone();
    let builtins = config.builtin_skills.clone();
    let trust = config.trust_workspace;

    // Spawn the event processing loop.
    tokio::spawn(async move {
        // Keep the watcher alive for the duration of the task.
        let _watcher = watcher;

        let mut pending_rebuild = false;
        let mut debounce_deadline: Option<tokio::time::Instant> = None;

        loop {
            tokio::select! {
                // Receive file-system events.
                event = event_rx.recv() => {
                    match event {
                        Some(ev) => {
                            match ev.kind {
                                EventKind::Create(_)
                                | EventKind::Modify(_)
                                | EventKind::Remove(_) => {
                                    pending_rebuild = true;
                                    debounce_deadline = Some(
                                        tokio::time::Instant::now() + debounce,
                                    );
                                }
                                _ => {}
                            }
                        }
                        None => {
                            // Channel closed, exit.
                            break;
                        }
                    }
                }
                // Debounce timer fires.
                _ = async {
                    match debounce_deadline {
                        Some(deadline) => tokio::time::sleep_until(deadline).await,
                        None => std::future::pending::<()>().await,
                    }
                }, if pending_rebuild => {
                    pending_rebuild = false;
                    debounce_deadline = None;

                    let mut reg = registry.write().await;
                    match reg.rebuild(
                        ws_dir.as_deref(),
                        user_dir.as_deref(),
                        builtins.clone(),
                        trust,
                    ).await {
                        Ok(()) => {
                            info!(
                                count = reg.len(),
                                "skill registry reloaded after file change"
                            );
                        }
                        Err(e) => {
                            warn!(error = %e, "skill registry rebuild failed");
                        }
                    }
                }
                // Shutdown signal.
                _ = &mut shutdown_rx => {
                    active_for_task.store(false, std::sync::atomic::Ordering::SeqCst);
                    info!("skill watcher shutting down");
                    break;
                }
            }
        }
        active_for_task.store(false, std::sync::atomic::Ordering::SeqCst);
    });

    Ok(SkillWatcherHandle {
        shutdown_tx: Some(shutdown_tx),
        active,
    })
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::{AtomicU64, Ordering};

    use tokio::sync::RwLock;

    use super::super::skills_v2::SkillRegistry;
    use clawft_types::skill::SkillDefinition;

    static TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    fn temp_dir(prefix: &str) -> PathBuf {
        let id = TEST_COUNTER.fetch_add(1, Ordering::Relaxed);
        let pid = std::process::id();
        std::env::temp_dir().join(format!("clawft_watcher_{prefix}_{pid}_{id}"))
    }

    fn create_skill_md(dir: &std::path::Path, name: &str, desc: &str) {
        let skill_dir = dir.join(name);
        std::fs::create_dir_all(&skill_dir).unwrap();
        let content =
            format!("---\nname: {name}\ndescription: {desc}\n---\n\nInstructions for {name}.");
        std::fs::write(skill_dir.join("SKILL.md"), content).unwrap();
    }

    #[tokio::test]
    async fn test_watcher_config_defaults() {
        let config = SkillWatcherConfig::default();
        assert_eq!(config.debounce, Duration::from_millis(500));
        assert!(config.workspace_dir.is_none());
        assert!(config.user_dir.is_none());
        assert!(config.extra_dirs.is_empty());
        assert!(config.builtin_skills.is_empty());
        assert!(!config.trust_workspace);
    }

    #[tokio::test]
    async fn test_watcher_config_watch_dirs() {
        let config = SkillWatcherConfig {
            workspace_dir: Some(PathBuf::from("/ws/skills")),
            user_dir: Some(PathBuf::from("/user/skills")),
            extra_dirs: vec![PathBuf::from("/extra/skills")],
            ..Default::default()
        };
        assert_eq!(config.watch_dirs().len(), 3);
    }

    #[tokio::test]
    async fn test_watcher_handle_drop_sends_shutdown() {
        let dir = temp_dir("handle_drop");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "drop_test", "Drop test skill");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&dir), None, vec![])
                .await
                .unwrap(),
        ));

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            debounce: Duration::from_millis(50),
            ..Default::default()
        };

        let handle = start_watching(config, registry).unwrap();
        // Dropping the handle should send shutdown
        drop(handle);

        // Give the task a moment to receive the signal
        tokio::time::sleep(Duration::from_millis(100)).await;

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_watcher_detects_new_skill() {
        let dir = temp_dir("detect_new");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "original", "Original skill");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&dir), None, vec![])
                .await
                .unwrap(),
        ));

        {
            let reg = registry.read().await;
            assert_eq!(reg.len(), 1);
            assert!(reg.get("original").is_some());
        }

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            debounce: Duration::from_millis(100),
            ..Default::default()
        };

        let handle = start_watching(config, registry.clone()).unwrap();

        // Add a new skill
        create_skill_md(&dir, "added", "Added skill");

        // Wait for debounce + rebuild
        tokio::time::sleep(Duration::from_millis(500)).await;

        {
            let reg = registry.read().await;
            assert_eq!(reg.len(), 2);
            assert!(reg.get("original").is_some());
            assert!(reg.get("added").is_some());
        }

        handle.stop();
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_watcher_detects_skill_removal() {
        let dir = temp_dir("detect_remove");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "alpha", "Alpha");
        create_skill_md(&dir, "beta", "Beta");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&dir), None, vec![])
                .await
                .unwrap(),
        ));

        {
            let reg = registry.read().await;
            assert_eq!(reg.len(), 2);
        }

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            debounce: Duration::from_millis(100),
            ..Default::default()
        };

        let handle = start_watching(config, registry.clone()).unwrap();

        // Remove one skill
        std::fs::remove_dir_all(dir.join("beta")).unwrap();

        // Wait for debounce + rebuild
        tokio::time::sleep(Duration::from_millis(500)).await;

        {
            let reg = registry.read().await;
            assert_eq!(reg.len(), 1);
            assert!(reg.get("alpha").is_some());
            assert!(reg.get("beta").is_none());
        }

        handle.stop();
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_watcher_detects_skill_modification() {
        let dir = temp_dir("detect_modify");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "mutable", "Original description");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&dir), None, vec![])
                .await
                .unwrap(),
        ));

        {
            let reg = registry.read().await;
            assert_eq!(
                reg.get("mutable").unwrap().description,
                "Original description"
            );
        }

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            debounce: Duration::from_millis(100),
            ..Default::default()
        };

        let handle = start_watching(config, registry.clone()).unwrap();

        // Modify the skill
        let modified =
            "---\nname: mutable\ndescription: Updated description\n---\n\nUpdated instructions.";
        std::fs::write(dir.join("mutable").join("SKILL.md"), modified).unwrap();

        // Wait for debounce + rebuild
        tokio::time::sleep(Duration::from_millis(500)).await;

        {
            let reg = registry.read().await;
            assert_eq!(
                reg.get("mutable").unwrap().description,
                "Updated description"
            );
        }

        handle.stop();
        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_watcher_missing_dir_does_not_error() {
        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(None, None, vec![]).await.unwrap(),
        ));

        let config = SkillWatcherConfig {
            workspace_dir: Some(PathBuf::from("/definitely/does/not/exist")),
            debounce: Duration::from_millis(50),
            ..Default::default()
        };

        // Should not panic or error
        let handle = start_watching(config, registry);
        assert!(handle.is_ok());
        handle.unwrap().stop();
    }

    #[tokio::test]
    async fn test_registry_upsert_and_remove() {
        let mut registry = SkillRegistry::discover(None, None, vec![]).await.unwrap();
        assert!(registry.is_empty());

        // Upsert a skill
        let skill = SkillDefinition::new("dynamic", "A dynamically added skill");
        let prev = registry.upsert(skill);
        assert!(prev.is_none());
        assert_eq!(registry.len(), 1);
        assert!(registry.get("dynamic").is_some());

        // Upsert again (replace)
        let updated = SkillDefinition::new("dynamic", "Updated description");
        let prev = registry.upsert(updated);
        assert!(prev.is_some());
        assert_eq!(prev.unwrap().description, "A dynamically added skill");
        assert_eq!(
            registry.get("dynamic").unwrap().description,
            "Updated description"
        );

        // Remove
        let removed = registry.remove("dynamic");
        assert!(removed.is_some());
        assert!(registry.is_empty());

        // Remove nonexistent
        let removed = registry.remove("nonexistent");
        assert!(removed.is_none());
    }

    #[tokio::test]
    async fn test_registry_rebuild() {
        let dir = temp_dir("rebuild");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "first", "First skill");

        let mut registry = SkillRegistry::discover(Some(&dir), None, vec![])
            .await
            .unwrap();
        assert_eq!(registry.len(), 1);

        // Add another skill to disk
        create_skill_md(&dir, "second", "Second skill");

        // Rebuild
        registry
            .rebuild(Some(&dir), None, vec![], true)
            .await
            .unwrap();
        assert_eq!(registry.len(), 2);
        assert!(registry.get("first").is_some());
        assert!(registry.get("second").is_some());

        let _ = std::fs::remove_dir_all(&dir);
    }

    // ── WEFT-76: manual refresh (watcher disabled path) ────────────

    #[tokio::test]
    async fn test_refresh_registry_reloads_when_watcher_disabled() {
        let dir = temp_dir("refresh_disabled");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "alpha", "Alpha skill");

        // Start with an empty registry (stale / pre-discover state).
        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(None, None, vec![]).await.unwrap(),
        ));
        assert_eq!(registry.read().await.len(), 0);

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            ..Default::default()
        };

        // watcher_active = false → full rebuild from disk.
        let outcome = refresh_registry(&registry, &config, false)
            .await
            .unwrap();
        match outcome {
            SkillRefreshOutcome::Reloaded { count, names } => {
                assert_eq!(count, 1);
                assert_eq!(names, vec!["alpha".to_string()]);
            }
            SkillRefreshOutcome::NoOpWatcherEnabled { .. } => {
                panic!("expected Reloaded when watcher disabled");
            }
        }
        assert!(registry.read().await.get("alpha").is_some());

        // Add a second skill and refresh again.
        create_skill_md(&dir, "beta", "Beta skill");
        let outcome = refresh_registry(&registry, &config, false)
            .await
            .unwrap();
        match outcome {
            SkillRefreshOutcome::Reloaded { count, names } => {
                assert_eq!(count, 2);
                assert!(names.contains(&"alpha".to_string()));
                assert!(names.contains(&"beta".to_string()));
            }
            SkillRefreshOutcome::NoOpWatcherEnabled { .. } => {
                panic!("expected Reloaded when watcher disabled");
            }
        }

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_refresh_registry_noop_when_watcher_active() {
        let dir = temp_dir("refresh_noop");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "only", "Only skill");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&dir), None, vec![])
                .await
                .unwrap(),
        ));
        assert_eq!(registry.read().await.len(), 1);

        // Disk gains a new skill, but watcher_active=true must not rebuild.
        create_skill_md(&dir, "extra", "Extra skill");

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            ..Default::default()
        };

        let outcome = refresh_registry(&registry, &config, true)
            .await
            .unwrap();
        match outcome {
            SkillRefreshOutcome::NoOpWatcherEnabled { count } => {
                assert_eq!(count, 1, "registry must stay at pre-refresh size");
            }
            SkillRefreshOutcome::Reloaded { .. } => {
                panic!("expected NoOpWatcherEnabled when watcher active");
            }
        }
        // Still only the original skill — manual refresh was a no-op.
        assert!(registry.read().await.get("only").is_some());
        assert!(registry.read().await.get("extra").is_none());

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_watcher_handle_is_active() {
        let dir = temp_dir("handle_active");
        std::fs::create_dir_all(&dir).unwrap();
        create_skill_md(&dir, "alive", "Alive skill");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&dir), None, vec![])
                .await
                .unwrap(),
        ));

        let config = SkillWatcherConfig {
            workspace_dir: Some(dir.clone()),
            trust_workspace: true,
            debounce: Duration::from_millis(50),
            ..Default::default()
        };

        let handle = start_watching(config, registry).unwrap();
        assert!(handle.is_active());
        handle.stop();
        // After stop, active flag is cleared (handle moved into stop).

        let _ = std::fs::remove_dir_all(&dir);
    }

    #[tokio::test]
    async fn test_precedence_on_reload() {
        let ws_dir = temp_dir("prec_ws");
        let user_dir = temp_dir("prec_user");

        std::fs::create_dir_all(&ws_dir).unwrap();
        std::fs::create_dir_all(&user_dir).unwrap();

        // Both have a skill with the same name
        create_skill_md(&ws_dir, "shared", "Workspace version");
        create_skill_md(&user_dir, "shared", "User version");

        let registry = Arc::new(RwLock::new(
            SkillRegistry::discover(Some(&ws_dir), Some(&user_dir), vec![])
                .await
                .unwrap(),
        ));

        {
            let reg = registry.read().await;
            // Workspace takes priority
            assert_eq!(reg.get("shared").unwrap().description, "Workspace version");
        }

        let config = SkillWatcherConfig {
            workspace_dir: Some(ws_dir.clone()),
            user_dir: Some(user_dir.clone()),
            trust_workspace: true,
            debounce: Duration::from_millis(100),
            ..Default::default()
        };

        let handle = start_watching(config, registry.clone()).unwrap();

        // Remove the workspace skill -- user version should be revealed
        std::fs::remove_dir_all(ws_dir.join("shared")).unwrap();

        tokio::time::sleep(Duration::from_millis(500)).await;

        {
            let reg = registry.read().await;
            assert_eq!(reg.get("shared").unwrap().description, "User version");
        }

        handle.stop();
        let _ = std::fs::remove_dir_all(&ws_dir);
        let _ = std::fs::remove_dir_all(&user_dir);
    }
}