Tests: event-driven conflict responder, sink-derived prompt counts

- Replace both copies of the racy polling `spawn_conflict_responder` (in `volume_merge_tests.rs` and `volume_rename_merge_tests.rs`) with one shared `ConflictResponderSink` in `conflict_responder_test_support.rs`. The sink wraps a `CollectorEventSink`, forwards every event, and answers a Stop-mode prompt synchronously the instant it observes the `write-conflict` event — no 2 ms polling loop, no parallel `AtomicUsize` answer counter.
- Switch every `answered`-based assertion to sink-derived counts via `file_conflict_count(&events.inner)`. Counts are authoritative once the op future returns, and the assertions got stronger: the deep-clash test now pins the file-conflict count AND the path/flag shape, not just a counter.
- Make the responder design sound by storing the oneshot sender BEFORE emitting the `write-conflict` event in `volume_conflict.rs`'s Stop branch. Previously the emit ran before the store, so a sink taking the sender inside `emit_conflict` would miss; storing first makes the take race-free for both the test sink and the real FE responder.
- Drop the obsolete "spawn_conflict_responder must bump its counter before send" gotcha in `transfer/CLAUDE.md`; replace with a one-liner naming the responder-sink pattern.
- Determinism: the previously-flaky case-fold and stop-clash tests now pass 40x at `-j 8`; the design is order-independent.

Author: vdavid

apps/desktop/src-tauri/src/file_system/write_operations/transfer/CLAUDE.md

@@ -182,5 +182,4 @@ Our chunked copy (1 MB read/write chunks) provides: identical speed for non-clon
 **Gotcha**: Cross-type Overwrite (file↔folder) is delete-first, NOT a merge or safe-replace.
 **Why**: A type swap can't temp-rename across backends, so `apply_volume_conflict_resolution` deletes the dest first (`delete_volume_path_recursive` for a folder dest, `Volume::delete` for a file dest) before the source materializes. These are rare and lower-stakes (a type mismatch already means wholesale content replacement). Same-type dir-vs-dir never reaches `apply_volume_conflict_resolution` for the folder — it short-circuits to merge in `resolve_volume_conflict` before any policy dispatch.
 
-**Gotcha (test harness)**: `volume_rename_merge_tests.rs::spawn_conflict_responder` must bump its `answered` counter BEFORE `tx.send`, never after.
-**Why**: `tx.send` unblocks the operation's `rx.await` synchronously, so the op can run to completion — and the test can read `answered` — before the responder task is rescheduled to run the line after the send. Incrementing after the send left a window (negligible on an idle Mac, ~6% under the CPU contention of a full `rust-tests-linux` run) where the op finished with the counter still reading 0, producing a false "a clash must prompt exactly once" failure (`left: 0, right: 1`) on the case-fold and Stop-clash tests. The production rename-merge is correct; only the test's answer-accounting raced. Don't reorder it back.
+**Test harness: the conflict responder is an event sink, prompt counts come from the sink.** The folder-merge suites (`volume_merge_tests.rs`, `volume_rename_merge_tests.rs`) drive Stop-mode prompts with `ConflictResponderSink` (`conflict_responder_test_support.rs`): it wraps a `CollectorEventSink`, forwards every event, and the instant it observes a `write-conflict` it `take()`s `state.conflict_resolution_tx` and synchronously sends the scripted answer. This works because the Stop branch stores the sender BEFORE emitting the event (`volume_conflict.rs`), so the take can't miss. Assertions derive the prompt count from the recorded conflicts via `file_conflict_count(&events.inner)` — race-free once the op future returns — never from a side-channel counter. The pattern is order-independent by design, so there's no polling loop and no answer-accounting race to defend.

apps/desktop/src-tauri/src/file_system/write_operations/transfer/conflict_responder_test_support.rs

@@ -0,0 +1,111 @@
+//! Shared test support for driving Stop-mode conflict prompts in the
+//! folder-merge suites (`volume_merge_tests.rs` and `volume_rename_merge_tests.rs`).
+//!
+//! ## Why an event-driven responder, not a polling one
+//!
+//! A Stop-mode clash emits a `write-conflict` event and then blocks the
+//! operation on a `tokio::sync::oneshot` receiver until something fills the
+//! `state.conflict_resolution_tx` slot. The merge engine stores that sender
+//! BEFORE emitting the event (see `volume_conflict.rs`'s Stop branch), so by the
+//! time any sink observes the `emit_conflict` call the sender is already in the
+//! slot. [`ConflictResponderSink`] exploits exactly that: it wraps an inner
+//! [`CollectorEventSink`], forwards every event, and — the instant it sees a
+//! conflict — synchronously `take()`s the sender and sends the scripted
+//! [`ConflictResolutionResponse`]. The op's `rx.await` then returns immediately.
+//!
+//! This is order-independent by construction: there is no parallel counter to
+//! race against the op future, and no 2 ms polling loop. Once the driven op
+//! future completes, the inner collector's recorded conflicts ARE the
+//! authoritative, race-free prompt count — `events` carries the paths and
+//! file/folder flags too, so assertions derive from the sink, not a side-channel
+//! `AtomicUsize`. See [`file_conflict_count`].
+
+use std::sync::Arc;
+
+use super::super::state::{ConflictResolutionResponse, WriteOperationState};
+use super::super::types::{
+    CollectorEventSink, ConflictInfo, ConflictResolution, DryRunResult, OperationEventSink, ScanProgressEvent,
+    WriteCancelledEvent, WriteCompleteEvent, WriteConflictEvent, WriteErrorEvent, WriteProgressEvent,
+    WriteSettledEvent, WriteSourceItemDoneEvent,
+};
+use crate::ignore_poison::IgnorePoison;
+
+/// An event sink that auto-answers Stop-mode `write-conflict` prompts with a
+/// scripted resolution, the moment it observes them. Forwards every event to an
+/// inner [`CollectorEventSink`], so the driving test can derive race-free prompt
+/// counts (and richer path/flag assertions) from `sink.inner` after the op
+/// completes.
+///
+/// Use it as the operation's `events` sink directly — it replaces the old
+/// pattern of a `CollectorEventSink` plus a separately-spawned polling responder
+/// task. Because it answers synchronously inside `emit_conflict` (the sender is
+/// already stored by then), there is no task to abort and no polling window.
+pub(super) struct ConflictResponderSink {
+    pub inner: CollectorEventSink,
+    state: Arc<WriteOperationState>,
+    resolution: ConflictResolution,
+    apply_to_all: bool,
+}
+
+impl ConflictResponderSink {
+    /// Answers every prompt with `resolution` / `apply_to_all`.
+    pub(super) fn new(state: &Arc<WriteOperationState>, resolution: ConflictResolution, apply_to_all: bool) -> Self {
+        Self {
+            inner: CollectorEventSink::new(),
+            state: Arc::clone(state),
+            resolution,
+            apply_to_all,
+        }
+    }
+}
+
+impl OperationEventSink for ConflictResponderSink {
+    fn emit_progress(&self, event: WriteProgressEvent) {
+        self.inner.emit_progress(event);
+    }
+    fn emit_complete(&self, e: WriteCompleteEvent) {
+        self.inner.emit_complete(e);
+    }
+    fn emit_cancelled(&self, e: WriteCancelledEvent) {
+        self.inner.emit_cancelled(e);
+    }
+    fn emit_error(&self, e: WriteErrorEvent) {
+        self.inner.emit_error(e);
+    }
+    fn emit_conflict(&self, e: WriteConflictEvent) {
+        // Record the prompt first (so the count is authoritative even if the
+        // send below races teardown), then answer it.
+        self.inner.emit_conflict(e);
+
+        // The sender was stored before this event was emitted, so the `take()`
+        // can't miss. Sending unblocks the op's `rx.await` synchronously.
+        if let Some(tx) = self.state.conflict_resolution_tx.lock_ignore_poison().take() {
+            let _ = tx.send(ConflictResolutionResponse {
+                resolution: self.resolution,
+                apply_to_all: self.apply_to_all,
+            });
+        }
+    }
+    fn emit_source_item_done(&self, _e: WriteSourceItemDoneEvent) {}
+    fn emit_scan_progress(&self, _e: ScanProgressEvent) {}
+    fn emit_scan_conflict(&self, _c: ConflictInfo) {}
+    fn emit_dry_run_complete(&self, _r: DryRunResult) {}
+    fn emit_settled(&self, _e: WriteSettledEvent) {}
+}
+
+/// Counts `write-conflict` events that are a FILE-vs-FILE clash (neither side a
+/// directory) — i.e. the per-file prompts a merge can legitimately raise. This
+/// is the authoritative, race-free prompt count once the driven op future has
+/// completed, replacing the old parallel `AtomicUsize` answer counter. Dir-vs-dir
+/// merges never emit a conflict at all (the resolver short-circuits before the
+/// emit), so this equals the total emitted conflicts in a pure file-clash merge;
+/// filtering to file-vs-file keeps it honest if a cross-type clash is ever mixed
+/// in.
+pub(super) fn file_conflict_count(events: &CollectorEventSink) -> usize {
+    events
+        .conflicts
+        .lock_ignore_poison()
+        .iter()
+        .filter(|c| !c.source_is_directory && !c.destination_is_directory)
+        .count()
+}

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

@@ -31,6 +31,8 @@ pub(super) mod volume_strategy;
 #[allow(unused_imports, reason = "used by transaction_integration_test")]
 pub(crate) use super::state::CopyTransaction;
 
+#[cfg(test)]
+pub(crate) mod conflict_responder_test_support;
 #[cfg(test)]
 mod copy_integration_test;
 #[cfg(test)]

apps/desktop/src-tauri/src/file_system/write_operations/transfer/volume_conflict.rs

@@ -196,6 +196,16 @@ pub(super) async fn resolve_volume_conflict(
                 _ => None,
             };
 
+            // Store the oneshot sender BEFORE emitting the event. A responder
+            // (the FE's `resolve_write_conflict`, or a test responder sink that
+            // takes the sender inside its `emit_conflict` callback) can only
+            // answer a conflict it has observed; if the event reached it before
+            // the sender slot was filled, its `take()` would miss and the op's
+            // `rx.await` below would hang. Storing first makes the sender
+            // available the instant the event is in the responder's hands.
+            let (tx, rx) = tokio::sync::oneshot::channel();
+            *state.conflict_resolution_tx.lock_ignore_poison() = Some(tx);
+
             events.emit_conflict(WriteConflictEvent {
                 operation_id: operation_id.to_string(),
                 source_path: source_path.display().to_string(),
@@ -210,10 +220,6 @@ pub(super) async fn resolve_volume_conflict(
                 destination_is_directory,
             });
 
-            // Create a oneshot channel for this conflict resolution
-            let (tx, rx) = tokio::sync::oneshot::channel();
-            *state.conflict_resolution_tx.lock_ignore_poison() = Some(tx);
-
             // Wait for user to call resolve_write_conflict.
             match rx.await {
                 Ok(response) => {

apps/desktop/src-tauri/src/file_system/write_operations/transfer/volume_merge_tests.rs

@@ -9,17 +9,16 @@
 //! exactly as in production. Shared fixtures `make_state` / `make_volumes` live in
 //! `volume_copy_tests.rs` (`super::tests`).
 
+use super::super::conflict_responder_test_support::{ConflictResponderSink, file_conflict_count};
 use super::tests::{make_state, make_volumes};
 use super::*;
 use crate::file_system::volume::Volume;
-use crate::file_system::write_operations::state::{
-    ConflictResolutionResponse, WRITE_OPERATION_STATE, cancel_write_operation,
-};
+use crate::file_system::write_operations::state::{WRITE_OPERATION_STATE, cancel_write_operation};
 use crate::file_system::write_operations::types::{
     CollectorEventSink, ConflictResolution, WriteCancelledEvent, WriteCompleteEvent, WriteConflictEvent,
     WriteErrorEvent, WriteSourceItemDoneEvent,
 };
-use std::sync::atomic::{AtomicU8, AtomicUsize};
+use std::sync::atomic::AtomicU8;
 
 // ============================================================================
 // Helpers
@@ -95,35 +94,6 @@ async fn make_rich_merge() -> (Arc<dyn Volume>, Arc<dyn Volume>) {
     (source, dest)
 }
 
-/// Background task that auto-answers EVERY Stop-mode `write-conflict` with a
-/// scripted response by polling `conflict_resolution_tx`. Runs until aborted by
-/// the caller (after the driven op returns), so it never blocks the test for a
-/// fixed duration — the count of prompts answered is read from the returned
-/// `Arc<AtomicUsize>`.
-fn spawn_conflict_responder(
-    state: &Arc<WriteOperationState>,
-    resolution: ConflictResolution,
-    apply_to_all: bool,
-) -> (tokio::task::JoinHandle<()>, Arc<AtomicUsize>) {
-    let state = Arc::clone(state);
-    let answered = Arc::new(AtomicUsize::new(0));
-    let answered_task = Arc::clone(&answered);
-    let handle = tokio::spawn(async move {
-        loop {
-            let tx = state.conflict_resolution_tx.lock().unwrap().take();
-            if let Some(tx) = tx {
-                let _ = tx.send(ConflictResolutionResponse {
-                    resolution,
-                    apply_to_all,
-                });
-                answered_task.fetch_add(1, Ordering::Relaxed);
-            }
-            tokio::time::sleep(Duration::from_millis(2)).await;
-        }
-    });
-    (handle, answered)
-}
-
 /// Count `write-conflict` events whose paths refer to a DIRECTORY on either side
 /// — i.e. a folder-level prompt. The contract is ZERO of these for a dir-vs-dir
 /// merge (folders always merge silently).
@@ -169,9 +139,15 @@ async fn merge_never_deletes_unshadowed_dest_files_under_every_policy() {
         let (source, dest) = make_rich_merge().await;
         let state = make_state();
 
-        let responder = scripted.map(|answer| spawn_conflict_responder(&state, answer, true));
-
-        let events = Arc::new(CollectorEventSink::new());
+        // The responder sink IS the events sink: it forwards every event to its
+        // inner collector and auto-answers any Stop-mode prompt. For non-Stop
+        // policies no prompt is ever emitted, so the scripted answer (defaulted
+        // to Skip here) never fires — the sink is a plain collector in that case.
+        let events = Arc::new(ConflictResponderSink::new(
+            &state,
+            scripted.unwrap_or(ConflictResolution::Skip),
+            true,
+        ));
         let config = VolumeCopyConfig {
             conflict_resolution: *policy,
             progress_interval_ms: 0,
@@ -190,9 +166,6 @@ async fn merge_never_deletes_unshadowed_dest_files_under_every_policy() {
         )
         .await;
 
-        if let Some((handle, _)) = responder {
-            handle.abort();
-        }
         assert!(
             result.is_ok(),
             "policy {policy:?}/{scripted:?} should complete, got {result:?}"
@@ -224,7 +197,7 @@ async fn merge_never_deletes_unshadowed_dest_files_under_every_policy() {
 
         // Zero folder-level prompts under EVERY policy, even Stop.
         assert_eq!(
-            folder_conflict_count(&events),
+            folder_conflict_count(&events.inner),
             0,
             "policy {policy:?}/{scripted:?}: a dir-vs-dir merge wrongly emitted a folder conflict"
         );
@@ -457,9 +430,7 @@ async fn stop_mode_deep_file_clash_emits_conflict_and_resumes() {
         .unwrap();
 
     let state = make_state();
-    let (responder, answered) = spawn_conflict_responder(&state, ConflictResolution::Overwrite, false);
-
-    let events = Arc::new(CollectorEventSink::new());
+    let events = Arc::new(ConflictResponderSink::new(&state, ConflictResolution::Overwrite, false));
     let config = VolumeCopyConfig {
         conflict_resolution: ConflictResolution::Stop,
         progress_interval_ms: 0,
@@ -477,10 +448,11 @@ async fn stop_mode_deep_file_clash_emits_conflict_and_resumes() {
         &config,
     )
     .await;
-    responder.abort();
     assert!(result.is_ok(), "expected Ok, got {result:?}");
+    // The sink-recorded file prompts ARE the race-free count once the op future
+    // has completed — exactly one Stop prompt for the deep clash.
     assert_eq!(
-        answered.load(Ordering::Relaxed),
+        file_conflict_count(&events.inner),
         1,
         "exactly one Stop prompt expected for the deep clash"
     );
@@ -491,7 +463,7 @@ async fn stop_mode_deep_file_clash_emits_conflict_and_resumes() {
     // `await_holding_lock` flags a guard alive across `.await` even with an
     // explicit `drop`, so we end the borrow by leaving the block instead).
     let (src_path, dst_path, src_is_dir, dst_is_dir, n_conflicts) = {
-        let conflicts = events.conflicts.lock().unwrap();
+        let conflicts = events.inner.conflicts.lock().unwrap();
         let c = conflicts.first().expect("exactly one deep file conflict");
         (
             c.source_path.clone(),
@@ -725,9 +697,7 @@ async fn concurrent_merge_with_two_deep_clashes_serializes_prompts() {
     source.create_file(Path::new("/three.txt"), b"THREE").await.unwrap();
 
     let state = make_state();
-    let (responder, answered) = spawn_conflict_responder(&state, ConflictResolution::Overwrite, false);
-
-    let events = Arc::new(CollectorEventSink::new());
+    let events = Arc::new(ConflictResponderSink::new(&state, ConflictResolution::Overwrite, false));
     let config = VolumeCopyConfig {
         conflict_resolution: ConflictResolution::Stop,
         progress_interval_ms: 0,
@@ -749,9 +719,10 @@ async fn concurrent_merge_with_two_deep_clashes_serializes_prompts() {
         &config,
     )
     .await;
-    responder.abort();
     assert!(result.is_ok(), "expected Ok, got {result:?}");
-    let n = answered.load(Ordering::Relaxed);
+    // Sink-derived: both deep clashes prompted (the dispatch mutex serialized
+    // them through the single oneshot slot, each answered in turn).
+    let n = file_conflict_count(&events.inner);
     assert_eq!(n, 2, "both deep clashes should prompt and be answered, got {n}");
 
     // Both clashes overwritten (50_000 bytes of 1u8), both dest-only files kept.
@@ -790,9 +761,7 @@ async fn top_level_and_deep_clash_share_the_dispatch_mutex() {
     let state = make_state();
     // One "…all" answer collapses any queued prompt via the latch double-check,
     // so at most 2 prompts ever emit (top + deep), often just 1.
-    let (responder, _answered) = spawn_conflict_responder(&state, ConflictResolution::Overwrite, true);
-
-    let events = Arc::new(CollectorEventSink::new());
+    let events = Arc::new(ConflictResponderSink::new(&state, ConflictResolution::Overwrite, true));
     let config = VolumeCopyConfig {
         conflict_resolution: ConflictResolution::Stop,
         progress_interval_ms: 0,
@@ -814,7 +783,6 @@ async fn top_level_and_deep_clash_share_the_dispatch_mutex() {
         &config,
     )
     .await;
-    responder.abort();
     assert!(result.is_ok(), "expected Ok, got {result:?}");
 
     // Both clashes overwritten by the "…all" choice; dest-only file survives.