implemented sweep mode and better conflict handling

crates/pdfs-cli/src/main.rs

@@ -1291,7 +1291,10 @@ fn mount_once(mountpoint: Option<PathBuf>) -> Result<pdfs_fuse::MountOutcome> {
         cache,
         &control_socket,
         db,
-        username,
+        pdfs_fuse::MountOptions {
+            username,
+            sweep_mode: config.resolved_conflict_sweep(),
+        },
     );
 
     // Clean unmount check (as a best-effort final backup).

crates/pdfs-core/src/config.rs

@@ -20,6 +20,59 @@ pub const KEYRING_SERVICE: &str = "proton-drive-linux";
 /// blobs back under this; pinned files are exempt. See [`crate::cache`].
 pub const DEFAULT_CACHE_BUDGET_BYTES: u64 = 5 * 1024 * 1024 * 1024;
 
+/// Environment override for [`AppConfig::resolved_conflict_sweep`]. Exists so a
+/// support case can disable the sweep without editing the config file or
+/// reinstalling — it wins over the stored setting.
+pub const CONFLICT_SWEEP_ENV: &str = "PDFS_CONFLICT_SWEEP";
+
+/// What the background conflict sweep (`pdfs-fuse`'s `pdfs-conflict-sweep`
+/// thread) is allowed to do about a `(sync-conflict …)` copy it believes is a
+/// redundant duplicate.
+///
+/// Defaults to [`SweepMode::Report`]: the sweep's whole reason to exist is that
+/// the *previous* revision-identity check was wrong (B69), so the replacement is
+/// not trusted with deletion until it has a release cycle of field evidence
+/// behind it. See `docs/BUGS.md` B71.
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
+#[serde(rename_all = "lowercase")]
+pub enum SweepMode {
+    /// Do not run the sweep at all — the thread is never spawned.
+    Off,
+    /// Run, and log what it *would* remove, but never trash anything. Divergent
+    /// and orphaned copies are still surfaced to the activity feed as usual.
+    #[default]
+    Report,
+    /// Trash copies proven identical to their live sibling.
+    Enforce,
+}
+
+impl SweepMode {
+    /// Parse an operator-supplied value ([`CONFLICT_SWEEP_ENV`] or a CLI flag).
+    /// `None` for anything unrecognised, so a typo falls back to the stored
+    /// setting rather than silently picking a mode the operator did not ask for.
+    pub fn parse(value: &str) -> Option<Self> {
+        match value.trim().to_ascii_lowercase().as_str() {
+            "off" | "no" | "false" | "0" | "disabled" => Some(Self::Off),
+            "report" | "dry-run" | "dryrun" | "observe" => Some(Self::Report),
+            "enforce" | "on" | "yes" | "true" | "1" => Some(Self::Enforce),
+            _ => None,
+        }
+    }
+
+    /// Whether this mode is allowed to trash a duplicate.
+    pub fn is_enforcing(self) -> bool {
+        matches!(self, Self::Enforce)
+    }
+}
+
+/// Precedence for [`AppConfig::resolved_conflict_sweep`], split out from the
+/// environment read so it can be unit-tested without mutating process state.
+fn resolve_sweep_mode(env: Option<&str>, stored: Option<SweepMode>) -> SweepMode {
+    env.and_then(SweepMode::parse)
+        .or(stored)
+        .unwrap_or_default()
+}
+
 /// Configuration structure allowing the user to customize client identification headers.
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct AppConfig {
@@ -48,6 +101,11 @@ pub struct AppConfig {
     /// by hostname". Defaulted for configs predating the field.
     #[serde(default)]
     pub device_uid: Option<String>,
+    /// What the background conflict sweep may do. `None` means "use
+    /// [`SweepMode::default`]" (report-only). Defaulted for configs predating
+    /// the field, and overridable at runtime via [`CONFLICT_SWEEP_ENV`].
+    #[serde(default)]
+    pub conflict_sweep: Option<SweepMode>,
 }
 
 impl Default for AppConfig {
@@ -59,6 +117,7 @@ impl Default for AppConfig {
             mountpoint: None,
             ignore_patterns: None,
             device_uid: None,
+            conflict_sweep: None,
         }
     }
 }
@@ -70,6 +129,20 @@ impl AppConfig {
         self.cache_budget.unwrap_or(DEFAULT_CACHE_BUDGET_BYTES)
     }
 
+    /// The effective conflict-sweep mode: the [`CONFLICT_SWEEP_ENV`] override if
+    /// it parses, then the user's stored choice, then [`SweepMode::default`].
+    ///
+    /// The environment wins so a support case can stop a destructive background
+    /// loop without editing config or reinstalling (`docs/BUGS.md` B71). An
+    /// unparseable value is ignored rather than treated as "off", so a typo
+    /// cannot silently change behaviour in either direction.
+    pub fn resolved_conflict_sweep(&self) -> SweepMode {
+        resolve_sweep_mode(
+            std::env::var(CONFLICT_SWEEP_ENV).ok().as_deref(),
+            self.conflict_sweep,
+        )
+    }
+
     /// The effective global sync ignore patterns: the user's explicit list, or
     /// [`DEFAULT_IGNORE_PATTERNS`] when unset. An explicit empty list is
     /// honoured as "no global patterns", which is why this is not a
@@ -374,6 +447,7 @@ mod tests {
             mountpoint: Some("/tmp/x".to_string()),
             ignore_patterns: Some(vec!["build/".to_string()]),
             device_uid: Some("dev-uid".to_string()),
+            conflict_sweep: Some(SweepMode::Enforce),
         };
         let json = serde_json::to_string(&config).unwrap();
         let decoded: AppConfig = serde_json::from_str(&json).unwrap();
@@ -388,6 +462,62 @@ mod tests {
         assert_eq!(default_config.user_agent, USER_AGENT);
     }
 
+    #[test]
+    fn sweep_mode_defaults_to_report_only() {
+        // The sweep trashes user files, so an installed config that has never
+        // heard of the setting must not get the enforcing behaviour.
+        assert_eq!(SweepMode::default(), SweepMode::Report);
+        assert!(!SweepMode::default().is_enforcing());
+        assert_eq!(resolve_sweep_mode(None, None), SweepMode::Report);
+    }
+
+    #[test]
+    fn sweep_mode_parses_operator_spellings() {
+        for s in ["off", "OFF", " no ", "false", "0", "disabled"] {
+            assert_eq!(SweepMode::parse(s), Some(SweepMode::Off), "{s}");
+        }
+        for s in ["report", "dry-run", "dryrun", "observe"] {
+            assert_eq!(SweepMode::parse(s), Some(SweepMode::Report), "{s}");
+        }
+        for s in ["enforce", "on", "yes", "true", "1"] {
+            assert_eq!(SweepMode::parse(s), Some(SweepMode::Enforce), "{s}");
+        }
+        // A typo must not resolve to anything, so it falls through to the config.
+        assert_eq!(SweepMode::parse("offf"), None);
+        assert_eq!(SweepMode::parse(""), None);
+    }
+
+    #[test]
+    fn sweep_mode_env_overrides_config_but_a_typo_does_not() {
+        let stored = Some(SweepMode::Enforce);
+        // The point of the override: kill a destructive loop without editing config.
+        assert_eq!(resolve_sweep_mode(Some("off"), stored), SweepMode::Off);
+        // Unrecognised value falls back to the stored choice rather than guessing.
+        assert_eq!(resolve_sweep_mode(Some("nope"), stored), SweepMode::Enforce);
+        assert_eq!(resolve_sweep_mode(None, stored), SweepMode::Enforce);
+        // No env, no stored value: report-only.
+        assert_eq!(resolve_sweep_mode(Some("nope"), None), SweepMode::Report);
+    }
+
+    #[test]
+    fn a_config_predating_conflict_sweep_still_loads() {
+        let json = r#"{"app_version":"v","user_agent":"u"}"#;
+        let decoded: AppConfig = serde_json::from_str(json).unwrap();
+        assert!(decoded.conflict_sweep.is_none());
+    }
+
+    #[test]
+    fn sweep_mode_round_trips_through_the_config_file() {
+        let config = AppConfig {
+            conflict_sweep: Some(SweepMode::Enforce),
+            ..Default::default()
+        };
+        let json = serde_json::to_string(&config).unwrap();
+        assert!(json.contains(r#""conflict_sweep":"enforce""#), "{json}");
+        let decoded: AppConfig = serde_json::from_str(&json).unwrap();
+        assert_eq!(decoded.conflict_sweep, Some(SweepMode::Enforce));
+    }
+
     #[test]
     fn a_config_predating_ignore_patterns_still_loads() {
         // Every installed config on disk looks like this; the field must not be

crates/pdfs-core/src/db/ops.rs

@@ -280,6 +280,22 @@ impl Db {
         Ok(count > 0)
     }
 
+    /// Check whether *any* op of any kind is queued against `uid`.
+    ///
+    /// Used as a safety interlock by callers that are about to remove a node
+    /// they only know about from a stale snapshot: a queued op means the user's
+    /// bytes are still owed an upload, and dropping it would discard the staged
+    /// blob that holds them (`docs/BUGS.md` B71).
+    pub fn has_any_op(&self, uid: &str) -> Result<bool> {
+        let conn = self.conn.lock();
+        let count: i64 = conn.query_row(
+            "SELECT COUNT(*) FROM pending_op WHERE uid = ?1",
+            params![uid],
+            |row| row.get(0),
+        )?;
+        Ok(count > 0)
+    }
+
     /// Drop every op targeting a node **or anything queued beneath it**,
     /// returning the staged blobs they held so the caller can delete them.
     ///

crates/pdfs-fuse/src/lib.rs

@@ -53,7 +53,7 @@ use fuser::{
     ReplyEntry, ReplyIoctl, ReplyOpen, ReplyWrite, Request, Session, TimeOrNow, WriteFlags,
 };
 use pdfs_core::cache::{BLOCK_SIZE, Baseline, ContentCache, StagedWrite};
-use pdfs_core::config::AppDirs;
+use pdfs_core::config::{AppDirs, SweepMode};
 use pdfs_core::control::{
     ActivityEntry, ActivityKind, DirEntry, ErrorKind, LocalHit, PhotoKind, PublicLinkInfo,
     SearchFilters, SearchHit, SearchSource, SyncFolderInfo, SyncPhase, SyncProgress,
@@ -95,7 +95,7 @@ mod upload;
 mod workers;
 use background::{run_event_sync, run_local_index};
 pub(crate) use mount::is_stale_mount;
-pub use mount::{MountOutcome, mount};
+pub use mount::{MountOptions, MountOutcome, mount};
 use mount::{SecondaryMount, clear_stale_mount, fuse_connection_id, umount_session_unblocked};
 use reads::{ReaderSlot, STREAM_BYPASS_MIN, StreamRing};
 use state::{Entry, Intervals, PendingRevision, State, WriteHandle};
@@ -243,6 +243,11 @@ struct Core {
     /// run, so a divergent `(sync-conflict …)` file is logged once rather than
     /// on every sweep pass. See [`Core::run_conflict_sweep_loop`].
     conflict_notified: Arc<Mutex<HashSet<NodeUid>>>,
+    /// Whether the conflict sweep may actually trash the duplicates it finds, or
+    /// only report them. Resolved once at mount from config + environment
+    /// ([`AppConfig::resolved_conflict_sweep`]); [`SweepMode::Off`] means the
+    /// sweep thread is never spawned at all. See `docs/BUGS.md` B71.
+    sweep_mode: SweepMode,
     /// The latest server revision id this daemon has itself sealed, keyed by
     /// node. A queued write whose baseline names an *earlier* revision would
     /// normally fork into a `(sync-conflict)` copy when the drain finds the

crates/pdfs-fuse/src/mount.rs

@@ -143,6 +143,15 @@ fn fetch_or_recall_root(
     }
 }
 
+/// Per-mount settings resolved from [`pdfs_core::config::AppConfig`] by the
+/// caller, so `mount` itself never reads config or the environment.
+pub struct MountOptions {
+    /// Proton account the session belongs to, for display and per-user paths.
+    pub username: String,
+    /// What the background conflict sweep may do. See `docs/BUGS.md` B71.
+    pub sweep_mode: SweepMode,
+}
+
 /// Mount the filesystem at `mountpoint` and block until it is unmounted or the
 /// daemon is asked to stop.
 ///
@@ -162,8 +171,12 @@ pub fn mount(
     cache: ContentCache,
     control_socket: &Path,
     db: Arc<Db>,
-    username: String,
+    options: MountOptions,
 ) -> std::io::Result<MountOutcome> {
+    let MountOptions {
+        username,
+        sweep_mode,
+    } = options;
     let (root, online) = fetch_or_recall_root(&client, &rt, &db)?;
     let scope = root.tree_event_scope_id();
 
@@ -197,6 +210,7 @@ pub fn mount(
         timeline_refreshing: Arc::new(AtomicBool::new(false)),
         trash_refreshing: Arc::new(AtomicBool::new(false)),
         conflict_notified: Arc::new(Mutex::new(HashSet::new())),
+        sweep_mode,
         own_sealed_revs: Arc::new(Mutex::new(HashMap::new())),
         thumb_gen: Arc::new(Mutex::new(HashSet::new())),
         no_thumbnail: Arc::new(Mutex::new(HashMap::new())),
@@ -231,7 +245,12 @@ pub fn mount(
     // Reconcile leftover `(sync-conflict …)` copies: drop the ones that turned
     // out identical to the live file, surface the ones that genuinely diverge.
     // Its own thread — it enumerates and trashes, and must never block a FUSE call.
-    {
+    // `Off` skips the thread entirely rather than starting an idle one, so the
+    // setting is verifiable from the outside (`ls /proc/<pid>/task/*/comm`).
+    if sweep_mode == SweepMode::Off {
+        info!("conflict sweep disabled by configuration");
+    } else {
+        info!(mode = ?sweep_mode, "conflict sweep enabled");
         let core = core.clone();
         std::thread::Builder::new()
             .name("pdfs-conflict-sweep".into())

crates/pdfs-fuse/src/sweep.rs

@@ -21,11 +21,12 @@
 use std::collections::HashMap;
 use std::time::Duration;
 
+use pdfs_core::config::SweepMode;
 use pdfs_core::control::ActivityKind;
 use proton_drive_rs::proton_sdk::ids::NodeUid;
 use tracing::{debug, info, warn};
 
-use super::{Core, node_content_sha1, node_size};
+use super::{Core, node_content_sha1, node_revision_id, node_size};
 
 /// Time between sweeps. Conflicts are rare and non-urgent; a slow cadence keeps
 /// the sweep off the hot path while still cleaning up within a few minutes.
@@ -112,29 +113,35 @@ impl Core {
                 .unwrap_or_default();
             let sibling = by_parent_name.get(&(parent, base.as_str())).copied();
 
-            match sibling {
-                Some(sib) if is_identical(node, sib) => {
-                    // Redundant duplicate — remove it, but only while online (the
-                    // remote trash is the point). Offline, it waits for a later pass.
+            match decide(node, sibling, &base, self.sweep_mode) {
+                SweepAction::Remove => {
+                    // Only while online — the remote trash is the point. Offline,
+                    // it waits for a later pass.
                     if online {
-                        self.remove_conflict_copy(&node.uid, &node.name, &base);
+                        self.remove_conflict_copy(node, &base);
                     }
                 }
-                Some(_) => {
-                    self.flag_conflict(&node.uid, &node.name, format!("differs from {base}"))
-                }
-                None => self.flag_conflict(
-                    &node.uid,
-                    &node.name,
-                    format!("no original {base} to reconcile against"),
-                ),
+                SweepAction::Flag(detail) => self.flag_conflict(&node.uid, &node.name, detail),
             }
         }
     }
 
     /// Trash a conflict copy proven identical to its live sibling, and unhook it
     /// locally — the same sequence the interactive unlink path uses after a trash.
-    fn remove_conflict_copy(&self, uid: &NodeUid, name: &str, base: &str) {
+    ///
+    /// `snapshot` is the node as the pass saw it, which by now is stale: the pass
+    /// enumerated every node up front and has been doing network I/O since. Before
+    /// touching anything remote this re-reads the node and refuses to act unless
+    /// it is *still* the same revision, still untrashed, still unwritten and still
+    /// unopened. Without that, a user edit landing inside the window is destroyed
+    /// — `discard_queued_ops` below drops the queued write *and its staged blob*,
+    /// which may hold the only copy of those bytes (`docs/BUGS.md` B71).
+    fn remove_conflict_copy(&self, snapshot: &proton_drive_rs::Node, base: &str) {
+        let uid = &snapshot.uid;
+        let name = snapshot.name.as_str();
+        if !self.duplicate_still_removable(snapshot) {
+            return;
+        }
         match self
             .rt
             .block_on(self.client.trash_nodes(std::slice::from_ref(uid)))
@@ -164,6 +171,75 @@ impl Core {
         info!(%uid, name, base, "conflict sweep: removed identical conflict copy");
     }
 
+    /// Re-check, immediately before the destructive call, everything the pass
+    /// decided from its stale snapshot. Any doubt means "leave it alone": a
+    /// conflict copy that survives to the next pass costs nothing, a wrongly
+    /// removed one costs the user data.
+    fn duplicate_still_removable(&self, snapshot: &proton_drive_rs::Node) -> bool {
+        let uid = &snapshot.uid;
+        let name = snapshot.name.as_str();
+
+        // A queued op means bytes are still owed an upload. `discard_queued_ops`
+        // would throw those away along with the staged blob holding them.
+        match self.db.has_any_op(&uid.to_string()) {
+            Ok(false) => {}
+            Ok(true) => {
+                debug!(%uid, name, "conflict sweep: skipping duplicate with queued ops");
+                return false;
+            }
+            // Cannot prove the queue is empty — assume it is not.
+            Err(e) => {
+                warn!(%uid, name, error = ?e, "conflict sweep: op check failed; skipping");
+                return false;
+            }
+        }
+
+        // Someone is writing to it, or holds it open. Trashing now would pull the
+        // node out from under a live handle.
+        if self.is_busy(uid) {
+            debug!(%uid, name, "conflict sweep: skipping duplicate that is open or being written");
+            return false;
+        }
+
+        // Finally, the node itself must not have moved on since the snapshot.
+        match self.db.node_by_uid(&uid.to_string()) {
+            Ok(Some(fresh)) => {
+                if fresh.trashed {
+                    return false;
+                }
+                if fresh.name != snapshot.name || fresh.parent_uid != snapshot.parent_uid {
+                    debug!(%uid, name, "conflict sweep: duplicate was renamed or moved; skipping");
+                    return false;
+                }
+                if node_revision_id(&fresh) != node_revision_id(snapshot)
+                    || node_size(&fresh) != node_size(snapshot)
+                    || node_content_sha1(&fresh) != node_content_sha1(snapshot)
+                {
+                    debug!(%uid, name, "conflict sweep: duplicate changed since the pass began");
+                    return false;
+                }
+                true
+            }
+            // Already gone locally, or unreadable — either way, not ours to remove.
+            Ok(None) => false,
+            Err(e) => {
+                warn!(%uid, name, error = ?e, "conflict sweep: re-read failed; skipping");
+                false
+            }
+        }
+    }
+
+    /// Whether a node has a write in flight or a file handle open against it.
+    /// Read-only opens use fh 0 and are not tracked, so this is specifically the
+    /// writer/holder check.
+    fn is_busy(&self, uid: &NodeUid) -> bool {
+        let state = self.state.lock();
+        let Some(&ino) = state.by_uid.get(uid) else {
+            return false;
+        };
+        state.active_writes.contains_key(&ino) || state.handles.values().any(|&held| held == ino)
+    }
+
     /// Surface a conflict copy that needs the user's attention, at most once per
     /// daemon run so the activity feed is not spammed each pass.
     fn flag_conflict(&self, uid: &NodeUid, name: &str, detail: String) {
@@ -175,6 +251,46 @@ impl Core {
     }
 }
 
+/// What the sweep concluded about one `(sync-conflict …)` copy.
+#[derive(Debug, Clone, PartialEq, Eq)]
+enum SweepAction {
+    /// Provably redundant: trash it, subject to the runtime interlocks in
+    /// [`Core::duplicate_still_removable`].
+    Remove,
+    /// Only the user can resolve this one; surface it with this detail.
+    Flag(String),
+}
+
+/// Decide what to do about conflict copy `node` given its live sibling (if any)
+/// and the configured `mode`.
+///
+/// Pure, so the policy that governs deletion is testable without a mount, a
+/// runtime, or an account — the same split used for `revision_changed` and
+/// `is_own_self_supersede`. Everything that can fail or race lives in the caller.
+fn decide(
+    node: &proton_drive_rs::Node,
+    sibling: Option<&proton_drive_rs::Node>,
+    base: &str,
+    mode: SweepMode,
+) -> SweepAction {
+    match sibling {
+        Some(sib) if is_identical(node, sib) => {
+            if mode.is_enforcing() {
+                SweepAction::Remove
+            } else {
+                // Report-only: say what would happen, remove nothing. The point is
+                // to accumulate field evidence that the identity check is right
+                // before it is trusted with deletion (docs/BUGS.md B71).
+                SweepAction::Flag(format!(
+                    "identical to {base}; would be removed (sweep is report-only)"
+                ))
+            }
+        }
+        Some(_) => SweepAction::Flag(format!("differs from {base}")),
+        None => SweepAction::Flag(format!("no original {base} to reconcile against")),
+    }
+}
+
 /// Whether two file nodes provably hold the same bytes: equal size and equal,
 /// present content SHA-1. A missing digest on either side is not proof, so the
 /// sweep treats it as divergent rather than risk removing a real difference.
@@ -191,6 +307,113 @@ fn is_identical(a: &proton_drive_rs::Node, b: &proton_drive_rs::Node) -> bool {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use proton_drive_rs::NodeKind;
+    use proton_drive_rs::proton_sdk::ids::{LinkId, VolumeId};
+
+    /// A file node with the two fields the sweep's identity check reads.
+    fn file(name: &str, size: i64, sha1: Option<&str>) -> proton_drive_rs::Node {
+        proton_drive_rs::Node {
+            uid: NodeUid::new(VolumeId::from("v"), LinkId::from(name)),
+            parent_uid: None,
+            kind: NodeKind::File {
+                media_type: "text/plain".into(),
+                total_size_on_storage: size,
+                active_revision_state: None,
+                active_revision_id: Some("rev".into()),
+                claimed_size: Some(size),
+                claimed_modification_time: None,
+                content_sha1: sha1.map(String::from),
+            },
+            name: name.into(),
+            creation_time: 0,
+            modification_time: 0,
+            trashed: false,
+            is_shared: false,
+            is_shared_publicly: false,
+            signature_email: None,
+            verification: Default::default(),
+        }
+    }
+
+    /// The only input shape that may ever produce a removal: same size, same
+    /// present digest, live sibling, enforcing mode.
+    #[test]
+    fn removes_only_a_provably_identical_copy_when_enforcing() {
+        let copy = file("f (sync-conflict 1).txt", 100, Some("aaa"));
+        let live = file("f.txt", 100, Some("aaa"));
+        assert_eq!(
+            decide(&copy, Some(&live), "f.txt", SweepMode::Enforce),
+            SweepAction::Remove
+        );
+    }
+
+    #[test]
+    fn report_only_never_removes_but_still_says_what_it_would_do() {
+        let copy = file("f (sync-conflict 1).txt", 100, Some("aaa"));
+        let live = file("f.txt", 100, Some("aaa"));
+        // Same inputs as the removal case above; only the mode differs.
+        let SweepAction::Flag(detail) = decide(&copy, Some(&live), "f.txt", SweepMode::Report)
+        else {
+            panic!("report mode must never remove");
+        };
+        assert!(detail.contains("would be removed"), "{detail}");
+        assert!(detail.contains("report-only"), "{detail}");
+    }
+
+    #[test]
+    fn a_differing_size_is_a_real_conflict() {
+        let copy = file("f (sync-conflict 1).txt", 101, Some("aaa"));
+        let live = file("f.txt", 100, Some("aaa"));
+        assert_eq!(
+            decide(&copy, Some(&live), "f.txt", SweepMode::Enforce),
+            SweepAction::Flag("differs from f.txt".into())
+        );
+    }
+
+    #[test]
+    fn a_missing_digest_is_never_proof_of_identity() {
+        // Equal sizes are not enough: two different 100-byte files are common.
+        // A copy the sweep cannot fingerprint must survive, in either direction.
+        for (a, b) in [(None, Some("aaa")), (Some("aaa"), None), (None, None)] {
+            let copy = file("f (sync-conflict 1).txt", 100, a);
+            let live = file("f.txt", 100, b);
+            assert_eq!(
+                decide(&copy, Some(&live), "f.txt", SweepMode::Enforce),
+                SweepAction::Flag("differs from f.txt".into()),
+                "{a:?} vs {b:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn a_differing_digest_at_equal_size_is_a_real_conflict() {
+        let copy = file("f (sync-conflict 1).txt", 100, Some("aaa"));
+        let live = file("f.txt", 100, Some("bbb"));
+        assert_eq!(
+            decide(&copy, Some(&live), "f.txt", SweepMode::Enforce),
+            SweepAction::Flag("differs from f.txt".into())
+        );
+    }
+
+    #[test]
+    fn an_orphaned_copy_is_surfaced_not_removed() {
+        // No live original left to reconcile against — the copy may be the only
+        // surviving version of the file, so it is never the sweep's to remove.
+        let copy = file("f (sync-conflict 1).txt", 100, Some("aaa"));
+        assert_eq!(
+            decide(&copy, None, "f.txt", SweepMode::Enforce),
+            SweepAction::Flag("no original f.txt to reconcile against".into())
+        );
+    }
+
+    #[test]
+    fn identity_needs_both_size_and_digest_to_match() {
+        let a = file("a", 100, Some("aaa"));
+        assert!(is_identical(&a, &file("b", 100, Some("aaa"))));
+        assert!(!is_identical(&a, &file("b", 101, Some("aaa"))));
+        assert!(!is_identical(&a, &file("b", 100, Some("bbb"))));
+        assert!(!is_identical(&a, &file("b", 100, None)));
+    }
 
     #[test]
     fn parses_conflict_names_and_leaves_others_alone() {