fix(downloads): harden recovery, replacement, and cleanup

- make media credential recovery durable and explicit across lifecycle entry points
- protect exact output replacement with platform-aware ownership, fingerprints, locks, and crash-safe quarantine
- permanently remove unfinished assets while preserving safe completed and Torrent cleanup
- add frontend, native, and localization regressions

src-tauri/src/db.rs

@@ -1678,17 +1678,33 @@ pub fn load_ownership(connection: &Connection) -> Result<Vec<(String, String, Ve
         .map_err(|error| format!("failed to prepare ownership query: {error}"))?;
     let rows = statement
         .query_map([], |row| {
-            let primary_path: String = row.get(1)?;
-            let owned_paths = row
-                .get::<_, Option<String>>(2)?
-                .and_then(|paths| serde_json::from_str::<Vec<String>>(&paths).ok())
-                .filter(|paths| !paths.is_empty())
-                .unwrap_or_else(|| vec![primary_path.clone()]);
-            Ok((row.get(0)?, primary_path, owned_paths))
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, Option<String>>(2)?,
+            ))
         })
         .map_err(|error| format!("failed to query ownership data: {error}"))?;
-    rows.collect::<Result<Vec<_>, _>>()
-        .map_err(|error| format!("failed to read ownership data: {error}"))
+    let mut ownership = Vec::new();
+    for row in rows {
+        let (id, primary_path, encoded_paths) =
+            row.map_err(|error| format!("failed to read ownership data: {error}"))?;
+        let owned_paths = match encoded_paths {
+            Some(encoded) => {
+                let paths = serde_json::from_str::<Vec<String>>(&encoded).map_err(|error| {
+                    format!("failed to decode owned paths for download '{id}': {error}")
+                })?;
+                if paths.is_empty() {
+                    vec![primary_path.clone()]
+                } else {
+                    paths
+                }
+            }
+            None => vec![primary_path.clone()],
+        };
+        ownership.push((id, primary_path, owned_paths));
+    }
+    Ok(ownership)
 }
 
 pub fn set_ownership_paths(
@@ -1697,7 +1713,17 @@ pub fn set_ownership_paths(
     primary_path: &str,
     paths: &[String],
 ) -> Result<(), String> {
-    set_ownership_paths_checked(connection, id, primary_path, paths, &[])
+    // The path collision check and both ownership writes must be one SQLite
+    // transaction. Otherwise two concurrent admissions can both observe an
+    // empty registry and claim the same output before either insert becomes
+    // visible to the other.
+    let transaction = connection
+        .unchecked_transaction()
+        .map_err(|error| format!("failed to begin download ownership transaction: {error}"))?;
+    set_ownership_paths_checked(&transaction, id, primary_path, paths, &[])?;
+    transaction
+        .commit()
+        .map_err(|error| format!("failed to commit download ownership transaction: {error}"))
 }
 
 fn set_ownership_paths_checked(
@@ -1718,21 +1744,38 @@ fn set_ownership_paths_checked(
         .map_err(|error| format!("failed to prepare download ownership check: {error}"))?;
     let existing = statement
         .query_map(params![id], |row| {
-            let primary: String = row.get(1)?;
-            let owned = row
-                .get::<_, Option<String>>(2)?
-                .and_then(|value| serde_json::from_str::<Vec<String>>(&value).ok())
-                .unwrap_or_else(|| vec![primary.clone()]);
-            let removal = row
-                .get::<_, Option<String>>(3)?
-                .and_then(|value| serde_json::from_str::<Vec<String>>(&value).ok())
-                .unwrap_or_default();
-            Ok((primary, owned, removal))
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, Option<String>>(2)?,
+                row.get::<_, Option<String>>(3)?,
+            ))
         })
         .map_err(|error| format!("failed to check download ownership paths: {error}"))?;
     for row in existing {
-        let (existing_primary, owned, removal) =
+        let (existing_id, existing_primary, encoded_owned, encoded_removal) =
             row.map_err(|error| format!("failed to read download ownership paths: {error}"))?;
+        let owned = match encoded_owned {
+            Some(encoded) => {
+                let paths = serde_json::from_str::<Vec<String>>(&encoded).map_err(|error| {
+                    format!("failed to decode owned paths for download '{existing_id}': {error}")
+                })?;
+                if paths.is_empty() {
+                    vec![existing_primary.clone()]
+                } else {
+                    paths
+                }
+            }
+            None => vec![existing_primary.clone()],
+        };
+        let removal = match encoded_removal {
+            Some(encoded) => serde_json::from_str::<Vec<String>>(&encoded).map_err(|error| {
+                format!(
+                    "failed to decode removal paths for download '{existing_id}': {error}"
+                )
+            })?,
+            None => Vec::new(),
+        };
         let new_paths = std::iter::once(primary_path)
             .chain(paths.iter().map(String::as_str))
             .chain(removal_paths.iter().map(String::as_str));
@@ -1924,6 +1967,31 @@ pub fn load_torrent_removal_paths(
         .map(|paths| paths.unwrap_or_default())
 }
 
+pub fn load_all_torrent_removal_paths(
+    connection: &Connection,
+) -> Result<Vec<(String, Vec<String>)>, String> {
+    let mut statement = connection
+        .prepare("SELECT id, paths FROM download_removal_paths")
+        .map_err(|error| format!("failed to prepare torrent removal ownership query: {error}"))?;
+    let rows = statement
+        .query_map([], |row| {
+            let id: String = row.get(0)?;
+            let encoded: String = row.get(1)?;
+            Ok((id, encoded))
+        })
+        .map_err(|error| format!("failed to query torrent removal ownership: {error}"))?;
+
+    rows.map(|row| {
+        let (id, encoded) = row
+            .map_err(|error| format!("failed to read torrent removal ownership: {error}"))?;
+        let paths = serde_json::from_str::<Vec<String>>(&encoded).map_err(|error| {
+            format!("failed to decode torrent removal paths for download '{id}': {error}")
+        })?;
+        Ok((id, paths))
+    })
+    .collect()
+}
+
 pub fn has_user_data(connection: &Connection) -> Result<bool, String> {
     connection
         .query_row(
@@ -3471,6 +3539,90 @@ mod tests {
         .unwrap();
     }
 
+    #[test]
+    fn malformed_owned_path_json_fails_closed_for_loading_and_new_claims() {
+        let temp = TempDir::new().unwrap();
+        let state = init_at_path(temp.path()).unwrap();
+        let connection = state.lock().unwrap();
+        connection
+            .execute(
+                "INSERT INTO download_ownership (id, primary_path) VALUES (?1, ?2)",
+                params!["broken-owned", "/downloads/broken.bin"],
+            )
+            .unwrap();
+        connection
+            .execute(
+                "INSERT INTO download_owned_paths (id, paths) VALUES (?1, ?2)",
+                params!["broken-owned", "{not-json"],
+            )
+            .unwrap();
+
+        let error =
+            load_ownership(&connection).expect_err("malformed owned paths must not be ignored");
+        assert!(error.contains("broken-owned"));
+
+        let error = set_ownership_paths(
+            &connection,
+            "later",
+            "/downloads/later.bin",
+            &["/downloads/later.bin".to_string()],
+        )
+        .expect_err("new ownership claims must fail closed");
+        assert!(error.contains("broken-owned"));
+        assert_eq!(
+            connection
+                .query_row(
+                    "SELECT COUNT(*) FROM download_ownership WHERE id = 'later'",
+                    [],
+                    |row| row.get::<_, i64>(0),
+                )
+                .unwrap(),
+            0
+        );
+    }
+
+    #[test]
+    fn malformed_removal_path_json_fails_closed_for_loading_and_new_claims() {
+        let temp = TempDir::new().unwrap();
+        let state = init_at_path(temp.path()).unwrap();
+        let connection = state.lock().unwrap();
+        connection
+            .execute(
+                "INSERT INTO download_ownership (id, primary_path) VALUES (?1, ?2)",
+                params!["broken-removal", "/downloads/broken.bin"],
+            )
+            .unwrap();
+        connection
+            .execute(
+                "INSERT INTO download_removal_paths (id, paths) VALUES (?1, ?2)",
+                params!["broken-removal", "{not-json"],
+            )
+            .unwrap();
+
+        let error = load_all_torrent_removal_paths(&connection)
+            .expect_err("malformed removal paths must not be ignored");
+        assert!(error.contains("broken-removal"));
+
+        let error = set_ownership_paths(
+            &connection,
+            "later",
+            "/downloads/later.bin",
+            &["/downloads/later.bin".to_string()],
+        )
+        .expect_err("new ownership claims must fail closed");
+        assert!(error.contains("broken-removal"));
+        assert_eq!(
+            connection
+                .query_row(
+                    "SELECT COUNT(*) FROM download_ownership WHERE id = 'later'",
+                    [],
+                    |row| row.get::<_, i64>(0),
+                )
+                .unwrap(),
+            0
+        );
+    }
+
     #[test]
     fn torrent_ownership_and_removal_reservation_commit_atomically() {
         let temp = TempDir::new().unwrap();

src-tauri/src/download_ownership.rs

@@ -346,10 +346,27 @@ pub fn known_primary_paths<R: tauri::Runtime>(
         })
         .collect();
 
-    // One-time compatibility for downloads created before the backend-owned
-    // registry existed. This imports the exact persisted queue path only.
-    for path in legacy_download_queue_paths(app_handle)? {
-        if !paths.iter().any(|existing| existing == &path) {
+    let database = app_handle.state::<crate::db::DbState>();
+    let connection = database.lock()?;
+    for (_, removal_paths) in crate::db::load_all_torrent_removal_paths(&connection)? {
+        for path in removal_paths.into_iter().map(PathBuf::from) {
+            if !paths
+                .iter()
+                .any(|existing| crate::platform::paths_equal(existing, &path))
+            {
+                paths.push(path);
+            }
+        }
+    }
+    drop(connection);
+
+    // Compatibility for downloads created before the backend-owned registry
+    // existed. Import only the exact persisted queue paths.
+    for (_, path) in legacy_download_queue_path_records(app_handle)? {
+        if !paths
+            .iter()
+            .any(|existing| crate::platform::paths_equal(existing, &path))
+        {
             paths.push(path);
         }
     }
@@ -357,6 +374,65 @@ pub fn known_primary_paths<R: tauri::Runtime>(
     Ok(paths)
 }
 
+/// Return the Firelink download that owns an exact output path, if any.
+///
+/// This is intentionally based on the persisted ownership registry rather
+/// than on the visible download list. The renderer can be stale while a
+/// queued/native lifecycle is being admitted, so duplicate replacement must
+/// make this decision at the native boundary.
+pub fn owner_for_path<R: tauri::Runtime>(
+    app_handle: &tauri::AppHandle<R>,
+    path: &Path,
+) -> Result<Option<String>, String> {
+    let canonical = crate::canonicalize_with_missing_components(path)
+        .ok_or_else(|| "Download target could not be canonicalized".to_string())?;
+    let mut owners = Vec::new();
+    for record in load_records(app_handle)? {
+        let primary = PathBuf::from(&record.primary_path);
+        if crate::platform::paths_equal(&primary, &canonical)
+            || record
+                .owned_paths
+                .iter()
+                .map(PathBuf::from)
+                .any(|owned| crate::platform::paths_equal(&owned, &canonical))
+        {
+            owners.push(record.id);
+        }
+    }
+
+    let database = app_handle.state::<crate::db::DbState>();
+    let connection = database.lock()?;
+    for (id, removal_paths) in crate::db::load_all_torrent_removal_paths(&connection)? {
+        if removal_paths
+            .into_iter()
+            .map(PathBuf::from)
+            .any(|removal| crate::platform::paths_equal(&removal, &canonical))
+            && !owners.contains(&id)
+        {
+            owners.push(id);
+        }
+    }
+    drop(connection);
+
+    // Older rows may predate the ownership registry. They still represent
+    // Firelink-owned targets and must not be downgraded to unmanaged disk
+    // files merely because their migration record is absent.
+    for (id, legacy_path) in legacy_download_queue_path_records(app_handle)? {
+        if crate::platform::paths_equal(&legacy_path, &canonical) && !owners.contains(&id) {
+            owners.push(id);
+        }
+    }
+
+    match owners.len() {
+        0 => Ok(None),
+        1 => Ok(owners.pop()),
+        _ => Err(format!(
+            "Download target is claimed by multiple Firelink downloads: {}",
+            owners.join(", ")
+        )),
+    }
+}
+
 fn load_records<R: tauri::Runtime>(app_handle: &tauri::AppHandle<R>) -> Result<Vec<DownloadOwnershipRecord>, String> {
     let database = app_handle.state::<crate::db::DbState>();
     let connection = database.lock()?;
@@ -372,9 +448,9 @@ fn load_records<R: tauri::Runtime>(app_handle: &tauri::AppHandle<R>) -> Result<V
     })
 }
 
-fn legacy_download_queue_paths<R: tauri::Runtime>(
+fn legacy_download_queue_path_records<R: tauri::Runtime>(
     app_handle: &tauri::AppHandle<R>,
-) -> Result<Vec<PathBuf>, String> {
+) -> Result<Vec<(String, PathBuf)>, String> {
     let settings = crate::settings::load_settings(app_handle).ok();
 
     let downloads = {
@@ -383,7 +459,7 @@ fn legacy_download_queue_paths<R: tauri::Runtime>(
         parse_legacy_download_items(crate::db::load_downloads(&connection)?)
     };
 
-    let mut paths = Vec::new();
+    let mut paths: Vec<(String, PathBuf)> = Vec::new();
     for download in downloads {
         let category = format!("{:?}", download.category);
         let mut destinations = Vec::new();
@@ -442,8 +518,10 @@ fn legacy_download_queue_paths<R: tauri::Runtime>(
 
         for destination in destinations {
             if let Ok(path) = expected_primary_path(app_handle, &destination, &download.file_name) {
-                if !paths.iter().any(|existing| existing == &path) {
-                    paths.push(path);
+                if !paths.iter().any(|(id, existing)| {
+                    id == &download.id && crate::platform::paths_equal(existing, &path)
+                }) {
+                    paths.push((download.id.clone(), path));
                 }
             }
         }

src-tauri/src/ipc.rs

@@ -153,6 +153,38 @@ pub enum DownloadErrorKind {
     DestinationAccess,
 }
 
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export, export_to = "../../src/bindings/")]
+pub enum DownloadTargetKind {
+    Missing,
+    RegularFile,
+    Directory,
+    Symlink,
+    Special,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export, export_to = "../../src/bindings/")]
+pub struct DownloadTargetInfo {
+    pub kind: DownloadTargetKind,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[ts(optional)]
+    pub fingerprint: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[ts(optional)]
+    pub owned_by: Option<String>,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(export, export_to = "../../src/bindings/")]
+pub enum DownloadAssetRemovalPolicy {
+    Trash,
+    PermanentIfUnfinished,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize, TS)]
 #[serde(rename_all = "camelCase")]
 #[ts(export, export_to = "../../src/bindings/")]
@@ -222,6 +254,9 @@ pub struct DownloadItem {
     #[serde(default, skip_serializing_if = "Option::is_none")]
     #[ts(optional)]
     pub last_resolver_fallback: Option<bool>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    #[ts(optional)]
+    pub replace_existing_fingerprint: Option<String>,
     #[ts(optional)]
     pub last_try: Option<String>,
     #[serde(default)]

src-tauri/src/platform.rs

@@ -305,39 +305,115 @@ fn trusted_system_path_entries() -> Vec<PathBuf> {
 pub fn path_is_within(path: &Path, root: &Path) -> bool {
     #[cfg(target_os = "windows")]
     {
-        let path = path.to_string_lossy().to_lowercase();
-        let root = root.to_string_lossy().to_lowercase();
+        let path = path_identity(path);
+        let root = path_identity(root);
         path == root
+            || (root.len() == 3
+                && root.ends_with('/')
+                && root.as_bytes()[1] == b':'
+                && path.starts_with(&root))
             || path
                 .strip_prefix(&root)
-                .is_some_and(|suffix| suffix.starts_with(['\\', '/']))
+                .is_some_and(|suffix| suffix.starts_with('/'))
     }
 
-    #[cfg(not(target_os = "windows"))]
+    #[cfg(target_os = "macos")]
+    {
+        // Containment is a scope check, not an equality check. Do not fold
+        // case here: case-sensitive APFS/HFS+ volumes are valid macOS
+        // configurations, and lowercasing could admit `/Users/nima2` or a
+        // differently-cased sibling outside the approved root. Callers pass
+        // canonical paths (with only missing leaf components preserved), so
+        // NFC normalization is enough to compare macOS path spellings.
+        use unicode_normalization::UnicodeNormalization;
+
+        let path = path.to_string_lossy().nfc().collect::<String>();
+        let root = root.to_string_lossy().nfc().collect::<String>();
+        let root = root.trim_end_matches('/');
+        if path == root || (root.is_empty() && path == "/") {
+            return true;
+        }
+
+        if root.is_empty() {
+            return path.starts_with('/');
+        }
+
+        path.strip_prefix(root)
+            .is_some_and(|suffix| suffix.starts_with('/'))
+    }
+
+    #[cfg(all(unix, not(target_os = "macos")))]
+    {
+        path.starts_with(root)
+    }
+
+    #[cfg(not(any(unix, target_os = "windows", target_os = "macos")))]
     {
         path.starts_with(root)
     }
 }
 
 pub fn paths_equal(left: &Path, right: &Path) -> bool {
+    path_identity(left) == path_identity(right)
+}
+
+/// Return the in-process lock identity for a path using the same platform
+/// equivalence rules as `paths_equal`. Callers use this for serialization,
+/// not for display or persistence.
+pub fn path_identity(path: &Path) -> String {
     #[cfg(target_os = "windows")]
     {
-        left.to_string_lossy()
-            .to_lowercase()
-            == right.to_string_lossy().to_lowercase()
+        let mut normalized = path.to_string_lossy().replace('\\', "/");
+        if normalized
+            .get(..8)
+            .is_some_and(|prefix| prefix.eq_ignore_ascii_case("//?/UNC/"))
+        {
+            normalized.replace_range(..8, "//");
+        } else if normalized
+            .get(..4)
+            .is_some_and(|prefix| prefix.eq_ignore_ascii_case("//?/"))
+        {
+            normalized.replace_range(..4, "");
+        }
+
+        let is_unc = normalized.starts_with("//");
+        let mut collapsed = String::with_capacity(normalized.len());
+        for character in normalized.chars() {
+            if character == '/' && collapsed.ends_with('/') && !(is_unc && collapsed.len() == 1) {
+                continue;
+            }
+            collapsed.push(character);
+        }
+        while collapsed.len() > 1
+            && collapsed.ends_with('/')
+            && !(collapsed.len() == 3 && collapsed.as_bytes()[1] == b':')
+        {
+            collapsed.pop();
+        }
+        collapsed.to_lowercase()
     }
     #[cfg(target_os = "macos")]
     {
         use unicode_normalization::UnicodeNormalization;
 
-        let normalize = |path: &Path| {
-            path.to_string_lossy().to_lowercase().nfc().collect::<String>()
-        };
-        normalize(left) == normalize(right)
+        path.to_string_lossy()
+            .to_lowercase()
+            .nfc()
+            .collect::<String>()
     }
-    #[cfg(not(any(target_os = "windows", target_os = "macos")))]
+    #[cfg(all(unix, not(target_os = "macos")))]
     {
-        left == right
+        use std::os::unix::ffi::OsStrExt;
+
+        path.as_os_str()
+            .as_bytes()
+            .iter()
+            .map(|byte| format!("{byte:02x}"))
+            .collect()
+    }
+    #[cfg(not(any(unix, target_os = "windows", target_os = "macos")))]
+    {
+        path.to_string_lossy().to_string()
     }
 }
 
@@ -362,6 +438,8 @@ fn numbered_windows_device(stem: &str, prefix: &str) -> bool {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(any(target_os = "windows", target_os = "macos"))]
+    use super::path_is_within;
     use super::{engine_binary_name, is_windows_reserved_filename, paths_equal, target_triple};
     use std::path::Path;
 
@@ -406,6 +484,27 @@ mod tests {
         }
     }
 
+    #[cfg(target_os = "windows")]
+    #[test]
+    fn windows_path_identity_normalizes_separators_and_verbatim_prefixes() {
+        assert!(paths_equal(
+            Path::new(r"C:\downloads\file.bin"),
+            Path::new("c:/DOWNLOADS/file.bin")
+        ));
+        assert!(paths_equal(
+            Path::new(r"C:\downloads\file.bin"),
+            Path::new(r"\\?\C:\downloads\file.bin")
+        ));
+        assert!(paths_equal(
+            Path::new(r"\\server\share\file.bin"),
+            Path::new(r"\\?\UNC\server\share\file.bin")
+        ));
+        assert!(path_is_within(
+            Path::new("c:/downloads/file.bin"),
+            Path::new(r"C:\downloads")
+        ));
+    }
+
     #[test]
     fn path_identity_handles_non_ascii_case_differences() {
         let left = Path::new("/downloads/Ärt/File.bin");
@@ -427,4 +526,34 @@ mod tests {
             assert!(!paths_equal(composed, decomposed));
         }
     }
+
+    #[cfg(target_os = "macos")]
+    #[test]
+    fn macos_path_is_within_preserves_scope_and_unicode_identity() {
+        assert!(path_is_within(
+            Path::new("/Downloads/cafe\u{301}/movie.bin"),
+            Path::new("/Downloads/café")
+        ));
+        assert!(path_is_within(
+            Path::new("/Downloads/movie.bin"),
+            Path::new("/Downloads")
+        ));
+        assert!(path_is_within(
+            Path::new("/Downloads"),
+            Path::new("/Downloads/")
+        ));
+        assert!(path_is_within(Path::new("/"), Path::new("////")));
+        assert!(path_is_within(
+            Path::new("/Downloads/movie.bin"),
+            Path::new("/")
+        ));
+        assert!(!path_is_within(
+            Path::new("/downloads/cafeteria/movie.bin"),
+            Path::new("/Downloads/café")
+        ));
+        assert!(!path_is_within(
+            Path::new("/downloads/movie.bin"),
+            Path::new("/Downloads")
+        ));
+    }
 }

src-tauri/src/queue.rs

@@ -2121,24 +2121,93 @@ impl<R: tauri::Runtime> QueueManager<R> {
     }
 
     pub async fn commit_reserved_enqueue(
+        &self,
+        task: QueuedTask,
+        generation: u64,
+        previous_generation: Option<u64>,
+    ) -> Result<(), String> {
+        self.commit_reserved_enqueue_with_finalizer(task, generation, previous_generation, || async {
+            Ok(())
+        })
+        .await
+    }
+
+    /// Commit an enqueue and its final durable admission marker as one
+    /// dispatcher-visible boundary. The task is placed in the pending list
+    /// before the finalizer runs, but the admission gate stays held so the
+    /// dispatcher cannot pop it until the finalizer succeeds. If the
+    /// finalizer fails, the task is removed before any worker can observe it.
+    pub async fn commit_reserved_enqueue_with_finalizer<F, Fut>(
         &self,
         mut task: QueuedTask,
         generation: u64,
-    ) -> Result<(), String> {
+        previous_generation: Option<u64>,
+        finalizer: F,
+    ) -> Result<(), String>
+    where
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = Result<(), String>>,
+    {
+        let id = task.id.clone();
         let _admission_gate = self.admission_gate.lock().await;
         if self.system_action_pending.load(Ordering::Acquire) {
+            self.rollback_enqueue_reservation(&id, generation, previous_generation)
+                .await;
             return Err("System action is already being performed".to_string());
         }
-        let id = task.id.clone();
-        let cancellations = self.enqueue_cancellations.lock().await;
-        if cancellations
-            .get(&id)
-            .is_some_and(|cancelled| *cancelled >= generation)
+        if self
+            .registered_lifecycle_generation(&id)
+            .await
+            .is_none_or(|registered| registered != generation)
         {
-            return Err("Download enqueue was superseded by a newer user action".to_string());
+            self.rollback_enqueue_reservation(&id, generation, previous_generation)
+                .await;
+            return Err("Download enqueue reservation is no longer current".to_string());
+        }
+        {
+            let cancellations = self.enqueue_cancellations.lock().await;
+            if cancellations
+                .get(&id)
+                .is_some_and(|cancelled| *cancelled >= generation)
+            {
+                self.rollback_enqueue_reservation(&id, generation, previous_generation)
+                    .await;
+                return Err("Download enqueue was superseded by a newer user action".to_string());
+            }
         }
         task.lifecycle_generation = generation;
         self.pending.lock().await.push_back(task);
+
+        if let Err(error) = finalizer().await {
+            let mut pending = self.pending.lock().await;
+            pending.retain(|candidate| {
+                !(candidate.id == id && candidate.lifecycle_generation == generation)
+            });
+            self.rollback_enqueue_reservation(&id, generation, previous_generation)
+                .await;
+            return Err(error);
+        }
+
+        // Cancellation can arrive while the durable admission marker is being
+        // written. Recheck it before making the task visible to the rest of
+        // the lifecycle; the admission gate prevents a dispatcher or queue
+        // mutation from observing a half-committed replacement.
+        {
+            let cancellations = self.enqueue_cancellations.lock().await;
+            if cancellations
+                .get(&id)
+                .is_some_and(|cancelled| *cancelled >= generation)
+            {
+                let mut pending = self.pending.lock().await;
+                pending.retain(|candidate| {
+                    !(candidate.id == id && candidate.lifecycle_generation == generation)
+                });
+                self.rollback_enqueue_reservation(&id, generation, previous_generation)
+                    .await;
+                return Err("Download enqueue was superseded by a newer user action".to_string());
+            }
+        }
+
         self.emit_state(id, DownloadStatus::Queued);
         self.notify.notify_one();
         Ok(())
@@ -2152,7 +2221,10 @@ impl<R: tauri::Runtime> QueueManager<R> {
     ) -> Result<(), String> {
         let id = task.id.clone();
         let previous_generation = self.reserve_enqueue_generation(&id, generation).await?;
-        if let Err(error) = self.commit_reserved_enqueue(task, generation).await {
+        if let Err(error) = self
+            .commit_reserved_enqueue(task, generation, previous_generation)
+            .await
+        {
             self.rollback_enqueue_reservation(&id, generation, previous_generation)
                 .await;
             return Err(error);
@@ -3395,6 +3467,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
 
     /// Pop the next task, or None if empty.
     pub async fn pop_front(&self) -> Option<QueuedTask> {
+        let _admission_gate = self.admission_gate.lock().await;
         self.pending.lock().await.pop_front()
     }
 
@@ -5789,6 +5862,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
         queue_id: &str,
         direction: QueueDirection,
     ) -> Vec<String> {
+        let _admission_gate = self.admission_gate.lock().await;
         let mut pending = self.pending.lock().await;
         let queue_positions = pending
             .iter()
@@ -5852,6 +5926,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
         queue_id: &str,
         target_index: usize,
     ) -> Vec<String> {
+        let _admission_gate = self.admission_gate.lock().await;
         let mut pending = self.pending.lock().await;
         let queue_positions = pending
             .iter()
@@ -5880,6 +5955,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
     /// Does NOT release a permit (the caller handles active permits via
     /// release_permit if the task was already dispatched).
     pub async fn remove_from_pending(&self, id: &str) -> bool {
+        let _admission_gate = self.admission_gate.lock().await;
         let mut pending = self.pending.lock().await;
         let before = pending.len();
         pending.retain(|t| t.id != id);
@@ -5891,6 +5967,7 @@ impl<R: tauri::Runtime> QueueManager<R> {
     }
 
     pub async fn remove_from_pending_for_generation(&self, id: &str, generation: u64) -> bool {
+        let _admission_gate = self.admission_gate.lock().await;
         let mut pending = self.pending.lock().await;
         let before = pending.len();
         pending.retain(|task| !(task.id == id && task.lifecycle_generation == generation));
@@ -8556,6 +8633,9 @@ pub struct EnqueueItem {
     #[serde(default)]
     #[ts(optional)]
     pub lifecycle_generation: Option<String>,
+    #[serde(default)]
+    #[ts(optional)]
+    pub replace_existing_fingerprint: Option<String>,
 }
 
 impl EnqueueItem {
@@ -8756,7 +8836,7 @@ mod tests {
                 release: Arc::clone(&release),
             }),
         ));
-        manager
+        let previous_generation = manager
             .reserve_enqueue_generation("allocation", 7)
             .await
             .expect("lifecycle reservation");
@@ -8770,6 +8850,7 @@ mod tests {
                     lifecycle_generation: 7,
                 },
                 7,
+                previous_generation,
             )
             .await
             .expect("queued task");
@@ -8796,6 +8877,106 @@ mod tests {
         dispatcher.abort();
     }
 
+    #[tokio::test]
+    async fn enqueue_finalizer_failure_removes_pending_task_before_dispatch() {
+        let app = tauri::test::mock_builder()
+            .build(tauri::test::mock_context(tauri::test::noop_assets()))
+            .expect("mock app");
+        let manager = QueueManager::test_new(app.handle().clone(), 1, Arc::new(TestSpawner));
+        let id = "finalizer-failure";
+        let generation = 3;
+        let previous_generation = manager
+            .reserve_enqueue_generation(id, generation)
+            .await
+            .expect("lifecycle reservation");
+
+        let error = manager
+            .commit_reserved_enqueue_with_finalizer(
+                QueuedTask {
+                    id: id.to_string(),
+                    queue_id: "main".to_string(),
+                    kind: TaskKind::Aria2,
+                    payload: SpawnPayload::default(),
+                    lifecycle_generation: generation,
+                    },
+                    generation,
+                    previous_generation,
+                    || async { Err("journal commit failed".to_string()) },
+            )
+            .await
+            .expect_err("a failed finalizer must reject admission");
+
+        assert_eq!(error, "journal commit failed");
+        assert!(manager.pending_order(None).await.is_empty());
+        assert_eq!(manager.registered_lifecycle_generation(id).await, None);
+        assert!(!manager.is_registered(id).await);
+    }
+
+    #[tokio::test]
+    async fn enqueue_cancellation_during_finalizer_rejects_admission() {
+        let app = tauri::test::mock_builder()
+            .build(tauri::test::mock_context(tauri::test::noop_assets()))
+            .expect("mock app");
+        let manager = Arc::new(QueueManager::test_new(
+            app.handle().clone(),
+            1,
+            Arc::new(TestSpawner),
+        ));
+        let started = Arc::new(tokio::sync::Notify::new());
+        let release = Arc::new(tokio::sync::Notify::new());
+        let finalizer_started = Arc::clone(&started);
+        let finalizer_release = Arc::clone(&release);
+        let id = "finalizer-cancelled".to_string();
+        let generation = 4;
+        let previous_generation = manager
+            .reserve_enqueue_generation(&id, generation)
+            .await
+            .expect("lifecycle reservation");
+
+        let commit_manager = Arc::clone(&manager);
+        let commit_id = id.clone();
+        let commit = tokio::spawn(async move {
+            commit_manager
+                .commit_reserved_enqueue_with_finalizer(
+                    QueuedTask {
+                        id: commit_id,
+                        queue_id: "main".to_string(),
+                        kind: TaskKind::Aria2,
+                        payload: SpawnPayload::default(),
+                        lifecycle_generation: generation,
+                        },
+                        generation,
+                        previous_generation,
+                        {
+                        move || async move {
+                            finalizer_started.notify_one();
+                            finalizer_release.notified().await;
+                            Ok(())
+                        }
+                    },
+                )
+                .await
+        });
+
+        tokio::time::timeout(Duration::from_secs(1), started.notified())
+            .await
+            .expect("finalizer should begin");
+        manager.cancel_enqueue_generation(&id, generation).await;
+        release.notify_one();
+
+        let error = tokio::time::timeout(Duration::from_secs(1), commit)
+            .await
+            .expect("enqueue should finish")
+            .expect("enqueue task should not panic")
+            .expect_err("cancellation must reject the in-flight admission");
+        assert_eq!(
+            error,
+            "Download enqueue was superseded by a newer user action"
+        );
+        assert!(manager.pending_order(None).await.is_empty());
+        assert!(!manager.is_registered(&id).await);
+    }
+
     #[tokio::test]
     async fn download_start_before_gid_registration_is_buffered_and_consumed() {
         let app = tauri::test::mock_builder()

src-tauri/tests/queue_manager.rs

@@ -433,7 +433,9 @@ async fn cancellation_between_reservation_and_commit_cannot_start_the_task() {
         .expect("reservation should succeed");
     mgr.cancel_enqueue_generation("a", 7).await;
 
-    let committed = mgr.commit_reserved_enqueue(sample_task("a"), 7).await;
+    let committed = mgr
+        .commit_reserved_enqueue(sample_task("a"), 7, previous)
+        .await;
     assert!(committed.is_err(), "cancelled reservation must not commit");
     mgr.rollback_enqueue_reservation("a", 7, previous).await;
 
@@ -1799,12 +1801,16 @@ async fn duplicate_pending_id_cannot_replace_an_existing_queue_ownership() {
     assert!(manager
         .ensure_aria2_permit_for_queue("duplicate", "queue-b")
         .await);
-    manager
+    let previous = manager
         .reserve_enqueue_generation("duplicate", 1)
         .await
         .unwrap();
     manager
-        .commit_reserved_enqueue(aria2_task_in_queue("duplicate", "queue-a"), 1)
+        .commit_reserved_enqueue(
+            aria2_task_in_queue("duplicate", "queue-a"),
+            1,
+            previous,
+        )
         .await
         .unwrap();