fix(local_file): address review feedback on change_stream() watcher implementation

crates/recoco-core/src/ops/sources/local_file.rs

@@ -210,20 +210,25 @@ impl SourceExecutor for Executor {
 
         let mut watcher = RecommendedWatcher::new(
             move |res: notify::Result<notify::Event>| {
-                if let Ok(event) = res {
-                    for path in event.paths {
-                        if let Err(err) = tx.try_send(path) {
-                            use tokio::sync::mpsc::error::TrySendError;
-                            match err {
-                                TrySendError::Full(_) => {
-                                    warn!("File watcher channel is full; dropping file change event");
-                                }
-                                TrySendError::Closed(_) => {
-                                    warn!("File watcher channel is closed; dropping file change event");
+                match res {
+                    Ok(event) => {
+                        for path in event.paths {
+                            if let Err(err) = tx.try_send(path) {
+                                use tokio::sync::mpsc::error::TrySendError;
+                                match err {
+                                    TrySendError::Full(_) => {
+                                        warn!("File watcher channel is full; dropping file change event");
+                                    }
+                                    TrySendError::Closed(_) => {
+                                        warn!("File watcher channel is closed; dropping file change event");
+                                    }
                                 }
                             }
                         }
                     }
+                    Err(e) => {
+                        warn!("File watcher error: {}", e);
+                    }
                 }
             },
             Config::default(),
@@ -239,6 +244,22 @@ impl SourceExecutor for Executor {
             let _watcher = watcher;
 
             while let Some(path) = rx.recv().await {
+                // Skip directory paths - notify can emit events for directories,
+                // and reading a directory as a file would produce an EISDIR error.
+                let is_dir = match std::fs::metadata(&path) {
+                    Ok(metadata) => metadata.is_dir(),
+                    Err(err) => {
+                        // If the file no longer exists, this may be a deletion event; do not skip it.
+                        if err.kind() != std::io::ErrorKind::NotFound {
+                            warn!("Failed to read metadata for path {:?}: {}", path, err);
+                        }
+                        false
+                    }
+                };
+                if is_dir {
+                    continue;
+                }
+
                 let mut path_components = path.components();
                 for _ in 0..root_component_size {
                     path_components.next();