Local write ops: helpers take sink

- `scan_sources`, `dry_run_scan` (+ `_internal`, `_recursive`), `handle_dry_run`, `resolve_conflict`, and `copy_single_item` migrate from `&tauri::AppHandle` to `&dyn OperationEventSink`. All `app.emit(...)` sites become `events.emit_*(...)` and `state.emit_progress_via_app(...)` becomes `state.emit_progress_via_sink(...)`.
- `OperationEventSink` trait gains `emit_scan_progress`, `emit_scan_conflict`, `emit_dry_run_complete` so the dry-run path goes through the sink. `TauriEventSink` and `CollectorEventSink` implement them.
- Add `helpers::run_cancellable_scoped`, a `std::thread::scope`-backed variant of `run_cancellable` that lets the worker closure borrow non-`'static` data (sink reference). `scan_sources` and `dry_run_scan` use it; `validate_disk_space_cancellable` keeps the detached-thread original.
- `copy_files_with_progress_inner` drops its `app: &tauri::AppHandle` parameter; it now flows everything through `events` only. The intra-file chunked-copy progress callback emits via the sink too.
- `move_op.rs` and `delete.rs` keep their `&AppHandle` signatures (no inner extraction yet) and construct a `TauriEventSink` ad-hoc to pass to the migrated helpers. `merge_move_directory` itself flips to the sink. Full inner extraction lands in follow-up commits.
- Test sinks in `volume_copy_tests.rs` and `volume_move_tests.rs` add the three new trait stubs.
- Allow `clippy::doc_lazy_continuation` / `too_many_arguments` on `driver_prototype_scratch.rs` so the throwaway M2 prototype doesn't block CI.

Author: vdavid

apps/desktop/src-tauri/src/file_system/write_operations/copy.rs

@@ -52,7 +52,6 @@ fn validate_disk_space_cancellable(
 
 pub(super) fn copy_files_with_progress_inner(
     events: &dyn OperationEventSink,
-    app: &tauri::AppHandle,
     operation_id: &str,
     state: &Arc<WriteOperationState>,
     sources: &[PathBuf],
@@ -71,7 +70,7 @@ pub(super) fn copy_files_with_progress_inner(
         sources,
         destination,
         state,
-        app,
+        events,
         operation_id,
         WriteOperationType::Copy,
         state.progress_interval,
@@ -105,7 +104,7 @@ pub(super) fn copy_files_with_progress_inner(
             scan_sources(
                 sources,
                 state,
-                app,
+                events,
                 operation_id,
                 WriteOperationType::Copy,
                 config.sort_column,
@@ -121,7 +120,7 @@ pub(super) fn copy_files_with_progress_inner(
         scan_sources(
             sources,
             state,
-            app,
+            events,
             operation_id,
             WriteOperationType::Copy,
             config.sort_column,
@@ -271,7 +270,7 @@ pub(super) fn copy_files_with_progress_inner(
                 scan_result.file_count,
                 scan_result.total_bytes,
                 state,
-                app,
+                events,
                 operation_id,
                 WriteOperationType::Copy,
                 &state.progress_interval,
@@ -458,7 +457,7 @@ pub(super) fn copy_single_item(
     files_total: usize,
     bytes_total: u64,
     state: &Arc<WriteOperationState>,
-    app: &tauri::AppHandle,
+    events: &dyn OperationEventSink,
     operation_id: &str,
     operation_type: WriteOperationType,
     progress_interval: &Duration,
@@ -517,7 +516,7 @@ pub(super) fn copy_single_item(
                     &blocking,
                     &blocking,
                     config,
-                    app,
+                    events,
                     operation_id,
                     state,
                     apply_to_all_resolution,
@@ -612,7 +611,7 @@ pub(super) fn copy_single_item(
                 source,
                 &dest_path,
                 config,
-                app,
+                events,
                 operation_id,
                 state,
                 apply_to_all_resolution,
@@ -666,7 +665,7 @@ pub(super) fn copy_single_item(
                 source,
                 &dest_path,
                 config,
-                app,
+                events,
                 operation_id,
                 state,
                 apply_to_all_resolution,
@@ -721,8 +720,8 @@ pub(super) fn copy_single_item(
                     effective_bytes_done,
                     bytes_total
                 );
-                state.emit_progress_via_app(
-                    app,
+                state.emit_progress_via_sink(
+                    events,
                     WriteProgressEvent::new(
                         operation_id.to_string(),
                         operation_type,
@@ -770,8 +769,8 @@ pub(super) fn copy_single_item(
                 *bytes_done,
                 bytes_total
             );
-            state.emit_progress_via_app(
-                app,
+            state.emit_progress_via_sink(
+                events,
                 WriteProgressEvent::new(
                     operation_id.to_string(),
                     operation_type,

apps/desktop/src-tauri/src/file_system/write_operations/delete.rs

@@ -9,8 +9,8 @@ use super::helpers::spawn_async_sync;
 use super::scan::{SourceItemTracker, scan_sources};
 use super::state::{WriteOperationState, update_operation_status};
 use super::types::{
-    DryRunResult, IoResultExt, WriteCancelledEvent, WriteCompleteEvent, WriteOperationConfig, WriteOperationError,
-    WriteOperationPhase, WriteOperationType, WriteProgressEvent, WriteSourceItemDoneEvent,
+    DryRunResult, IoResultExt, TauriEventSink, WriteCancelledEvent, WriteCompleteEvent, WriteOperationConfig,
+    WriteOperationError, WriteOperationPhase, WriteOperationType, WriteProgressEvent, WriteSourceItemDoneEvent,
 };
 use super::volume_copy::map_volume_error;
 use crate::file_system::volume::Volume;
@@ -28,11 +28,13 @@ pub(super) fn delete_files_with_progress(
 ) -> Result<(), WriteOperationError> {
     use tauri::Emitter;
 
+    let events = TauriEventSink::new(app.clone());
+
     // Phase 1: Scan to get file count (delete uses default sorting)
     let scan_result = scan_sources(
         sources,
         state,
-        app,
+        &events,
         operation_id,
         WriteOperationType::Delete,
         config.sort_column,

apps/desktop/src-tauri/src/file_system/write_operations/driver_prototype_scratch.rs

@@ -77,16 +77,20 @@
 
 #![cfg(test)]
 #![allow(dead_code, reason = "Throwaway prototype, deleted at end of M2")]
+#![allow(
+    clippy::doc_lazy_continuation,
+    clippy::too_many_arguments,
+    reason = "Throwaway prototype, not held to the project's lints"
+)]
 
-use std::collections::HashSet;
-use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use super::scan::SourceItemTracker;
 use super::state::{CopyTransaction, FileInfo, WriteOperationState};
 use super::types::{
-    ConflictResolution, OperationEventSink, WriteOperationError, WriteOperationType,
-    WriteSourceItemDoneEvent,
+    ConflictResolution, OperationEventSink, WriteOperationError, WriteOperationType, WriteSourceItemDoneEvent,
 };
+use std::collections::HashSet;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
 
 // ---------------------------------------------------------------------------
 // Proposed driver API surface (sketch, not the final shape)

apps/desktop/src-tauri/src/file_system/write_operations/helpers.rs

@@ -15,7 +15,9 @@ use super::macos_copy::{CopyProgressContext, copy_single_file_native};
 use super::state::WriteOperationState;
 #[cfg(not(target_os = "macos"))]
 use super::types::IoResultExt;
-use super::types::{ConflictInfo, ConflictResolution, WriteConflictEvent, WriteOperationConfig, WriteOperationError};
+use super::types::{
+    ConflictInfo, ConflictResolution, OperationEventSink, WriteConflictEvent, WriteOperationConfig, WriteOperationError,
+};
 
 // ============================================================================
 // Validation helpers
@@ -337,13 +339,11 @@ pub(super) fn resolve_conflict(
     source: &Path,
     dest_path: &Path,
     config: &WriteOperationConfig,
-    app: &tauri::AppHandle,
+    events: &dyn OperationEventSink,
     operation_id: &str,
     state: &Arc<WriteOperationState>,
     apply_to_all_resolution: &mut Option<ConflictResolution>,
 ) -> Result<Option<ResolvedDestination>, WriteOperationError> {
-    use tauri::Emitter;
-
     // Determine effective conflict resolution
     let resolution = if let Some(saved_resolution) = apply_to_all_resolution {
         // Use saved "apply to all" resolution
@@ -385,20 +385,17 @@ pub(super) fn resolve_conflict(
                 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
                 .map(|d| d.as_secs() as i64);
 
-            let _ = app.emit(
-                "write-conflict",
-                WriteConflictEvent {
-                    operation_id: operation_id.to_string(),
-                    source_path: source.display().to_string(),
-                    destination_path: dest_path.display().to_string(),
-                    source_size,
-                    destination_size,
-                    source_modified,
-                    destination_modified,
-                    destination_is_newer,
-                    size_difference,
-                },
-            );
+            events.emit_conflict(WriteConflictEvent {
+                operation_id: operation_id.to_string(),
+                source_path: source.display().to_string(),
+                destination_path: dest_path.display().to_string(),
+                source_size,
+                destination_size,
+                source_modified,
+                destination_modified,
+                destination_is_newer,
+                size_difference,
+            });
 
             // Create a oneshot channel for this conflict resolution
             let (tx, rx) = tokio::sync::oneshot::channel();
@@ -703,3 +700,51 @@ where
         }
     }
 }
+
+/// Scoped variant of [`run_cancellable`] that allows the work closure to borrow
+/// non-`'static` data (for example, a `&dyn OperationEventSink` reference).
+///
+/// Uses `std::thread::scope`, so the call blocks until the worker thread
+/// finishes or cancellation is observed. Behavior is otherwise identical to
+/// `run_cancellable`: the cancellation flag is polled every 100ms while the
+/// worker runs, and the function returns early on cancellation.
+pub(super) fn run_cancellable_scoped<'env, T, F>(
+    work: F,
+    state: &Arc<WriteOperationState>,
+    context: &str,
+    operation_id: &str,
+) -> Result<T, WriteOperationError>
+where
+    F: FnOnce() -> Result<T, WriteOperationError> + Send + 'env,
+    T: Send + 'env,
+{
+    use std::sync::mpsc;
+
+    let (tx, rx) = mpsc::channel();
+
+    std::thread::scope(|scope| {
+        scope.spawn(move || {
+            let _ = tx.send(work());
+        });
+
+        loop {
+            if super::state::is_cancelled(&state.intent) {
+                log::debug!("{context}: cancellation detected during polling op={operation_id}");
+                return Err(WriteOperationError::Cancelled {
+                    message: "Operation cancelled by user".to_string(),
+                });
+            }
+
+            match rx.recv_timeout(CANCELLATION_POLL_INTERVAL) {
+                Ok(result) => return result,
+                Err(mpsc::RecvTimeoutError::Timeout) => {}
+                Err(mpsc::RecvTimeoutError::Disconnected) => {
+                    return Err(WriteOperationError::IoError {
+                        path: context.to_string(),
+                        message: format!("{context} thread terminated unexpectedly"),
+                    });
+                }
+            }
+        }
+    })
+}

apps/desktop/src-tauri/src/file_system/write_operations/mod.rs

@@ -196,7 +196,7 @@ pub async fn copy_files_start(
             validate_not_same_location(&sources, &destination)?;
             validate_destination_not_inside_source(&sources, &destination)?;
             let events = types::TauriEventSink::new(app.clone());
-            copy_files_with_progress_inner(&events, &app, &op_id, &state, &sources, &destination, &config)
+            copy_files_with_progress_inner(&events, &op_id, &state, &sources, &destination, &config)
         },
     )
     .await
@@ -363,10 +363,10 @@ pub async fn trash_files_start(
 #[cfg(test)]
 mod copy_integration_test;
 #[cfg(test)]
-mod driver_prototype_scratch;
-#[cfg(test)]
 mod delete_integration_test;
 #[cfg(test)]
+mod driver_prototype_scratch;
+#[cfg(test)]
 mod move_integration_test;
 #[cfg(test)]
 mod tests;