From a24aa98cf0f9f03a120ed8a31d59a6985ace97dc Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sat, 1 Aug 2026 23:44:44 -0400
Subject: [PATCH 1/7] =?UTF-8?q?feat(transfers):=20real=20queue=20=E2=80=94?=
 =?UTF-8?q?=20bounded=20concurrency,=20FIFO,=20move,=20pause-all?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plan 17 Phase 1: Queued now genuinely waits. TransferManager gains a FIFO
(waiting deque + watch-channel generation), a semaphore bounding concurrent
transfers (default 3, live-adjustable via transfer_set_concurrency and the
transferConcurrency setting), FIFO admission with a manager-level pause-all
gate, transfer_move for queue priority, and a transfer://queue event plus
transfer_queue_state for the panel. cancel removes waiting ids from the FIFO
and reports via the new transfer://updated event.
---
 src-tauri/src/commands.rs |  61 ++++++++++-
 src-tauri/src/lib.rs      |  17 +++
 src-tauri/src/transfer.rs | 212 +++++++++++++++++++++++++++++++++++++-
 3 files changed, 286 insertions(+), 4 deletions(-)

diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs
index 2e74829..44d4c1c 100644
--- a/src-tauri/src/commands.rs
+++ b/src-tauri/src/commands.rs
@@ -737,9 +737,68 @@ pub async fn start_upload(
 #[tauri::command]
 pub async fn cancel_transfer(
     transfer_id: String,
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state
+        .transfers
+        .cancel(&transfer_id, &app)
+        .await
+        .map_err(err)
+}
+
+/// Reorder a waiting transfer in the queue FIFO (Plan 17). `direction` is
+/// "up" (sooner) or "down" (later); active transfers are untouched.
+#[tauri::command]
+pub async fn transfer_move(
+    transfer_id: String,
+    direction: String,
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state
+        .transfers
+        .move_in_queue(&transfer_id, direction == "up", &app)
+        .await
+        .map_err(err)
+}
+
+/// Pause admission of new transfers; running ones keep going.
+#[tauri::command]
+pub async fn transfer_pause_all(
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state.transfers.pause_all(&app).await;
+    Ok(())
+}
+
+#[tauri::command]
+pub async fn transfer_resume_all(
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state.transfers.resume_all(&app).await;
+    Ok(())
+}
+
+/// Live-adjust how many transfers may run at once (1..=32).
+#[tauri::command]
+pub async fn transfer_set_concurrency(
+    count: u32,
     state: State<'_, AppState>,
 ) -> Result<(), String> {
-    state.transfers.cancel(&transfer_id).await.map_err(err)
+    state.transfers.set_concurrency(count as usize);
+    Ok(())
+}
+
+/// Queue snapshot for the panel's initial load (waiting FIFO + pause-all +
+/// concurrency + throttle).
+#[tauri::command]
+pub async fn transfer_queue_state(
+    state: State<'_, AppState>,
+) -> Result<crate::transfer::QueueState, String> {
+    Ok(state.transfers.queue_state().await)
 }
 
 #[tauri::command]
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index 8dcc94f..8009e9d 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -250,6 +250,18 @@ pub fn run() {
             };
             app.manage(state);
 
+            // Apply persisted transfer-queue settings (Plan 17). The frontend
+            // writes `transferConcurrency`/`transferThrottleKbps` via the
+            // settings table; the manager starts from those values.
+            {
+                let st = app.state::<AppState>();
+                if let Ok(Some(raw)) = st.db.settings_get("transferConcurrency") {
+                    if let Ok(n) = serde_json::from_str::<usize>(&raw) {
+                        st.transfers.set_concurrency(n);
+                    }
+                }
+            }
+
             // Bring the Agent Bridge back up if the user left its master switch
             // on, so the `faro-cli agent …` path keeps working across restarts.
             // Spawned off the async runtime so the sync setup() returns at once.
@@ -389,6 +401,11 @@ pub fn run() {
             commands::start_directory_upload,
             commands::cancel_transfer,
             commands::list_transfers,
+            commands::transfer_move,
+            commands::transfer_pause_all,
+            commands::transfer_resume_all,
+            commands::transfer_set_concurrency,
+            commands::transfer_queue_state,
             commands::rename_path,
             commands::delete_path,
             commands::create_directory,
diff --git a/src-tauri/src/transfer.rs b/src-tauri/src/transfer.rs
index 7830b64..6adf311 100644
--- a/src-tauri/src/transfer.rs
+++ b/src-tauri/src/transfer.rs
@@ -5,13 +5,14 @@ use crate::session::{
 };
 use anyhow::{Context, Result};
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::collections::{HashMap, VecDeque};
 use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tauri::{AppHandle, Emitter};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
-use tokio::sync::Mutex;
+use tokio::sync::{watch, Mutex, OwnedSemaphorePermit, Semaphore};
 use tokio::task::JoinHandle;
 use uuid::Uuid;
 
@@ -42,6 +43,47 @@ pub enum TransferStatus {
     Canceled,
 }
 
+/// Payload of the `transfer://queue` event: the FIFO of waiting transfer ids
+/// plus the manager-level state the panel header renders (Plan 17).
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct QueueState {
+    pub waiting: Vec<String>,
+    pub paused_all: bool,
+    pub concurrency: usize,
+    pub throttle_kbps: u64,
+}
+
+/// A pause gate shared by the scheduler (pause-all) and individual transfers
+/// (Phase 2). watch-channel based so waiters never miss a wakeup.
+#[derive(Debug, Clone)]
+pub struct PauseGate {
+    tx: watch::Sender<bool>,
+}
+
+impl PauseGate {
+    fn new() -> Self {
+        let (tx, _) = watch::channel(false);
+        Self { tx }
+    }
+    fn is_paused(&self) -> bool {
+        *self.tx.borrow()
+    }
+    fn set(&self, paused: bool) {
+        let _ = self.tx.send(paused);
+    }
+    /// Park until the gate opens. Returns immediately if already open.
+    #[allow(dead_code)] // used by the Phase 2 chunk checkpoints
+    async fn wait_open(&self) {
+        let mut rx = self.tx.subscribe();
+        while *rx.borrow_and_update() {
+            if rx.changed().await.is_err() {
+                break;
+            }
+        }
+    }
+}
+
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "camelCase")]
 pub struct Transfer {
@@ -57,9 +99,23 @@ pub struct Transfer {
     pub started_at: i64,
 }
 
+/// Default bound on concurrently running transfers (Plan 17); the rest wait
+/// in the FIFO as `Queued`. Overridden by the `transferConcurrency` setting.
+const DEFAULT_CONCURRENCY: usize = 3;
+
 pub struct TransferManager {
     transfers: Mutex<HashMap<String, Transfer>>,
     tasks: Mutex<HashMap<String, JoinHandle<()>>>,
+    /// FIFO of transfer ids waiting for a slot (Plan 17). An id is popped only
+    /// once it is at the front, the pause-all gate is open, and a concurrency
+    /// permit is available — until then it waits as `Queued`.
+    waiting: Mutex<VecDeque<String>>,
+    semaphore: Arc<Semaphore>,
+    concurrency: AtomicUsize,
+    /// Manager-level pause gate. Admission checks it; Phase 2 checkpoints do too.
+    pause_all: PauseGate,
+    /// Bumped on every queue change so admission waiters re-check their turn.
+    queue_gen: watch::Sender<u64>,
 }
 
 fn now_ts() -> i64 {
@@ -127,6 +183,11 @@ impl TransferManager {
         Self {
             transfers: Mutex::new(HashMap::new()),
             tasks: Mutex::new(HashMap::new()),
+            waiting: Mutex::new(VecDeque::new()),
+            semaphore: Arc::new(Semaphore::new(DEFAULT_CONCURRENCY)),
+            concurrency: AtomicUsize::new(DEFAULT_CONCURRENCY),
+            pause_all: PauseGate::new(),
+            queue_gen: watch::channel(0).0,
         }
     }
 
@@ -156,7 +217,133 @@ impl TransferManager {
         self.transfers.lock().await.insert(t.id.clone(), t);
     }
 
-    pub async fn cancel(&self, id: &str) -> Result<()> {
+    // ---------- Queue scheduling (Plan 17) ----------
+
+    fn build_queue_state(&self, waiting: &VecDeque<String>) -> QueueState {
+        QueueState {
+            waiting: waiting.iter().cloned().collect(),
+            paused_all: self.pause_all.is_paused(),
+            concurrency: self.concurrency.load(Ordering::Relaxed),
+            throttle_kbps: 0, // wired up in Phase 4
+        }
+    }
+
+    /// Current queue snapshot for the panel's initial load.
+    pub async fn queue_state(&self) -> QueueState {
+        let w = self.waiting.lock().await;
+        self.build_queue_state(&w)
+    }
+
+    /// Emit `transfer://queue` and bump the generation so admission waiters
+    /// re-check whether it is their turn.
+    async fn bump_queue(&self, app: &AppHandle) {
+        let state = {
+            let w = self.waiting.lock().await;
+            self.build_queue_state(&w)
+        };
+        self.queue_gen.send_modify(|g| *g += 1);
+        let _ = app.emit("transfer://queue", &state);
+    }
+
+    /// Wait until this transfer is at the front of the FIFO with the pause-all
+    /// gate open, take a concurrency permit, and pop it from the queue.
+    /// Returns `None` if the id left the queue (canceled while waiting).
+    /// The permit is returned so the caller holds it for the transfer's life.
+    async fn admit(&self, id: &str) -> Option<OwnedSemaphorePermit> {
+        let mut rx = self.queue_gen.subscribe();
+        loop {
+            {
+                let w = self.waiting.lock().await;
+                if !w.iter().any(|x| x == id) {
+                    return None;
+                }
+                if w.front().map(String::as_str) != Some(id) || self.pause_all.is_paused() {
+                    drop(w);
+                    if rx.changed().await.is_err() {
+                        return None;
+                    }
+                    continue;
+                }
+            }
+            // Front of the queue with the gate open: take a permit, then pop.
+            let permit = self.semaphore.clone().acquire_owned().await.ok()?;
+            let mut w = self.waiting.lock().await;
+            if w.front().map(String::as_str) == Some(id) && !self.pause_all.is_paused() {
+                w.pop_front();
+                return Some(permit);
+            }
+            // Lost a race (pause-all engaged mid-acquire): release, re-evaluate.
+            drop(permit);
+            if !w.iter().any(|x| x == id) {
+                return None;
+            }
+            drop(w);
+        }
+    }
+
+    /// Reorder a waiting transfer (active transfers are untouched).
+    pub async fn move_in_queue(&self, id: &str, up: bool, app: &AppHandle) -> Result<()> {
+        {
+            let mut w = self.waiting.lock().await;
+            let Some(pos) = w.iter().position(|x| x == id) else {
+                anyhow::bail!("transfer {id} is not waiting in the queue");
+            };
+            let swap_with = if up {
+                pos.checked_sub(1)
+            } else if pos + 1 < w.len() {
+                Some(pos + 1)
+            } else {
+                None
+            };
+            if let Some(other) = swap_with {
+                w.swap(pos, other);
+            }
+        }
+        self.bump_queue(app).await;
+        Ok(())
+    }
+
+    /// Pause admission of new transfers (running ones keep going until Phase
+    /// 2's chunk checkpoints let them park too).
+    pub async fn pause_all(&self, app: &AppHandle) {
+        self.pause_all.set(true);
+        self.bump_queue(app).await;
+    }
+
+    pub async fn resume_all(&self, app: &AppHandle) {
+        self.pause_all.set(false);
+        self.bump_queue(app).await;
+    }
+
+    pub fn is_paused_all(&self) -> bool {
+        self.pause_all.is_paused()
+    }
+
+    /// Live-adjust the concurrency bound. Growing adds permits at once;
+    /// shrinking forgets permits as running transfers release them, so
+    /// in-flight transfers are never killed to satisfy the new bound.
+    pub fn set_concurrency(&self, n: usize) {
+        let n = n.clamp(1, 32);
+        let old = self.concurrency.swap(n, Ordering::Relaxed);
+        if n > old {
+            self.semaphore.add_permits(n - old);
+        } else if n < old {
+            let sem = Arc::clone(&self.semaphore);
+            tokio::spawn(async move {
+                if let Ok(p) = sem.acquire_many((old - n) as u32).await {
+                    p.forget();
+                }
+            });
+        }
+    }
+
+    pub async fn cancel(&self, id: &str, app: &AppHandle) -> Result<()> {
+        {
+            let mut w = self.waiting.lock().await;
+            if let Some(pos) = w.iter().position(|x| x == id) {
+                w.remove(pos);
+            }
+        }
         if let Some(h) = self.tasks.lock().await.remove(id) {
             h.abort();
         }
@@ -166,6 +353,11 @@ impl TransferManager {
             }
         })
         .await;
+        // There is no `transfer://canceled` event — `updated` carries the row.
+        if let Some(t) = self.get(id).await {
+            let _ = app.emit("transfer://updated", &t);
+        }
+        self.bump_queue(app).await;
         Ok(())
     }
 
@@ -213,9 +405,15 @@ impl TransferManager {
             return Ok(id);
         }
 
+        self.waiting.lock().await.push_back(id.clone());
+        self.bump_queue(&app).await;
+
         let mgr = Arc::clone(self);
         let id_for_task = id.clone();
         let task = tokio::spawn(async move {
+            let Some(_permit) = mgr.admit(&id_for_task).await else {
+                return;
+            };
             let res = match &*session {
                 Session::Ssh(ssh) => {
                     mgr.run_ssh_download(
@@ -349,6 +547,7 @@ impl TransferManager {
                 }
             };
             finalize(&mgr, &id_for_task, &app, res).await;
+            mgr.bump_queue(&app).await;
         });
         self.tasks.lock().await.insert(id.clone(), task);
         Ok(id)
@@ -498,9 +697,15 @@ impl TransferManager {
             return Ok(id);
         }
 
+        self.waiting.lock().await.push_back(id.clone());
+        self.bump_queue(&app).await;
+
         let mgr = Arc::clone(self);
         let id_for_task = id.clone();
         let task = tokio::spawn(async move {
+            let Some(_permit) = mgr.admit(&id_for_task).await else {
+                return;
+            };
             let res = match &*session {
                 Session::Ssh(ssh) => {
                     mgr.run_ssh_upload(
@@ -627,6 +832,7 @@ impl TransferManager {
                 }
             };
             finalize(&mgr, &id_for_task, &app, res).await;
+            mgr.bump_queue(&app).await;
         });
         self.tasks.lock().await.insert(id.clone(), task);
         Ok(id)

From 255ecec1f9b26270a26c5c8ae4da6cefd4821694 Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sun, 2 Aug 2026 00:07:19 -0400
Subject: [PATCH 2/7] feat(transfers): per-transfer pause/resume with chunk
 checkpoints
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plan 17 Phase 2: a PauseGate per transfer (watch-channel, no missed wakeups)
plus a shared checkpoint() called at every chunk boundary in all 26 backend
copy loops. Pause parks the task at the boundary; resume unwinds the loop
with RestartFromPause and the runner re-runs the file from byte 0 — honest
on every backend, no seek support needed. Admission skips per-transfer-
paused rows so a paused item never head-of-line blocks the queue; pause-all
and per-transfer pauses compose (resume requires both gates open). New
transfer_pause/transfer_resume commands; Paused status; virtualfs hydration
and foldersync's wait_for_transfers treat Paused as in-flight.
---
 src-tauri/src/commands.rs      |  24 +++++
 src-tauri/src/foldersync.rs    |   2 +-
 src-tauri/src/lib.rs           |   2 +
 src-tauri/src/transfer.rs      | 183 +++++++++++++++++++++++++++++++--
 src-tauri/src/virtualfs/mod.rs |   2 +-
 5 files changed, 202 insertions(+), 11 deletions(-)

diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs
index 44d4c1c..a67ae7b 100644
--- a/src-tauri/src/commands.rs
+++ b/src-tauri/src/commands.rs
@@ -747,6 +747,30 @@ pub async fn cancel_transfer(
         .map_err(err)
 }
 
+/// Pause a queued or running transfer (Plan 17 Phase 2). A running one parks
+/// at the next chunk boundary; resume re-runs its file from byte 0.
+#[tauri::command]
+pub async fn transfer_pause(
+    transfer_id: String,
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state.transfers.pause(&transfer_id, &app).await.map_err(err)
+}
+
+#[tauri::command]
+pub async fn transfer_resume(
+    transfer_id: String,
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state
+        .transfers
+        .resume(&transfer_id, &app)
+        .await
+        .map_err(err)
+}
+
 /// Reorder a waiting transfer in the queue FIFO (Plan 17). `direction` is
 /// "up" (sooner) or "down" (later); active transfers are untouched.
 #[tauri::command]
diff --git a/src-tauri/src/foldersync.rs b/src-tauri/src/foldersync.rs
index 204fbc6..4c7bb78 100644
--- a/src-tauri/src/foldersync.rs
+++ b/src-tauri/src/foldersync.rs
@@ -605,7 +605,7 @@ async fn wait_for_transfers(state: &AppState, ids: Vec<String>) {
         for id in &ids {
             if let Some(t) = state.transfers.snapshot(id).await {
                 match t.status {
-                    TransferStatus::Queued | TransferStatus::Transferring => remaining += 1,
+                    TransferStatus::Queued | TransferStatus::Transferring | TransferStatus::Paused => remaining += 1,
                     _ => {}
                 }
             }
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index 8009e9d..d62e7de 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -402,6 +402,8 @@ pub fn run() {
             commands::cancel_transfer,
             commands::list_transfers,
             commands::transfer_move,
+            commands::transfer_pause,
+            commands::transfer_resume,
             commands::transfer_pause_all,
             commands::transfer_resume_all,
             commands::transfer_set_concurrency,
diff --git a/src-tauri/src/transfer.rs b/src-tauri/src/transfer.rs
index 6adf311..31d16e0 100644
--- a/src-tauri/src/transfer.rs
+++ b/src-tauri/src/transfer.rs
@@ -37,12 +37,27 @@ pub enum TransferKind {
 pub enum TransferStatus {
     Queued,
     Transferring,
+    Paused,
     Done,
     Skipped,
     Error,
     Canceled,
 }
 
+/// Marker error: a paused transfer was resumed — the copy loop unwinds with
+/// this and the runner re-runs the file from byte 0 (Plan 17 Phase 2).
+/// Honest on every backend: no per-backend seek support needed.
+#[derive(Debug)]
+struct RestartFromPause;
+
+impl std::fmt::Display for RestartFromPause {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("transfer resumed after pause; restarting file from the beginning")
+    }
+}
+
+impl std::error::Error for RestartFromPause {}
+
 /// Payload of the `transfer://queue` event: the FIFO of waiting transfer ids
 /// plus the manager-level state the panel header renders (Plan 17).
 #[derive(Debug, Clone, Serialize)]
@@ -73,7 +88,6 @@ impl PauseGate {
         let _ = self.tx.send(paused);
     }
     /// Park until the gate opens. Returns immediately if already open.
-    #[allow(dead_code)] // used by the Phase 2 chunk checkpoints
     async fn wait_open(&self) {
         let mut rx = self.tx.subscribe();
         while *rx.borrow_and_update() {
@@ -114,6 +128,8 @@ pub struct TransferManager {
     concurrency: AtomicUsize,
     /// Manager-level pause gate. Admission checks it; Phase 2 checkpoints do too.
     pause_all: PauseGate,
+    /// Per-transfer pause gates, created at enqueue time (Plan 17 Phase 2).
+    pauses: Mutex<HashMap<String, PauseGate>>,
     /// Bumped on every queue change so admission waiters re-check their turn.
     queue_gen: watch::Sender<u64>,
 }
@@ -187,6 +203,7 @@ impl TransferManager {
             semaphore: Arc::new(Semaphore::new(DEFAULT_CONCURRENCY)),
             concurrency: AtomicUsize::new(DEFAULT_CONCURRENCY),
             pause_all: PauseGate::new(),
+            pauses: Mutex::new(HashMap::new()),
             queue_gen: watch::channel(0).0,
         }
     }
@@ -257,7 +274,7 @@ impl TransferManager {
                 if !w.iter().any(|x| x == id) {
                     return None;
                 }
-                if w.front().map(String::as_str) != Some(id) || self.pause_all.is_paused() {
+                if !self.is_my_turn(&w, id).await {
                     drop(w);
                     if rx.changed().await.is_err() {
                         return None;
@@ -265,14 +282,16 @@ impl TransferManager {
                     continue;
                 }
             }
-            // Front of the queue with the gate open: take a permit, then pop.
+            // My turn: take a permit, then pop.
             let permit = self.semaphore.clone().acquire_owned().await.ok()?;
             let mut w = self.waiting.lock().await;
-            if w.front().map(String::as_str) == Some(id) && !self.pause_all.is_paused() {
-                w.pop_front();
+            if self.is_my_turn(&w, id).await {
+                if let Some(pos) = w.iter().position(|x| x == id) {
+                    w.remove(pos);
+                }
                 return Some(permit);
             }
-            // Lost a race (pause-all engaged mid-acquire): release, re-evaluate.
+            // Lost a race (pause engaged mid-acquire): release, re-evaluate.
             drop(permit);
             if !w.iter().any(|x| x == id) {
                 return None;
@@ -281,6 +300,21 @@ impl TransferManager {
         }
     }
 
+    /// Is `id` the first waiting transfer allowed to run? Strict FIFO except
+    /// that per-transfer-paused rows are skipped (a paused row must not
+    /// head-of-line block the queue); pause-all blocks everyone. Caller must
+    /// hold the `waiting` lock; lock order is waiting → pauses.
+    async fn is_my_turn(&self, w: &VecDeque<String>, id: &str) -> bool {
+        if self.pause_all.is_paused() {
+            return false;
+        }
+        let pauses = self.pauses.lock().await;
+        let first_open = w
+            .iter()
+            .position(|x| pauses.get(x).map_or(true, |g| !g.is_paused()));
+        first_open == w.iter().position(|x| x == id)
+    }
+
     /// Reorder a waiting transfer (active transfers are untouched).
     pub async fn move_in_queue(&self, id: &str, up: bool, app: &AppHandle) -> Result<()> {
         {
@@ -319,6 +353,83 @@ impl TransferManager {
         self.pause_all.is_paused()
     }
 
+    /// Chunk-boundary checkpoint shared by every copy loop (Plan 17). Phase 4
+    /// draws `_bytes` from the bandwidth bucket here. When the transfer (or
+    /// the whole manager) is paused, this parks until resumed, then returns
+    /// `RestartFromPause` so the runner re-runs the file from byte 0.
+    async fn checkpoint(&self, id: &str, _bytes: u64) -> Result<()> {
+        let gate = self.pauses.lock().await.get(id).cloned();
+        let parked = self.pause_all.is_paused() || gate.as_ref().is_some_and(|g| g.is_paused());
+        if !parked {
+            return Ok(());
+        }
+        // Park until BOTH gates are open (resume requires both).
+        loop {
+            self.pause_all.wait_open().await;
+            if let Some(g) = &gate {
+                g.wait_open().await;
+            }
+            if !self.pause_all.is_paused() && gate.as_ref().is_none_or(|g| !g.is_paused()) {
+                break;
+            }
+        }
+        Err(RestartFromPause.into())
+    }
+
+    /// Pause a queued or transferring transfer. A running one parks at the
+    /// next chunk boundary; a queued one is skipped by admission until resumed.
+    pub async fn pause(&self, id: &str, app: &AppHandle) -> Result<()> {
+        match self.get(id).await.map(|t| t.status) {
+            Some(TransferStatus::Transferring) | Some(TransferStatus::Queued) => {}
+            _ => anyhow::bail!("transfer {id} is not running or queued"),
+        }
+        let gate = self
+            .pauses
+            .lock()
+            .await
+            .get(id)
+            .cloned()
+            .ok_or_else(|| anyhow::anyhow!("transfer {id} not found"))?;
+        gate.set(true);
+        self.update(id, |t| t.status = TransferStatus::Paused).await;
+        if let Some(t) = self.get(id).await {
+            let _ = app.emit("transfer://updated", &t);
+        }
+        Ok(())
+    }
+
+    /// Resume a paused transfer. A parked one re-runs its file from byte 0;
+    /// a queued one re-enters FIFO admission.
+    pub async fn resume(&self, id: &str, app: &AppHandle) -> Result<()> {
+        if self.get(id).await.map(|t| t.status) != Some(TransferStatus::Paused) {
+            anyhow::bail!("transfer {id} is not paused");
+        }
+        let still_queued = self.waiting.lock().await.iter().any(|x| x == id);
+        let gate = self
+            .pauses
+            .lock()
+            .await
+            .get(id)
+            .cloned()
+            .ok_or_else(|| anyhow::anyhow!("transfer {id} not found"))?;
+        self.update(id, |t| {
+            t.status = if still_queued {
+                TransferStatus::Queued
+            } else {
+                TransferStatus::Transferring
+            };
+            t.transferred = 0;
+        })
+        .await;
+        gate.set(false);
+        if let Some(t) = self.get(id).await {
+            let _ = app.emit("transfer://updated", &t);
+        }
+        // Wake admission waiters: a queued row may have become runnable.
+        self.bump_queue(app).await;
+        Ok(())
+    }
+
     /// Live-adjust the concurrency bound. Growing adds permits at once;
     /// shrinking forgets permits as running transfers release them, so
     /// in-flight transfers are never killed to satisfy the new bound.
@@ -348,7 +459,10 @@ impl TransferManager {
             h.abort();
         }
         self.update(id, |t| {
-            if t.status == TransferStatus::Transferring || t.status == TransferStatus::Queued {
+            if matches!(
+                t.status,
+                TransferStatus::Transferring | TransferStatus::Queued | TransferStatus::Paused
+            ) {
                 t.status = TransferStatus::Canceled;
             }
         })
@@ -406,6 +520,7 @@ impl TransferManager {
         }
 
         self.waiting.lock().await.push_back(id.clone());
+        self.pauses.lock().await.insert(id.clone(), PauseGate::new());
         self.bump_queue(&app).await;
 
         let mgr = Arc::clone(self);
@@ -414,7 +529,10 @@ impl TransferManager {
             let Some(_permit) = mgr.admit(&id_for_task).await else {
                 return;
             };
-            let res = match &*session {
+            // A resume-after-pause unwinds the copy loop with RestartFromPause;
+            // re-run the file from byte 0 (Plan 17 Phase 2).
+            let res = loop {
+            let attempt = match &*session {
                 Session::Ssh(ssh) => {
                     mgr.run_ssh_download(
                         &id_for_task,
@@ -546,6 +664,14 @@ impl TransferManager {
                     .await
                 }
             };
+            match attempt {
+                Err(e) if e.downcast_ref::<RestartFromPause>().is_some() => {
+                    mgr.update(&id_for_task, |t| t.transferred = 0).await;
+                    continue;
+                }
+                other => break other,
+            }
+            };
             finalize(&mgr, &id_for_task, &app, res).await;
             mgr.bump_queue(&app).await;
         });
@@ -586,6 +712,7 @@ impl TransferManager {
             let bytes = base64::engine::general_purpose::STANDARD
                 .decode(&data_b64)
                 .context("decode chunk")?;
+            self.checkpoint(id, bytes.len() as u64).await?;
             if !bytes.is_empty() {
                 local_file.write_all(&bytes).await?;
                 offset += bytes.len() as u64;
@@ -636,6 +763,7 @@ impl TransferManager {
             if n == 0 {
                 break;
             }
+            self.checkpoint(id, n as u64).await?;
             local_file.write_all(&buf[..n]).await?;
             transferred += n as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -698,6 +826,7 @@ impl TransferManager {
         }
 
         self.waiting.lock().await.push_back(id.clone());
+        self.pauses.lock().await.insert(id.clone(), PauseGate::new());
         self.bump_queue(&app).await;
 
         let mgr = Arc::clone(self);
@@ -706,7 +835,10 @@ impl TransferManager {
             let Some(_permit) = mgr.admit(&id_for_task).await else {
                 return;
             };
-            let res = match &*session {
+            // A resume-after-pause unwinds the copy loop with RestartFromPause;
+            // re-run the file from byte 0 (Plan 17 Phase 2).
+            let res = loop {
+            let attempt = match &*session {
                 Session::Ssh(ssh) => {
                     mgr.run_ssh_upload(
                         &id_for_task,
@@ -831,6 +963,14 @@ impl TransferManager {
                     .await
                 }
             };
+            match attempt {
+                Err(e) if e.downcast_ref::<RestartFromPause>().is_some() => {
+                    mgr.update(&id_for_task, |t| t.transferred = 0).await;
+                    continue;
+                }
+                other => break other,
+            }
+            };
             finalize(&mgr, &id_for_task, &app, res).await;
             mgr.bump_queue(&app).await;
         });
@@ -866,6 +1006,7 @@ impl TransferManager {
             if n == 0 {
                 break;
             }
+            self.checkpoint(id, n as u64).await?;
             let data = base64::engine::general_purpose::STANDARD.encode(&buf[..n]);
             let resp = session
                 .request(Request::WriteChunk {
@@ -927,6 +1068,7 @@ impl TransferManager {
             if n == 0 {
                 break;
             }
+            self.checkpoint(id, n as u64).await?;
             remote_file.write_all(&buf[..n]).await?;
             transferred += n as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1111,6 +1253,7 @@ impl TransferManager {
 
         let _ = (mgr_for_emit, id_for_emit, app_for_emit);
 
+        self.checkpoint(id, 0).await?;
         let res: Result<u64> = session
             .with_stream(move |stream| {
                 let file = std::fs::File::create(&final_path)
@@ -1138,6 +1281,7 @@ impl TransferManager {
             .await;
         let _ = app; // progress events come at completion for FTP
 
+        self.checkpoint(id, 0).await?;
         let local = local_path.to_path_buf();
         let remote = remote_path.to_string();
         let res: Result<u64> = session
@@ -1184,6 +1328,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("s3 chunk for {key}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1227,6 +1372,7 @@ impl TransferManager {
             .with_context(|| format!("open {}", local_path.display()))?;
 
         if size <= 16 * 1024 * 1024 {
+            self.checkpoint(id, size).await?;
             let mut buf = Vec::with_capacity(size as usize);
             file.read_to_end(&mut buf).await?;
             session
@@ -1265,6 +1411,7 @@ impl TransferManager {
             if filled == 0 {
                 break;
             }
+            self.checkpoint(id, filled as u64).await?;
             let chunk = bytes::Bytes::copy_from_slice(&buf[..filled]);
             upload
                 .put_part(chunk.into())
@@ -1326,6 +1473,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("webdav chunk for {remote_path}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1365,6 +1513,7 @@ impl TransferManager {
             .await
             .with_context(|| format!("open {}", local_path.display()))?;
         let body = reqwest::Body::wrap_stream(ReaderStream::new(file));
+        self.checkpoint(id, size).await?;
 
         let url = session.url_for(remote_path, false);
         let resp = session
@@ -1419,6 +1568,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("http chunk for {remote_path}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1461,6 +1611,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("dropbox chunk for {remote_path}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1515,6 +1666,7 @@ impl TransferManager {
 
         // Proactive refresh covers the common case; on a hard 401 we refresh and
         // retry once, re-opening the file for a fresh streamed body.
+        self.checkpoint(id, size).await?;
         let mut attempt = 0;
         loop {
             let token = session.access_token().await?;
@@ -1563,6 +1715,7 @@ impl TransferManager {
             .await;
         let _ = app; // single-shot API: progress is reported at completion.
 
+        self.checkpoint(id, 0).await?;
         let data = crate::remotefs::shopify::read_asset(&session, remote_path).await?;
         let mut file = tokio::fs::File::create(local_path)
             .await
@@ -1591,6 +1744,7 @@ impl TransferManager {
             .await
             .with_context(|| format!("read {}", local_path.display()))?;
         let size = data.len() as u64;
+        self.checkpoint(id, size).await?;
         crate::remotefs::shopify::write_asset(&session, remote_path, &data).await?;
         self.update(id, |t| t.transferred = size).await;
         Ok(())
@@ -1611,6 +1765,7 @@ impl TransferManager {
             .await;
         let _ = app; // single-shot API: progress is reported at completion.
 
+        self.checkpoint(id, 0).await?;
         let data = crate::remotefs::hubspot::read_file(&session, remote_path).await?;
         let mut file = tokio::fs::File::create(local_path)
             .await
@@ -1640,6 +1795,7 @@ impl TransferManager {
             .await
             .with_context(|| format!("read {}", local_path.display()))?;
         let size = data.len() as u64;
+        self.checkpoint(id, size).await?;
         crate::remotefs::hubspot::write_file(&session, remote_path, &data).await?;
         self.update(id, |t| t.transferred = size).await;
         Ok(())
@@ -1660,6 +1816,7 @@ impl TransferManager {
             .await;
         let _ = app; // single-shot API: progress is reported at completion.
 
+        self.checkpoint(id, 0).await?;
         let data = crate::remotefs::dynamics::read_file(&session, remote_path).await?;
         let mut file = tokio::fs::File::create(local_path)
             .await
@@ -1688,6 +1845,7 @@ impl TransferManager {
             .await
             .with_context(|| format!("read {}", local_path.display()))?;
         let size = data.len() as u64;
+        self.checkpoint(id, size).await?;
         crate::remotefs::dynamics::write_file(&session, remote_path, &data).await?;
         self.update(id, |t| t.transferred = size).await;
         Ok(())
@@ -1719,6 +1877,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("onedrive chunk for {remote_path}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1758,6 +1917,7 @@ impl TransferManager {
             .unwrap_or(4 * 1024 * 1024);
 
         if size <= simple_max {
+            self.checkpoint(id, size).await?;
             self.onedrive_simple_upload(id, &session, local_path, remote_path)
                 .await?;
         } else {
@@ -1852,6 +2012,7 @@ impl TransferManager {
             if filled == 0 {
                 break;
             }
+            self.checkpoint(id, filled as u64).await?;
             let start = offset;
             let end = offset + filled as u64 - 1;
             let range = format!("bytes {start}-{end}/{size}");
@@ -1912,6 +2073,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("drive chunk for {remote_path}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -1954,6 +2116,7 @@ impl TransferManager {
             .await
             .with_context(|| format!("read {}", local_path.display()))?;
         let token = session.access_token().await?;
+        self.checkpoint(id, size).await?;
 
         let resp = if let Some((file_id, _)) = existing {
             // Update the existing file's content in place.
@@ -2038,6 +2201,7 @@ impl TransferManager {
         let mut last_emit = Instant::now();
         while let Some(chunk) = stream.next().await {
             let chunk = chunk.with_context(|| format!("box chunk for {remote_path}"))?;
+            self.checkpoint(id, chunk.len() as u64).await?;
             file.write_all(&chunk).await?;
             transferred += chunk.len() as u64;
             if last_emit.elapsed() > Duration::from_millis(100) {
@@ -2080,6 +2244,7 @@ impl TransferManager {
             .await
             .with_context(|| format!("read {}", local_path.display()))?;
         let token = session.access_token().await?;
+        self.checkpoint(id, size).await?;
 
         let file_part = reqwest::multipart::Part::bytes(bytes).file_name(name.clone());
         let (url, form) = match existing {
diff --git a/src-tauri/src/virtualfs/mod.rs b/src-tauri/src/virtualfs/mod.rs
index b423940..0cc8b8e 100644
--- a/src-tauri/src/virtualfs/mod.rs
+++ b/src-tauri/src/virtualfs/mod.rs
@@ -357,7 +357,7 @@ impl Hydrator for SessionHydrator {
                         anyhow::bail!("hydration download failed: {}", t.error.unwrap_or_default())
                     }
                     TransferStatus::Canceled => anyhow::bail!("hydration download canceled"),
-                    TransferStatus::Queued | TransferStatus::Transferring => {
+                    TransferStatus::Queued | TransferStatus::Transferring | TransferStatus::Paused => {
                         tokio::time::sleep(std::time::Duration::from_millis(120)).await;
                     }
                 },

From 44cbdbb05110bae3c6341201d421bdd2e0586dd8 Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sun, 2 Aug 2026 00:14:35 -0400
Subject: [PATCH 3/7] feat(transfers): manual + auto retry with backoff
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plan 17 Phase 3: errors are classified with Plan 12's structured kinds
(classify_message, now pub(crate)); Network/Timeout failures auto-retry up
to 2 times with 5s/20s backoff — the row shows "retrying in Ns (attempt
N/3)" and the concurrency slot is held through backoff so a retrying
transfer keeps its place. Auth/Permission/NotFound never auto-retry. A
manual transfer_retry re-enqueues Error/Canceled rows with their original
policy-resolved source/destination (stored as RetryInfo at enqueue), same
id, reopening a pause gate left closed by cancel-while-paused. The spawn
closures are extracted into shared run_download_task/run_upload_task +
dispatch_download/dispatch_upload so enqueue and retry share one runner.
---
 src-tauri/src/commands.rs |  11 +
 src-tauri/src/error.rs    |   2 +-
 src-tauri/src/lib.rs      |   1 +
 src-tauri/src/transfer.rs | 651 +++++++++++++++++++++-----------------
 4 files changed, 371 insertions(+), 294 deletions(-)

diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs
index a67ae7b..da64a7a 100644
--- a/src-tauri/src/commands.rs
+++ b/src-tauri/src/commands.rs
@@ -771,6 +771,17 @@ pub async fn transfer_resume(
         .map_err(err)
 }
 
+/// Re-enqueue a failed or canceled transfer with its original source/
+/// destination (Plan 17 Phase 3). Same id — the panel row resets in place.
+#[tauri::command]
+pub async fn transfer_retry(
+    transfer_id: String,
+    app: AppHandle,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state.transfers.retry(&transfer_id, &app).await.map_err(err)
+}
+
 /// Reorder a waiting transfer in the queue FIFO (Plan 17). `direction` is
 /// "up" (sooner) or "down" (later); active transfers are untouched.
 #[tauri::command]
diff --git a/src-tauri/src/error.rs b/src-tauri/src/error.rs
index 46f5f88..8d8a7c7 100644
--- a/src-tauri/src/error.rs
+++ b/src-tauri/src/error.rs
@@ -117,7 +117,7 @@ fn classify_io(kind: std::io::ErrorKind) -> Option<ErrorKind> {
 
 /// Keyword heuristics over an error message. Backend-side only — the frontend
 /// gets the resulting `kind` and never repeats this matching.
-fn classify_message(message: &str) -> ErrorKind {
+pub(crate) fn classify_message(message: &str) -> ErrorKind {
     let s = message.to_ascii_lowercase();
     let has = |needles: &[&str]| needles.iter().any(|n| s.contains(n));
 
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index d62e7de..6836e05 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -404,6 +404,7 @@ pub fn run() {
             commands::transfer_move,
             commands::transfer_pause,
             commands::transfer_resume,
+            commands::transfer_retry,
             commands::transfer_pause_all,
             commands::transfer_resume_all,
             commands::transfer_set_concurrency,
diff --git a/src-tauri/src/transfer.rs b/src-tauri/src/transfer.rs
index 31d16e0..950385b 100644
--- a/src-tauri/src/transfer.rs
+++ b/src-tauri/src/transfer.rs
@@ -110,13 +110,38 @@ pub struct Transfer {
     pub status: TransferStatus,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub error: Option<String>,
+    /// Auto-retry round in progress (1- or 2-of-2), for the panel's
+    /// "retrying in Ns (attempt N/3)" state (Plan 17 Phase 3).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub retry_attempt: Option<u32>,
     pub started_at: i64,
 }
 
+/// Everything needed to re-run a failed/canceled transfer with its already
+/// policy-resolved destination (overwrite/skip/rename was applied at enqueue
+/// time) — Plan 17 Phase 3 manual retry.
+#[derive(Clone)]
+enum RetryInfo {
+    Download {
+        session: Arc<Session>,
+        remote_path: String,
+        final_path: PathBuf,
+    },
+    Upload {
+        session: Arc<Session>,
+        local: PathBuf,
+        final_remote: String,
+    },
+}
+
 /// Default bound on concurrently running transfers (Plan 17); the rest wait
 /// in the FIFO as `Queued`. Overridden by the `transferConcurrency` setting.
 const DEFAULT_CONCURRENCY: usize = 3;
 
+/// How many times a transfer auto-retries on transient (Network/Timeout)
+/// errors before surfacing the failure (Plan 17 Phase 3).
+const MAX_AUTO_RETRIES: u32 = 2;
+
 pub struct TransferManager {
     transfers: Mutex<HashMap<String, Transfer>>,
     tasks: Mutex<HashMap<String, JoinHandle<()>>>,
@@ -130,6 +155,8 @@ pub struct TransferManager {
     pause_all: PauseGate,
     /// Per-transfer pause gates, created at enqueue time (Plan 17 Phase 2).
     pauses: Mutex<HashMap<String, PauseGate>>,
+    /// Original resolved inputs per transfer, for manual retry (Phase 3).
+    retry: Mutex<HashMap<String, RetryInfo>>,
     /// Bumped on every queue change so admission waiters re-check their turn.
     queue_gen: watch::Sender<u64>,
 }
@@ -204,6 +231,7 @@ impl TransferManager {
             concurrency: AtomicUsize::new(DEFAULT_CONCURRENCY),
             pause_all: PauseGate::new(),
             pauses: Mutex::new(HashMap::new()),
+            retry: Mutex::new(HashMap::new()),
             queue_gen: watch::channel(0).0,
         }
     }
@@ -430,6 +458,73 @@ impl TransferManager {
         Ok(())
     }
 
+    /// Re-enqueue a failed or canceled transfer with its original (already
+    /// policy-resolved) source/destination. Same id — the panel row resets
+    /// in place (Plan 17 Phase 3 manual retry).
+    pub async fn retry(self: &Arc<Self>, id: &str, app: &AppHandle) -> Result<()> {
+        match self.get(id).await.map(|t| t.status) {
+            Some(TransferStatus::Error) | Some(TransferStatus::Canceled) => {}
+            _ => anyhow::bail!("only failed or canceled transfers can be retried"),
+        }
+        let info = self
+            .retry
+            .lock()
+            .await
+            .get(id)
+            .cloned()
+            .ok_or_else(|| anyhow::anyhow!("transfer {id} cannot be retried"))?;
+        if let Some(h) = self.tasks.lock().await.remove(id) {
+            h.abort();
+        }
+        // A cancel-while-paused leaves the gate closed — reopen it.
+        if let Some(g) = self.pauses.lock().await.get(id) {
+            g.set(false);
+        }
+        self.update(id, |t| {
+            t.status = TransferStatus::Queued;
+            t.transferred = 0;
+            t.error = None;
+            t.retry_attempt = None;
+        })
+        .await;
+        if let Some(t) = self.get(id).await {
+            let _ = app.emit("transfer://updated", &t);
+        }
+        self.waiting.lock().await.push_back(id.to_string());
+        self.bump_queue(app).await;
+        let mgr = Arc::clone(self);
+        let id_for_task = id.to_string();
+        let app_for_task = app.clone();
+        let task = match info {
+            RetryInfo::Download {
+                session,
+                remote_path,
+                final_path,
+            } => tokio::spawn(run_download_task(
+                mgr,
+                id_for_task,
+                session,
+                remote_path,
+                final_path,
+                app_for_task,
+            )),
+            RetryInfo::Upload {
+                session,
+                local,
+                final_remote,
+            } => tokio::spawn(run_upload_task(
+                mgr,
+                id_for_task,
+                session,
+                local,
+                final_remote,
+                app_for_task,
+            )),
+        };
+        self.tasks.lock().await.insert(id.to_string(), task);
+        Ok(())
+    }
+
     /// Live-adjust the concurrency bound. Growing adds permits at once;
     /// shrinking forgets permits as running transfers release them, so
     /// in-flight transfers are never killed to satisfy the new bound.
@@ -509,6 +604,7 @@ impl TransferManager {
                 TransferStatus::Queued
             },
             error: None,
+            retry_attempt: None,
             started_at: now_ts(),
         };
         self.insert(transfer.clone()).await;
@@ -519,162 +615,28 @@ impl TransferManager {
             return Ok(id);
         }
 
+        self.retry.lock().await.insert(
+            id.clone(),
+            RetryInfo::Download {
+                session: Arc::clone(&session),
+                remote_path: remote_path.clone(),
+                final_path: final_path.clone(),
+            },
+        );
         self.waiting.lock().await.push_back(id.clone());
         self.pauses.lock().await.insert(id.clone(), PauseGate::new());
         self.bump_queue(&app).await;
 
         let mgr = Arc::clone(self);
         let id_for_task = id.clone();
-        let task = tokio::spawn(async move {
-            let Some(_permit) = mgr.admit(&id_for_task).await else {
-                return;
-            };
-            // A resume-after-pause unwinds the copy loop with RestartFromPause;
-            // re-run the file from byte 0 (Plan 17 Phase 2).
-            let res = loop {
-            let attempt = match &*session {
-                Session::Ssh(ssh) => {
-                    mgr.run_ssh_download(
-                        &id_for_task,
-                        ssh.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Ftp(ftp) => {
-                    mgr.run_ftp_download(
-                        &id_for_task,
-                        ftp.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Object(obj) => {
-                    mgr.run_object_download(
-                        &id_for_task,
-                        obj.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Webdav(dav) => {
-                    mgr.run_webdav_download(
-                        &id_for_task,
-                        dav.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Http(http) => {
-                    mgr.run_http_download(
-                        &id_for_task,
-                        http.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Dropbox(dbx) => {
-                    mgr.run_dropbox_download(
-                        &id_for_task,
-                        dbx.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::OneDrive(od) => {
-                    mgr.run_onedrive_download(
-                        &id_for_task,
-                        od.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::GDrive(gd) => {
-                    mgr.run_gdrive_download(
-                        &id_for_task,
-                        gd.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Box(bx) => {
-                    mgr.run_box_download(
-                        &id_for_task,
-                        bx.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Shopify(sh) => {
-                    mgr.run_shopify_download(
-                        &id_for_task,
-                        sh.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::HubSpot(hs) => {
-                    mgr.run_hubspot_download(
-                        &id_for_task,
-                        hs.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Dynamics(dynm) => {
-                    mgr.run_dynamics_download(
-                        &id_for_task,
-                        dynm.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Agent(agent) => {
-                    mgr.run_agent_download(
-                        &id_for_task,
-                        agent.clone(),
-                        &remote_path,
-                        &final_path,
-                        &app,
-                    )
-                    .await
-                }
-            };
-            match attempt {
-                Err(e) if e.downcast_ref::<RestartFromPause>().is_some() => {
-                    mgr.update(&id_for_task, |t| t.transferred = 0).await;
-                    continue;
-                }
-                other => break other,
-            }
-            };
-            finalize(&mgr, &id_for_task, &app, res).await;
-            mgr.bump_queue(&app).await;
-        });
+        let task = tokio::spawn(run_download_task(
+            mgr,
+            id_for_task,
+            session,
+            remote_path,
+            final_path,
+            app,
+        ));
         self.tasks.lock().await.insert(id.clone(), task);
         Ok(id)
     }
@@ -815,6 +777,7 @@ impl TransferManager {
                 TransferStatus::Queued
             },
             error: None,
+            retry_attempt: None,
             started_at: now_ts(),
         };
         self.insert(transfer.clone()).await;
@@ -825,155 +788,28 @@ impl TransferManager {
             return Ok(id);
         }
 
+        self.retry.lock().await.insert(
+            id.clone(),
+            RetryInfo::Upload {
+                session: Arc::clone(&session),
+                local: local.clone(),
+                final_remote: final_remote.clone(),
+            },
+        );
         self.waiting.lock().await.push_back(id.clone());
         self.pauses.lock().await.insert(id.clone(), PauseGate::new());
         self.bump_queue(&app).await;
 
         let mgr = Arc::clone(self);
         let id_for_task = id.clone();
-        let task = tokio::spawn(async move {
-            let Some(_permit) = mgr.admit(&id_for_task).await else {
-                return;
-            };
-            // A resume-after-pause unwinds the copy loop with RestartFromPause;
-            // re-run the file from byte 0 (Plan 17 Phase 2).
-            let res = loop {
-            let attempt = match &*session {
-                Session::Ssh(ssh) => {
-                    mgr.run_ssh_upload(
-                        &id_for_task,
-                        ssh.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Ftp(ftp) => {
-                    mgr.run_ftp_upload(
-                        &id_for_task,
-                        ftp.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Object(obj) => {
-                    mgr.run_object_upload(
-                        &id_for_task,
-                        obj.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Webdav(dav) => {
-                    mgr.run_webdav_upload(
-                        &id_for_task,
-                        dav.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Http(_) => {
-                    Err(anyhow::anyhow!("HTTP source is read-only — upload not supported"))
-                }
-                Session::Dropbox(dbx) => {
-                    mgr.run_dropbox_upload(
-                        &id_for_task,
-                        dbx.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::OneDrive(od) => {
-                    mgr.run_onedrive_upload(
-                        &id_for_task,
-                        od.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::GDrive(gd) => {
-                    mgr.run_gdrive_upload(
-                        &id_for_task,
-                        gd.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Box(bx) => {
-                    mgr.run_box_upload(
-                        &id_for_task,
-                        bx.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Shopify(sh) => {
-                    mgr.run_shopify_upload(
-                        &id_for_task,
-                        sh.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::HubSpot(hs) => {
-                    mgr.run_hubspot_upload(
-                        &id_for_task,
-                        hs.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Dynamics(dynm) => {
-                    mgr.run_dynamics_upload(
-                        &id_for_task,
-                        dynm.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-                Session::Agent(agent) => {
-                    mgr.run_agent_upload(
-                        &id_for_task,
-                        agent.clone(),
-                        &local,
-                        &final_remote,
-                        &app,
-                    )
-                    .await
-                }
-            };
-            match attempt {
-                Err(e) if e.downcast_ref::<RestartFromPause>().is_some() => {
-                    mgr.update(&id_for_task, |t| t.transferred = 0).await;
-                    continue;
-                }
-                other => break other,
-            }
-            };
-            finalize(&mgr, &id_for_task, &app, res).await;
-            mgr.bump_queue(&app).await;
-        });
+        let task = tokio::spawn(run_upload_task(
+            mgr,
+            id_for_task,
+            session,
+            local,
+            final_remote,
+            app,
+        ));
         self.tasks.lock().await.insert(id.clone(), task);
         Ok(id)
     }
@@ -2649,6 +2485,235 @@ fn split_ext(path: &str) -> (&str, &str) {
     }
 }
 
+/// Transient = worth an auto-retry (Plan 12's structured error kinds):
+/// network and timeout failures; auth/permission/not-found never retry.
+fn is_transient(e: &anyhow::Error) -> bool {
+    matches!(
+        crate::error::classify_message(&format!("{e:#}")),
+        crate::error::ErrorKind::Network | crate::error::ErrorKind::Timeout
+    )
+}
+
+/// Shared download runner (Plan 17): admission → run loop → finalize → wake
+/// the queue. The loop re-runs the file from byte 0 after a resume-from-pause
+/// (Phase 2) and auto-retries transient errors with 5s/20s backoff (Phase 3).
+/// The concurrency permit is held through backoff — a deliberate trade-off so
+/// a retrying transfer keeps its slot.
+async fn run_download_task(
+    mgr: Arc<TransferManager>,
+    id: String,
+    session: Arc<Session>,
+    remote_path: String,
+    final_path: PathBuf,
+    app: AppHandle,
+) {
+    let Some(_permit) = mgr.admit(&id).await else {
+        return;
+    };
+    let mut auto_retries = 0u32;
+    let res = loop {
+        let attempt = dispatch_download(&mgr, &id, &session, &remote_path, &final_path, &app).await;
+        match attempt {
+            Err(e) if e.downcast_ref::<RestartFromPause>().is_some() => {
+                mgr.update(&id, |t| t.transferred = 0).await;
+                continue;
+            }
+            Err(e) if auto_retries < MAX_AUTO_RETRIES && is_transient(&e) => {
+                auto_retries += 1;
+                let delay = if auto_retries == 1 { 5 } else { 20 };
+                mgr.update(&id, |t| {
+                    t.transferred = 0;
+                    t.retry_attempt = Some(auto_retries);
+                    t.error = Some(format!("retrying in {delay}s (attempt {}/3)", auto_retries + 1));
+                })
+                .await;
+                if let Some(t) = mgr.get(&id).await {
+                    let _ = app.emit("transfer://updated", &t);
+                }
+                tokio::time::sleep(Duration::from_secs(delay)).await;
+                mgr.update(&id, |t| t.error = None).await;
+                continue;
+            }
+            other => break other,
+        }
+    };
+    finalize(&mgr, &id, &app, res).await;
+    mgr.bump_queue(&app).await;
+}
+
+/// Upload twin of `run_download_task`.
+async fn run_upload_task(
+    mgr: Arc<TransferManager>,
+    id: String,
+    session: Arc<Session>,
+    local: PathBuf,
+    final_remote: String,
+    app: AppHandle,
+) {
+    let Some(_permit) = mgr.admit(&id).await else {
+        return;
+    };
+    let mut auto_retries = 0u32;
+    let res = loop {
+        let attempt = dispatch_upload(&mgr, &id, &session, &local, &final_remote, &app).await;
+        match attempt {
+            Err(e) if e.downcast_ref::<RestartFromPause>().is_some() => {
+                mgr.update(&id, |t| t.transferred = 0).await;
+                continue;
+            }
+            Err(e) if auto_retries < MAX_AUTO_RETRIES && is_transient(&e) => {
+                auto_retries += 1;
+                let delay = if auto_retries == 1 { 5 } else { 20 };
+                mgr.update(&id, |t| {
+                    t.transferred = 0;
+                    t.retry_attempt = Some(auto_retries);
+                    t.error = Some(format!("retrying in {delay}s (attempt {}/3)", auto_retries + 1));
+                })
+                .await;
+                if let Some(t) = mgr.get(&id).await {
+                    let _ = app.emit("transfer://updated", &t);
+                }
+                tokio::time::sleep(Duration::from_secs(delay)).await;
+                mgr.update(&id, |t| t.error = None).await;
+                continue;
+            }
+            other => break other,
+        }
+    };
+    finalize(&mgr, &id, &app, res).await;
+    mgr.bump_queue(&app).await;
+}
+
+/// Backend dispatch for a single-file download. Extracted so the runner (and
+/// Phase 3 retry) can re-invoke it.
+async fn dispatch_download(
+    mgr: &Arc<TransferManager>,
+    id: &str,
+    session: &Arc<Session>,
+    remote_path: &str,
+    final_path: &Path,
+    app: &AppHandle,
+) -> Result<()> {
+    match &**session {
+        Session::Ssh(ssh) => {
+            mgr.run_ssh_download(id, ssh.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Ftp(ftp) => {
+            mgr.run_ftp_download(id, ftp.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Object(obj) => {
+            mgr.run_object_download(id, obj.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Webdav(dav) => {
+            mgr.run_webdav_download(id, dav.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Http(http) => {
+            mgr.run_http_download(id, http.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Dropbox(dbx) => {
+            mgr.run_dropbox_download(id, dbx.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::OneDrive(od) => {
+            mgr.run_onedrive_download(id, od.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::GDrive(gd) => {
+            mgr.run_gdrive_download(id, gd.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Box(bx) => {
+            mgr.run_box_download(id, bx.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Shopify(sh) => {
+            mgr.run_shopify_download(id, sh.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::HubSpot(hs) => {
+            mgr.run_hubspot_download(id, hs.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Dynamics(dynm) => {
+            mgr.run_dynamics_download(id, dynm.clone(), remote_path, final_path, app)
+                .await
+        }
+        Session::Agent(agent) => {
+            mgr.run_agent_download(id, agent.clone(), remote_path, final_path, app)
+                .await
+        }
+    }
+}
+
+/// Backend dispatch for a single-file upload.
+async fn dispatch_upload(
+    mgr: &Arc<TransferManager>,
+    id: &str,
+    session: &Arc<Session>,
+    local: &Path,
+    final_remote: &str,
+    app: &AppHandle,
+) -> Result<()> {
+    match &**session {
+        Session::Ssh(ssh) => {
+            mgr.run_ssh_upload(id, ssh.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Ftp(ftp) => {
+            mgr.run_ftp_upload(id, ftp.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Object(obj) => {
+            mgr.run_object_upload(id, obj.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Webdav(dav) => {
+            mgr.run_webdav_upload(id, dav.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Http(_) => Err(anyhow::anyhow!(
+            "HTTP source is read-only — upload not supported"
+        )),
+        Session::Dropbox(dbx) => {
+            mgr.run_dropbox_upload(id, dbx.clone(), local, final_remote, app)
+                .await
+        }
+        Session::OneDrive(od) => {
+            mgr.run_onedrive_upload(id, od.clone(), local, final_remote, app)
+                .await
+        }
+        Session::GDrive(gd) => {
+            mgr.run_gdrive_upload(id, gd.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Box(bx) => {
+            mgr.run_box_upload(id, bx.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Shopify(sh) => {
+            mgr.run_shopify_upload(id, sh.clone(), local, final_remote, app)
+                .await
+        }
+        Session::HubSpot(hs) => {
+            mgr.run_hubspot_upload(id, hs.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Dynamics(dynm) => {
+            mgr.run_dynamics_upload(id, dynm.clone(), local, final_remote, app)
+                .await
+        }
+        Session::Agent(agent) => {
+            mgr.run_agent_upload(id, agent.clone(), local, final_remote, app)
+                .await
+        }
+    }
+}
+
 async fn finalize(
     mgr: &Arc<TransferManager>,
     id: &str,

From a923185a6ab656651756ec55d04667be84e65766 Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sun, 2 Aug 2026 00:17:22 -0400
Subject: [PATCH 4/7] feat(transfers): global bandwidth throttle
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plan 17 Phase 4: a shared token bucket every copy loop draws from per chunk
inside checkpoint() — one bucket, so the cap is split across active
transfers rather than per-transfer. Grants are capped at one second of rate
(min 64 KiB) so tiny files never wait a full window; sleeps are capped at
250 ms so a live transfer_set_throttle takes effect on the next chunk.
transferThrottleKbps persists via the settings table and is applied at
startup alongside transferConcurrency; the queue-state event now reports
the live rate.
---
 src-tauri/src/commands.rs | 11 +++++
 src-tauri/src/lib.rs      |  6 +++
 src-tauri/src/transfer.rs | 88 +++++++++++++++++++++++++++++++++++----
 3 files changed, 98 insertions(+), 7 deletions(-)

diff --git a/src-tauri/src/commands.rs b/src-tauri/src/commands.rs
index da64a7a..eb10cb1 100644
--- a/src-tauri/src/commands.rs
+++ b/src-tauri/src/commands.rs
@@ -827,6 +827,17 @@ pub async fn transfer_set_concurrency(
     Ok(())
 }
 
+/// Live-adjust the global bandwidth cap in KiB/s (0 = unlimited). Takes
+/// effect on the next chunk of every active transfer (Plan 17 Phase 4).
+#[tauri::command]
+pub async fn transfer_set_throttle(
+    kbps: u64,
+    state: State<'_, AppState>,
+) -> Result<(), String> {
+    state.transfers.set_throttle_kbps(kbps);
+    Ok(())
+}
+
 /// Queue snapshot for the panel's initial load (waiting FIFO + pause-all +
 /// concurrency + throttle).
 #[tauri::command]
diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index 6836e05..858349c 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -260,6 +260,11 @@ pub fn run() {
                         st.transfers.set_concurrency(n);
                     }
                 }
+                if let Ok(Some(raw)) = st.db.settings_get("transferThrottleKbps") {
+                    if let Ok(kbps) = serde_json::from_str::<u64>(&raw) {
+                        st.transfers.set_throttle_kbps(kbps);
+                    }
+                }
             }
 
             // Bring the Agent Bridge back up if the user left its master switch
@@ -408,6 +413,7 @@ pub fn run() {
             commands::transfer_pause_all,
             commands::transfer_resume_all,
             commands::transfer_set_concurrency,
+            commands::transfer_set_throttle,
             commands::transfer_queue_state,
             commands::rename_path,
             commands::delete_path,
diff --git a/src-tauri/src/transfer.rs b/src-tauri/src/transfer.rs
index 950385b..cd9f714 100644
--- a/src-tauri/src/transfer.rs
+++ b/src-tauri/src/transfer.rs
@@ -7,7 +7,7 @@ use anyhow::{Context, Result};
 use serde::{Deserialize, Serialize};
 use std::collections::{HashMap, VecDeque};
 use std::path::{Path, PathBuf};
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 use tauri::{AppHandle, Emitter};
@@ -69,6 +69,70 @@ pub struct QueueState {
     pub throttle_kbps: u64,
 }
 
+/// Global bandwidth cap shared by every active copy loop (Plan 17 Phase 4):
+/// a token bucket refilling at `rate` bytes/sec (0 = unlimited). Because all
+/// transfers draw from this one bucket, the cap is split across active
+/// transfers rather than applied per transfer.
+struct TokenBucket {
+    inner: Mutex<BucketInner>,
+    rate_bps: AtomicU64,
+}
+
+struct BucketInner {
+    tokens: f64,
+    last: Instant,
+}
+
+impl TokenBucket {
+    fn new() -> Self {
+        Self {
+            inner: Mutex::new(BucketInner {
+                tokens: 0.0,
+                last: Instant::now(),
+            }),
+            rate_bps: AtomicU64::new(0),
+        }
+    }
+
+    fn rate_kbps(&self) -> u64 {
+        self.rate_bps.load(Ordering::Relaxed) / 1024
+    }
+
+    fn set_rate_kbps(&self, kbps: u64) {
+        self.rate_bps
+            .store(kbps.saturating_mul(1024), Ordering::Relaxed);
+    }
+
+    /// Wait until `bytes` may flow under the cap. A grant is capped at one
+    /// second's worth of rate (min 64 KiB) so even a tiny rate always lets a
+    /// chunk through — a 1 KiB file never waits a full token window.
+    async fn acquire(&self, bytes: u64) {
+        loop {
+            let rate = self.rate_bps.load(Ordering::Relaxed);
+            if rate == 0 {
+                return;
+            }
+            let wait = {
+                let mut g = self.inner.lock().await;
+                let now = Instant::now();
+                let elapsed = now.duration_since(g.last).as_secs_f64();
+                let rate_f = rate as f64;
+                let cap = rate_f.max(64.0 * 1024.0);
+                g.tokens = (g.tokens + elapsed * rate_f).min(cap);
+                g.last = now;
+                let grant = (bytes as f64).min(cap);
+                if g.tokens >= grant {
+                    g.tokens -= grant;
+                    return;
+                }
+                Duration::from_secs_f64((grant - g.tokens) / rate_f)
+            };
+            // Cap the sleep so a live rate change takes effect promptly.
+            tokio::time::sleep(wait.min(Duration::from_millis(250))).await;
+        }
+    }
+}
+
 /// A pause gate shared by the scheduler (pause-all) and individual transfers
 /// (Phase 2). watch-channel based so waiters never miss a wakeup.
 #[derive(Debug, Clone)]
@@ -159,6 +223,8 @@ pub struct TransferManager {
     retry: Mutex<HashMap<String, RetryInfo>>,
     /// Bumped on every queue change so admission waiters re-check their turn.
     queue_gen: watch::Sender<u64>,
+    /// Global bandwidth cap every copy loop draws from per chunk (Phase 4).
+    bucket: TokenBucket,
 }
 
 fn now_ts() -> i64 {
@@ -233,6 +299,7 @@ impl TransferManager {
             pauses: Mutex::new(HashMap::new()),
             retry: Mutex::new(HashMap::new()),
             queue_gen: watch::channel(0).0,
+            bucket: TokenBucket::new(),
         }
     }
 
@@ -269,7 +336,7 @@ impl TransferManager {
             waiting: waiting.iter().cloned().collect(),
             paused_all: self.pause_all.is_paused(),
             concurrency: self.concurrency.load(Ordering::Relaxed),
-            throttle_kbps: 0, // wired up in Phase 4
+            throttle_kbps: self.bucket.rate_kbps(),
         }
     }
 
@@ -381,11 +448,12 @@ impl TransferManager {
         self.pause_all.is_paused()
     }
 
-    /// Chunk-boundary checkpoint shared by every copy loop (Plan 17). Phase 4
-    /// draws `_bytes` from the bandwidth bucket here. When the transfer (or
-    /// the whole manager) is paused, this parks until resumed, then returns
-    /// `RestartFromPause` so the runner re-runs the file from byte 0.
-    async fn checkpoint(&self, id: &str, _bytes: u64) -> Result<()> {
+    /// Chunk-boundary checkpoint shared by every copy loop (Plan 17). Draws
+    /// `bytes` from the global bandwidth bucket (Phase 4), then — when the
+    /// transfer (or the whole manager) is paused — parks until resumed and
+    /// returns `RestartFromPause` so the runner re-runs the file from byte 0.
+    async fn checkpoint(&self, id: &str, bytes: u64) -> Result<()> {
+        self.bucket.acquire(bytes).await;
         let gate = self.pauses.lock().await.get(id).cloned();
         let parked = self.pause_all.is_paused() || gate.as_ref().is_some_and(|g| g.is_paused());
         if !parked {
@@ -525,6 +593,12 @@ impl TransferManager {
         Ok(())
     }
 
+    /// Live-adjust the global bandwidth cap (KiB/s, 0 = unlimited). Takes
+    /// effect on the next chunk of every active transfer.
+    pub fn set_throttle_kbps(&self, kbps: u64) {
+        self.bucket.set_rate_kbps(kbps);
+    }
+
     /// Live-adjust the concurrency bound. Growing adds permits at once;
     /// shrinking forgets permits as running transfers release them, so
     /// in-flight transfers are never killed to satisfy the new bound.

From 01af86b70466b3d13a4524380623f8474a83f479 Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sun, 2 Aug 2026 00:25:59 -0400
Subject: [PATCH 5/7] =?UTF-8?q?feat(transfers):=20queue=20UI=20=E2=80=94?=
 =?UTF-8?q?=20pause/retry/reorder=20controls,=20throttle,=20settings?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plan 17 frontend: TransferQueue header gains pause-all/resume-all, active/
queued counts, and a live global throttle input (KiB/s, 0 = unlimited).
Rows gain pause/resume (transferring/paused), retry (error/canceled), and
move up/down with FIFO position for queued rows; mid-auto-retry rows render
"retrying in Ns (attempt N/3)" as a warning. Settings gains Concurrent
transfers + Bandwidth limit, persisted via the settings table and applied
live through transfer_set_concurrency/transfer_set_throttle. The batch
notification tracker consumes transfer://updated so canceled rows no longer
wedge a batch; paused rows still count as in flight.
---
 src/components/Settings.tsx      |  23 ++++
 src/components/TransferQueue.tsx | 207 +++++++++++++++++++++++++++----
 src/lib/ipc.ts                   |  37 +++++-
 src/lib/notifications.ts         |   8 ++
 src/lib/types.ts                 |  14 +++
 src/stores/settingsStore.ts      |  19 +++
 src/stores/transfersStore.ts     |  89 ++++++++++++-
 7 files changed, 369 insertions(+), 28 deletions(-)

diff --git a/src/components/Settings.tsx b/src/components/Settings.tsx
index 0a8db69..b61b2a9 100644
--- a/src/components/Settings.tsx
+++ b/src/components/Settings.tsx
@@ -149,6 +149,29 @@ export function Settings({ onClose }: Props) {
               checked={s.autoOpenTransferPanel}
               onChange={s.setAutoOpenTransferPanel}
             />
+            <Field
+              label="Concurrent transfers"
+              help="How many transfers run at once. Extra transfers wait in a first-in, first-out queue."
+            >
+              <NumberInput
+                min={1}
+                max={32}
+                value={s.transferConcurrency}
+                onChange={s.setTransferConcurrency}
+              />
+            </Field>
+            <Field
+              label="Bandwidth limit"
+              help="Global cap shared by all active transfers. 0 = unlimited."
+            >
+              <NumberInput
+                min={0}
+                max={1024 * 1024}
+                value={s.transferThrottleKbps}
+                onChange={s.setTransferThrottleKbps}
+                suffix="KiB/s"
+              />
+            </Field>
             <Field
               label="Download folder"
               help="Where downloads land. Leave blank to use your system Downloads folder."
diff --git a/src/components/TransferQueue.tsx b/src/components/TransferQueue.tsx
index c6c6b6b..05dd923 100644
--- a/src/components/TransferQueue.tsx
+++ b/src/components/TransferQueue.tsx
@@ -1,12 +1,40 @@
-import { useEffect } from "react";
-import { ArrowDownToLine, ArrowUpFromLine, X, ChevronDown, Trash2, ArrowDownUp } from "lucide-react";
+import { useEffect, useState } from "react";
+import {
+  ArrowDownToLine,
+  ArrowUpFromLine,
+  X,
+  ChevronDown,
+  ChevronUp,
+  Trash2,
+  ArrowDownUp,
+  Pause,
+  Play,
+  RotateCcw,
+} from "lucide-react";
 import { useTransfers } from "@/stores/transfersStore";
 import type { Transfer } from "@/lib/types";
 import { cn } from "@/lib/cn";
 
 export function TransferQueue() {
-  const { byId, panelOpen, setPanelOpen, cancel, clearFinished, initListeners, loadInitial } =
-    useTransfers();
+  const {
+    byId,
+    panelOpen,
+    setPanelOpen,
+    cancel,
+    clearFinished,
+    initListeners,
+    loadInitial,
+    queue,
+    pausedAll,
+    throttleKbps,
+    pauseAll,
+    resumeAll,
+    setThrottle,
+    pause,
+    resume,
+    retry,
+    move,
+  } = useTransfers();
 
   useEffect(() => {
     let unsub: (() => void) | undefined;
@@ -23,8 +51,9 @@ export function TransferQueue() {
   if (!panelOpen) return null;
 
   const active = transfers.filter(
-    (t) => t.status === "transferring" || t.status === "queued"
+    (t) => t.status === "transferring" || t.status === "paused"
   ).length;
+  const queued = queue.length;
 
   return (
     <div className="anim-slide-up flex max-h-64 flex-col border-t border-border bg-bg-panel">
@@ -32,12 +61,20 @@ export function TransferQueue() {
         <span className="text-xs font-semibold uppercase tracking-wider text-text-muted">
           Transfers
         </span>
-        {active > 0 && (
+        {(active > 0 || queued > 0) && (
           <span className="rounded bg-accent/20 px-1.5 py-0.5 text-[10px] font-medium text-accent">
-            {active} active
+            {active} active · {queued} queued
           </span>
         )}
         <div className="flex-1" />
+        <ThrottleInput value={throttleKbps} onCommit={setThrottle} />
+        <button
+          onClick={() => (pausedAll ? resumeAll() : pauseAll())}
+          className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text"
+          title={pausedAll ? "Resume all" : "Pause all"}
+        >
+          {pausedAll ? <Play size={12} /> : <Pause size={12} />}
+        </button>
         <button
           onClick={clearFinished}
           className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text"
@@ -66,16 +103,98 @@ export function TransferQueue() {
           </div>
         )}
         {transfers.map((t) => (
-          <Row key={t.id} t={t} onCancel={() => cancel(t.id)} />
+          <Row
+            key={t.id}
+            t={t}
+            queue={queue}
+            onCancel={() => cancel(t.id)}
+            onPause={() => pause(t.id)}
+            onResume={() => resume(t.id)}
+            onRetry={() => retry(t.id)}
+            onMove={(dir) => move(t.id, dir)}
+          />
         ))}
       </div>
     </div>
   );
 }
 
-function Row({ t, onCancel }: { t: Transfer; onCancel: () => void }) {
+/** Global bandwidth cap (KiB/s, 0 = unlimited). Commits on blur or Enter. */
+function ThrottleInput({
+  value,
+  onCommit,
+}: {
+  value: number;
+  onCommit: (kbps: number) => void;
+}) {
+  const [draft, setDraft] = useState(String(value));
+
+  // Follow external changes (settings sync / other windows) while not editing.
+  useEffect(() => {
+    setDraft(String(value));
+  }, [value]);
+
+  const commit = () => {
+    const n = Math.max(0, parseInt(draft) || 0);
+    setDraft(String(n));
+    if (n !== value) onCommit(n);
+  };
+
+  return (
+    <span className="flex items-center gap-1" title="Bandwidth limit (KiB/s, 0 = unlimited)">
+      <input
+        type="number"
+        min={0}
+        value={draft}
+        onChange={(e) => setDraft(e.target.value)}
+        onBlur={commit}
+        onKeyDown={(e) => {
+          if (e.key === "Enter") (e.target as HTMLInputElement).blur();
+        }}
+        className="w-14 rounded border border-border bg-bg-panel px-1.5 py-0.5 text-[11px] text-text outline-none focus:border-accent"
+      />
+      <span className="text-[10px] text-text-dim">KiB/s</span>
+    </span>
+  );
+}
+
+function Row({
+  t,
+  queue,
+  onCancel,
+  onPause,
+  onResume,
+  onRetry,
+  onMove,
+}: {
+  t: Transfer;
+  queue: string[];
+  onCancel: () => void;
+  onPause: () => void;
+  onResume: () => void;
+  onRetry: () => void;
+  onMove: (dir: "up" | "down") => void;
+}) {
   const Icon = t.kind === "download" ? ArrowDownToLine : ArrowUpFromLine;
   const pct = t.size > 0 ? Math.min(100, (t.transferred / t.size) * 100) : 0;
+  // Mid-auto-retry: the error text reads "retrying in Ns (attempt N/3)".
+  const retrying = t.retryAttempt !== undefined && !!t.error;
+  const queuePos = queue.indexOf(t.id);
+
+  const statusLabel =
+    t.status === "transferring"
+      ? `${pct.toFixed(0)}%`
+      : t.status === "done"
+        ? "done"
+        : t.status === "error"
+          ? "error"
+          : t.status === "canceled"
+            ? "canceled"
+            : t.status === "paused"
+              ? "Paused"
+              : t.status === "queued" && queuePos >= 0
+                ? `#${queuePos + 1} in queue`
+                : t.status;
 
   return (
     <div className="flex items-center gap-3 border-b border-border-subtle px-3 py-2 text-sm">
@@ -99,24 +218,70 @@ function Row({ t, onCancel }: { t: Transfer; onCancel: () => void }) {
           </div>
         )}
         {t.error && (
-          <div className="mt-1 text-[11px] text-danger">{t.error}</div>
+          <div
+            className={cn(
+              "mt-1 text-[11px]",
+              retrying ? "text-warning" : "text-danger"
+            )}
+          >
+            {t.error}
+          </div>
         )}
       </div>
       <div className="w-20 shrink-0 text-right text-[11px] text-text-muted">
         <div>{fmtSize(t.transferred)}</div>
-        <div className="text-text-dim">
-          {t.status === "transferring"
-            ? `${pct.toFixed(0)}%`
-            : t.status === "done"
-              ? "done"
-              : t.status === "error"
-                ? "error"
-                : t.status === "canceled"
-                  ? "canceled"
-                  : t.status}
-        </div>
+        <div className="text-text-dim">{statusLabel}</div>
       </div>
+      {t.status === "queued" && queuePos >= 0 && (
+        <>
+          <button
+            onClick={() => onMove("up")}
+            disabled={queuePos === 0}
+            className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text disabled:opacity-30 disabled:hover:bg-transparent"
+            title="Move up"
+          >
+            <ChevronUp size={12} />
+          </button>
+          <button
+            onClick={() => onMove("down")}
+            disabled={queuePos === queue.length - 1}
+            className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text disabled:opacity-30 disabled:hover:bg-transparent"
+            title="Move down"
+          >
+            <ChevronDown size={12} />
+          </button>
+        </>
+      )}
       {(t.status === "transferring" || t.status === "queued") && (
+        <button
+          onClick={onPause}
+          className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text"
+          title="Pause"
+        >
+          <Pause size={12} />
+        </button>
+      )}
+      {t.status === "paused" && (
+        <button
+          onClick={onResume}
+          className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text"
+          title="Resume (restarts from byte 0)"
+        >
+          <Play size={12} />
+        </button>
+      )}
+      {(t.status === "error" || t.status === "canceled") && (
+        <button
+          onClick={onRetry}
+          className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text"
+          title="Retry"
+        >
+          <RotateCcw size={12} />
+        </button>
+      )}
+      {(t.status === "transferring" ||
+        t.status === "queued" ||
+        t.status === "paused") && (
         <button
           onClick={onCancel}
           className="rounded p-1 text-text-muted hover:bg-bg-hover hover:text-text"
diff --git a/src/lib/ipc.ts b/src/lib/ipc.ts
index c00f95b..e369c42 100644
--- a/src/lib/ipc.ts
+++ b/src/lib/ipc.ts
@@ -23,6 +23,7 @@ import type {
   TerminalDataEvent,
   TerminalExitEvent,
   Transfer,
+  TransferQueueState,
   OverwritePolicy,
   BridgeStatus,
   BridgeActivity,
@@ -244,6 +245,30 @@ export const ipc = {
 
   listTransfers: () => invoke<Transfer[]>("list_transfers"),
 
+  transferPause: (transferId: string) =>
+    invoke<void>("transfer_pause", { transferId }),
+
+  transferResume: (transferId: string) =>
+    invoke<void>("transfer_resume", { transferId }),
+
+  transferRetry: (transferId: string) =>
+    invoke<void>("transfer_retry", { transferId }),
+
+  transferMove: (transferId: string, direction: "up" | "down") =>
+    invoke<void>("transfer_move", { transferId, direction }),
+
+  transferPauseAll: () => invoke<void>("transfer_pause_all"),
+
+  transferResumeAll: () => invoke<void>("transfer_resume_all"),
+
+  transferSetConcurrency: (count: number) =>
+    invoke<void>("transfer_set_concurrency", { count }),
+
+  transferSetThrottle: (kbps: number) =>
+    invoke<void>("transfer_set_throttle", { kbps }),
+
+  transferQueueState: () => invoke<TransferQueueState>("transfer_queue_state"),
+
   startDirectoryDownload: (
     sessionId: SessionId,
     remoteDir: string,
@@ -676,15 +701,25 @@ export async function onSearchEvent(
 }
 
 export async function onTransferEvent(
-  cb: (kind: "added" | "progress" | "done" | "error", t: Transfer) => void
+  cb: (
+    kind: "added" | "progress" | "done" | "error" | "updated",
+    t: Transfer
+  ) => void
 ): Promise<UnlistenFn> {
   const unsubs = await Promise.all([
     listen<Transfer>("transfer://added", (e) => cb("added", e.payload)),
     listen<Transfer>("transfer://progress", (e) => cb("progress", e.payload)),
     listen<Transfer>("transfer://done", (e) => cb("done", e.payload)),
     listen<Transfer>("transfer://error", (e) => cb("error", e.payload)),
+    listen<Transfer>("transfer://updated", (e) => cb("updated", e.payload)),
   ]);
   return () => {
     unsubs.forEach((u) => u());
   };
 }
+
+export async function onTransferQueue(
+  cb: (q: TransferQueueState) => void
+): Promise<UnlistenFn> {
+  return listen<TransferQueueState>("transfer://queue", (e) => cb(e.payload));
+}
diff --git a/src/lib/notifications.ts b/src/lib/notifications.ts
index ff953ee..681a2e7 100644
--- a/src/lib/notifications.ts
+++ b/src/lib/notifications.ts
@@ -146,6 +146,14 @@ export function initNotifications(): () => void {
           if (active.size === 0) flushBatch();
         }
         break;
+      case "updated":
+        // Canceled rows emit no terminal event — without this they'd wedge the
+        // batch forever. A paused row is still in flight, so it stays in
+        // `active` (it emits nothing until resumed).
+        if (t.status === "canceled" && active.delete(id)) {
+          if (active.size === 0) flushBatch();
+        }
+        break;
     }
   }));
 
diff --git a/src/lib/types.ts b/src/lib/types.ts
index 1cff0bf..a7b32c9 100644
--- a/src/lib/types.ts
+++ b/src/lib/types.ts
@@ -1017,6 +1017,7 @@ export type TransferKind = "download" | "upload";
 export type TransferStatus =
   | "queued"
   | "transferring"
+  | "paused"
   | "done"
   | "skipped"
   | "error"
@@ -1032,9 +1033,22 @@ export interface Transfer {
   transferred: number;
   status: TransferStatus;
   error?: string;
+  /** Set while the backend auto-retries a failed transfer (`error` reads
+   *  "retrying in Ns (attempt N/3)"). */
+  retryAttempt?: number;
   startedAt: number;
 }
 
+/** Live transfer-queue snapshot (Plan 17) — payload of `transfer://queue` and
+ *  the return of `transfer_queue_state`. `waiting` holds the FIFO order of
+ *  queued transfer ids. */
+export interface TransferQueueState {
+  waiting: string[];
+  pausedAll: boolean;
+  concurrency: number;
+  throttleKbps: number;
+}
+
 /** What an encrypted backup contains (Plan 12 Phase 4). */
 export interface BackupSummary {
   profiles: number;
diff --git a/src/stores/settingsStore.ts b/src/stores/settingsStore.ts
index 75c9d98..ee8a341 100644
--- a/src/stores/settingsStore.ts
+++ b/src/stores/settingsStore.ts
@@ -91,6 +91,10 @@ interface SettingsState {
    *  `overwritePolicy` silently (the pre-prompt behaviour). */
   promptOnOverwrite: boolean;
   autoOpenTransferPanel: boolean;
+  /** Max simultaneous transfers (1–32). Live-applied to the backend queue. */
+  transferConcurrency: number;
+  /** Global bandwidth cap in KiB/s (0 = unlimited). Live-applied. */
+  transferThrottleKbps: number;
   /** Where downloads land. Blank = the OS Downloads folder. */
   defaultDownloadFolder: string;
   /** Command/path used to open files for edit-in-place. Blank = OS default app. */
@@ -135,6 +139,8 @@ interface SettingsState {
   setOverwritePolicy: (p: OverwritePolicy) => void;
   setPromptOnOverwrite: (v: boolean) => void;
   setAutoOpenTransferPanel: (v: boolean) => void;
+  setTransferConcurrency: (n: number) => void;
+  setTransferThrottleKbps: (n: number) => void;
   setDefaultDownloadFolder: (s: string) => void;
   setDefaultEditor: (s: string) => void;
   setShowHiddenFiles: (v: boolean) => void;
@@ -165,6 +171,8 @@ type Persisted = Omit<
   | "setOverwritePolicy"
   | "setPromptOnOverwrite"
   | "setAutoOpenTransferPanel"
+  | "setTransferConcurrency"
+  | "setTransferThrottleKbps"
   | "setDefaultDownloadFolder"
   | "setDefaultEditor"
   | "setShowHiddenFiles"
@@ -192,6 +200,8 @@ const DEFAULTS: Persisted = {
   overwritePolicy: "overwrite",
   promptOnOverwrite: true,
   autoOpenTransferPanel: true,
+  transferConcurrency: 3,
+  transferThrottleKbps: 0,
   defaultDownloadFolder: "",
   defaultEditor: "",
   showHiddenFiles: false,
@@ -279,6 +289,15 @@ export const useSettings = create<SettingsState>((set, get) => ({
   setPromptOnOverwrite: (v) => mutate(set, get, "promptOnOverwrite", v),
   setAutoOpenTransferPanel: (v) =>
     mutate(set, get, "autoOpenTransferPanel", v),
+  setTransferConcurrency: (n) => {
+    mutate(set, get, "transferConcurrency", n);
+    // Live-apply to the running queue; the persisted value covers next launch.
+    ipc.transferSetConcurrency(n).catch(() => {});
+  },
+  setTransferThrottleKbps: (n) => {
+    mutate(set, get, "transferThrottleKbps", n);
+    ipc.transferSetThrottle(n).catch(() => {});
+  },
   setDefaultDownloadFolder: (s) =>
     mutate(set, get, "defaultDownloadFolder", s),
   setDefaultEditor: (s) => mutate(set, get, "defaultEditor", s),
diff --git a/src/stores/transfersStore.ts b/src/stores/transfersStore.ts
index 8926120..8e4f1f3 100644
--- a/src/stores/transfersStore.ts
+++ b/src/stores/transfersStore.ts
@@ -1,5 +1,5 @@
 import { create } from "zustand";
-import { ipc, onTransferEvent } from "@/lib/ipc";
+import { ipc, onTransferEvent, onTransferQueue } from "@/lib/ipc";
 import { useSettings } from "./settingsStore";
 import { toast } from "./toastStore";
 import { useConflicts, type ConflictDecision } from "./conflictStore";
@@ -38,6 +38,11 @@ export function toTransferItem(e: {
 interface TransfersState {
   byId: Record<string, Transfer>;
   panelOpen: boolean;
+  /** Waiting (queued) transfer ids in FIFO order — position = index + 1. */
+  queue: string[];
+  pausedAll: boolean;
+  concurrency: number;
+  throttleKbps: number;
 
   togglePanel: () => void;
   setPanelOpen: (open: boolean) => void;
@@ -84,6 +89,14 @@ interface TransfersState {
     remoteDir: string
   ) => Promise<void>;
   cancel: (id: string) => Promise<void>;
+  pause: (id: string) => Promise<void>;
+  resume: (id: string) => Promise<void>;
+  retry: (id: string) => Promise<void>;
+  move: (id: string, dir: "up" | "down") => Promise<void>;
+  pauseAll: () => Promise<void>;
+  resumeAll: () => Promise<void>;
+  setConcurrency: (n: number) => Promise<void>;
+  setThrottle: (kbps: number) => Promise<void>;
   clearFinished: () => void;
 }
 
@@ -177,12 +190,19 @@ function indexBy(transfers: Transfer[]): Record<string, Transfer> {
 export const useTransfers = create<TransfersState>((set, get) => ({
   byId: {},
   panelOpen: false,
+  queue: [],
+  pausedAll: false,
+  concurrency: 3,
+  throttleKbps: 0,
 
   togglePanel: () => set((s) => ({ panelOpen: !s.panelOpen })),
   setPanelOpen: (open) => set({ panelOpen: open }),
 
   initListeners: async () => {
-    const unlisten = await onTransferEvent((kind, t) => {
+    const unlistenEvents = await onTransferEvent((kind, t) => {
+      // "updated" is treated exactly like progress: replace the byId entry.
+      // It never toasts (it also carries the mid-auto-retry "retrying in Ns"
+      // state, which is not a terminal outcome).
       set((s) => ({ byId: { ...s.byId, [t.id]: t } }));
       if (kind === "added" && !get().panelOpen) {
         if (useSettings.getState().autoOpenTransferPanel) {
@@ -201,12 +221,32 @@ export const useTransfers = create<TransfersState>((set, get) => ({
         toast.error("Transfer failed", t.error || baseName(t.source));
       }
     });
-    return unlisten;
+    const unlistenQueue = await onTransferQueue((q) => {
+      set({
+        queue: q.waiting,
+        pausedAll: q.pausedAll,
+        concurrency: q.concurrency,
+        throttleKbps: q.throttleKbps,
+      });
+    });
+    return () => {
+      unlistenEvents();
+      unlistenQueue();
+    };
   },
 
   loadInitial: async () => {
-    const list = await ipc.listTransfers();
-    set({ byId: indexBy(list) });
+    const [list, q] = await Promise.all([
+      ipc.listTransfers(),
+      ipc.transferQueueState(),
+    ]);
+    set({
+      byId: indexBy(list),
+      queue: q.waiting,
+      pausedAll: q.pausedAll,
+      concurrency: q.concurrency,
+      throttleKbps: q.throttleKbps,
+    });
   },
 
   download: async (sessionId, remotePath, localDir, policy) => {
@@ -249,11 +289,48 @@ export const useTransfers = create<TransfersState>((set, get) => ({
     await ipc.cancelTransfer(id);
   },
 
+  pause: async (id) => {
+    await ipc.transferPause(id);
+  },
+
+  resume: async (id) => {
+    await ipc.transferResume(id);
+  },
+
+  retry: async (id) => {
+    await ipc.transferRetry(id);
+  },
+
+  move: async (id, dir) => {
+    await ipc.transferMove(id, dir);
+  },
+
+  pauseAll: async () => {
+    await ipc.transferPauseAll();
+  },
+
+  resumeAll: async () => {
+    await ipc.transferResumeAll();
+  },
+
+  setConcurrency: async (n) => {
+    await ipc.transferSetConcurrency(n);
+  },
+
+  setThrottle: async (kbps) => {
+    await ipc.transferSetThrottle(kbps);
+  },
+
   clearFinished: () =>
     set((s) => {
       const next: Record<string, Transfer> = {};
       for (const t of Object.values(s.byId)) {
-        if (t.status === "transferring" || t.status === "queued") {
+        // Paused rows are still in flight — keep them like active ones.
+        if (
+          t.status === "transferring" ||
+          t.status === "queued" ||
+          t.status === "paused"
+        ) {
           next[t.id] = t;
         }
       }

From b7db9559fc314e55731e63e6e714a264e4a50637 Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sun, 2 Aug 2026 00:38:31 -0400
Subject: [PATCH 6/7] test(transfers): scheduler/bucket unit tests + headless
 verify harness
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plan 17 verification: 8 Rust unit tests (token-bucket pacing under a paused
clock incl. full-amount charging for large chunks, tiny-file floor; pause
gate park/unpark; checkpoint → RestartFromPause; FIFO skip-paused + pause-all
precedence; concurrency grow/shrink; transient vs permanent retry
classification). Testing surfaced two real bugs, both fixed here: PauseGate
dropped watch::channel's initial receiver so send() silently failed (pause
would never have engaged in production), and TokenBucket charged only one
rate-second tranche per acquire (large chunks burst past the cap).
scripts/verify-transfers.mjs + a mock transfer-queue engine drive the panel
headlessly (25 checks: counts, positions, pause/resume/pause-all toggles,
retry/move/throttle command dispatch, retrying-in-Ns rendering); the
notifications harness still passes. tokio test-util added as a dev-dep.
---
 scripts/verify-transfers.mjs | 317 +++++++++++++++++++++++++++++++++++
 src-tauri/Cargo.toml         |   4 +
 src-tauri/src/transfer.rs    | 167 ++++++++++++++++--
 src/mock/core.ts             |  17 +-
 src/mock/demo.ts             |   5 +
 src/mock/transfers.ts        | 114 +++++++++++++
 6 files changed, 608 insertions(+), 16 deletions(-)
 create mode 100644 scripts/verify-transfers.mjs
 create mode 100644 src/mock/transfers.ts

diff --git a/scripts/verify-transfers.mjs b/scripts/verify-transfers.mjs
new file mode 100644
index 0000000..6cd9323
--- /dev/null
+++ b/scripts/verify-transfers.mjs
@@ -0,0 +1,317 @@
+// Headless runtime verification for Plan 17 (transfer queue depth). Spins up
+// the mock Vite build (no Rust), drives the transfer panel via window.__demo
+// plus the mock transfer engine (src/mock/transfers.ts, which records the
+// commands the UI invokes), and asserts on the real rendered DOM:
+// counts/positions, pause/resume, pause-all, retry, reorder, throttle, and the
+// mid-auto-retry rendering. Exit code 0 = all checks passed.
+import { spawn } from "node:child_process";
+import { setTimeout as sleep } from "node:timers/promises";
+import { existsSync } from "node:fs";
+import { fileURLToPath } from "node:url";
+import path from "node:path";
+import puppeteer from "puppeteer-core";
+
+const __dirname = path.dirname(fileURLToPath(import.meta.url));
+const ROOT = path.resolve(__dirname, "..");
+const OUT =
+  process.env.SHOT_DIR ||
+  "C:/Users/Juan/AppData/Local/Temp/claude/C--Users-Juan-Documents-GitHub-Faro/cbf0dfb0-7a23-418f-834f-25dcd5128966/scratchpad";
+const PORT = 1425;
+const URL = `http://localhost:${PORT}/`;
+
+const CHROME_CANDIDATES = [
+  "C:/Program Files/Google/Chrome/Application/chrome.exe",
+  "C:/Program Files (x86)/Google/Chrome/Application/chrome.exe",
+  "C:/Program Files (x86)/Microsoft/Edge/Application/msedge.exe",
+  "C:/Program Files/Microsoft/Edge/Application/msedge.exe",
+];
+const chrome = CHROME_CANDIDATES.find((p) => existsSync(p));
+if (!chrome) {
+  console.error("No Chrome/Edge found.");
+  process.exit(2);
+}
+const isWin = process.platform === "win32";
+
+let failures = 0;
+function check(name, cond, detail = "") {
+  const ok = !!cond;
+  if (!ok) failures++;
+  console.log(`  ${ok ? "✓ PASS" : "✗ FAIL"}  ${name}${detail ? "  — " + detail : ""}`);
+}
+
+async function waitForServer(url, ms = 60_000) {
+  const deadline = Date.now() + ms;
+  while (Date.now() < deadline) {
+    try {
+      const r = await fetch(url);
+      if (r.ok) return;
+    } catch {}
+    await sleep(500);
+  }
+  throw new Error(`mock server never came up at ${url}`);
+}
+function killTree(child) {
+  if (!child || child.killed) return;
+  if (isWin) spawn("taskkill", ["/pid", String(child.pid), "/T", "/F"], { stdio: "ignore" });
+  else try { process.kill(-child.pid, "SIGKILL"); } catch {}
+}
+
+async function main() {
+  console.log("› starting mock Vite server…");
+  const server = spawn("npm", ["run", "dev:mock"], {
+    cwd: ROOT,
+    shell: true,
+    stdio: "ignore",
+    detached: !isWin,
+  });
+  let browser;
+  try {
+    await waitForServer(URL);
+    browser = await puppeteer.launch({
+      executablePath: chrome,
+      headless: true,
+      defaultViewport: { width: 1440, height: 900, deviceScaleFactor: 1 },
+      args: ["--hide-scrollbars"],
+    });
+    const page = await browser.newPage();
+    page.on("pageerror", (e) => {
+      console.log("  page error:", e.message);
+      failures++;
+    });
+
+    const drive = (fn, ...a) => page.evaluate(fn, ...a);
+    const shot = async (name) => {
+      await page.screenshot({ path: path.join(OUT, `verify-transfers-${name}.png`) });
+      console.log("    · shot", `verify-transfers-${name}.png`);
+    };
+
+    let booted = false;
+    for (let attempt = 0; attempt < 5 && !booted; attempt++) {
+      await page.goto(URL, { waitUntil: "networkidle0" });
+      try {
+        await page.waitForFunction(() => !!window.__demo?.seedTransfers, { timeout: 8_000 });
+        booted = true;
+      } catch {
+        console.log(`    · boot attempt ${attempt + 1} missed __demo, reloading…`);
+        await sleep(1500);
+      }
+    }
+    if (!booted) throw new Error("app never exposed window.__demo");
+    await sleep(800); // let App effects (loadInitial + initListeners) run
+
+    // In-page helpers: locate a transfer row by its source path, click one of
+    // its buttons by title, and read the recorded mock transfer calls.
+    await drive(() => {
+      window.__vt = {
+        row(src) {
+          const el = [...document.querySelectorAll(".font-mono")].find(
+            (e) => e.textContent === src
+          );
+          return el?.closest("div.flex.items-center.gap-3") ?? null;
+        },
+        btn(src, title) {
+          const r = window.__vt.row(src);
+          return r?.querySelector(`button[title="${title}"]`) ?? null;
+        },
+        click(src, title) {
+          window.__vt.btn(src, title)?.click();
+        },
+        clickHeader(title) {
+          document.querySelector(`button[title="${title}"]`)?.click();
+        },
+        calls(cmd) {
+          return window.__demo.transferCalls.filter((c) => c.cmd === cmd);
+        },
+      };
+    });
+    const rowText = (src) =>
+      drive((s) => window.__vt.row(s)?.innerText ?? "", src);
+
+    const SRC = {
+      a1: "/srv/batch/a1.bin",
+      a2: "/srv/batch/a2.bin",
+      q1: "/srv/batch/q1.bin",
+      q2: "/srv/batch/q2.bin",
+      e1: "/srv/batch/e1.bin",
+      r1: "/srv/batch/r1.bin",
+    };
+    const seedBatch = () =>
+      drive((SRC) => {
+        const mk = (id, status, extra = {}) => ({
+          id,
+          kind: "download",
+          source: `/srv/batch/${id}.bin`,
+          destination: `C:\\Users\\demo\\Downloads\\${id}.bin`,
+          size: 1000,
+          transferred: 400,
+          status,
+          startedAt: Date.now(),
+          ...extra,
+        });
+        window.__demo.seedTransfers(
+          [
+            mk("a1", "transferring"),
+            mk("a2", "transferring"),
+            mk("q1", "queued", { transferred: 0 }),
+            mk("q2", "queued", { transferred: 0 }),
+            mk("e1", "error", { transferred: 0, error: "connection reset" }),
+          ],
+          { waiting: ["q1", "q2"], concurrency: 2, throttleKbps: 0 }
+        );
+        const st = window.__demo.useTransfers.getState();
+        st.setPanelOpen(true);
+        return st.loadInitial();
+      }, SRC);
+
+    // ---- 1. Seeded batch: header counts + queue positions ----
+    await seedBatch();
+    await sleep(400);
+    const hdr = await drive(() => document.body.innerText);
+    check(
+      "header shows '2 active · 2 queued'",
+      hdr.includes("2 active · 2 queued"),
+      hdr.match(/\d+ active · \d+ queued/)?.[0] ?? "no badge"
+    );
+    check("first queued row shows '#1 in queue'", (await rowText(SRC.q1)).includes("#1 in queue"));
+    check("second queued row shows '#2 in queue'", (await rowText(SRC.q2)).includes("#2 in queue"));
+    await shot("1-batch");
+
+    // ---- 2. Pause a transferring row → Paused; resume → transferring ----
+    await drive((s) => window.__vt.click(s, "Pause"), SRC.a1);
+    await sleep(300);
+    let t = await rowText(SRC.a1);
+    check("paused row shows Paused label", t.includes("Paused"), t.replace(/\n/g, " | "));
+    check(
+      "paused row offers Resume",
+      await drive((s) => !!window.__vt.btn(s, "Resume (restarts from byte 0)"), SRC.a1)
+    );
+    check(
+      "pause invoked transfer_pause on the mock",
+      (await drive((s) => window.__vt.calls("transfer_pause").map((c) => c.args.transferId), SRC.a1)).includes("a1")
+    );
+    await drive((s) => window.__vt.click(s, "Resume (restarts from byte 0)"), SRC.a1);
+    await sleep(300);
+    t = await rowText(SRC.a1);
+    check("resumed row is transferring again (0%)", t.includes("0%"), t.replace(/\n/g, " | "));
+    check(
+      "resume invoked transfer_resume on the mock",
+      (await drive(() => window.__vt.calls("transfer_resume").map((c) => c.args.transferId))).includes("a1")
+    );
+
+    // ---- 3. Pause-all → header toggle flips; queued rows stay queued ----
+    await drive(() => window.__vt.clickHeader("Pause all"));
+    await sleep(300);
+    check(
+      "pause-all flips the header toggle to Resume all",
+      await drive(() => !!document.querySelector('button[title="Resume all"]'))
+    );
+    check("queued rows stay queued under pause-all", (await rowText(SRC.q1)).includes("#1 in queue"));
+    await shot("3-pause-all");
+    await drive(() => window.__vt.clickHeader("Resume all"));
+    await sleep(300);
+    check(
+      "resume-all flips the header toggle back to Pause all",
+      await drive(() => !!document.querySelector('button[title="Pause all"]'))
+    );
+
+    // ---- 4. Retry an error row → transfer_retry recorded, row re-queued ----
+    await drive((s) => window.__vt.click(s, "Retry"), SRC.e1);
+    await sleep(300);
+    check(
+      "retry invoked transfer_retry on the mock",
+      (await drive(() => window.__vt.calls("transfer_retry").map((c) => c.args.transferId))).includes("e1")
+    );
+    t = await rowText(SRC.e1);
+    check("retried row returns to the queue tail (#3)", t.includes("#3 in queue"), t.replace(/\n/g, " | "));
+    check("retried row no longer offers Retry", await drive((s) => !window.__vt.btn(s, "Retry"), SRC.e1));
+
+    // ---- 5. Reorder: move q2 up; FIFO ends disable the buttons ----
+    await drive((s) => window.__vt.click(s, "Move up"), SRC.q2);
+    await sleep(300);
+    const moveCalls = await drive(() => window.__vt.calls("transfer_move"));
+    check(
+      "move invoked transfer_move(up) on q2",
+      moveCalls.some((c) => c.args.transferId === "q2" && c.args.direction === "up"),
+      JSON.stringify(moveCalls)
+    );
+    check("q2 is now '#1 in queue'", (await rowText(SRC.q2)).includes("#1 in queue"));
+    check(
+      "Move up disabled at the FIFO head",
+      await drive((s) => window.__vt.btn(s, "Move up")?.disabled === true, SRC.q2)
+    );
+    check(
+      "Move down disabled at the FIFO tail",
+      await drive((s) => window.__vt.btn(s, "Move down")?.disabled === true, SRC.e1)
+    );
+    await shot("5-reorder");
+
+    // ---- 6. Throttle input commits → transfer_set_throttle ----
+    await drive(() => {
+      const input = document.querySelector('input[type="number"]');
+      input.focus();
+      const setter = Object.getOwnPropertyDescriptor(
+        window.HTMLInputElement.prototype,
+        "value"
+      ).set;
+      setter.call(input, "512");
+      input.dispatchEvent(new Event("input", { bubbles: true }));
+      input.blur();
+    });
+    await sleep(300);
+    const throttleCalls = await drive(() => window.__vt.calls("transfer_set_throttle"));
+    check(
+      "throttle commit invoked transfer_set_throttle(512)",
+      throttleCalls.some((c) => c.args.kbps === 512),
+      JSON.stringify(throttleCalls)
+    );
+    check(
+      "store throttleKbps followed the queue event",
+      (await drive(() => window.__demo.useTransfers.getState().throttleKbps)) === 512
+    );
+
+    // ---- 7. Mid-auto-retry row renders the 'retrying in Ns' text as warning ----
+    await drive((SRC) => {
+      window.__demo.seedTransfers(
+        [
+          {
+            id: "r1",
+            kind: "upload",
+            source: SRC.r1,
+            destination: "/srv/app/r1.bin",
+            size: 2048,
+            transferred: 512,
+            status: "transferring",
+            retryAttempt: 2,
+            error: "retrying in 5s (attempt 2/3)",
+            startedAt: Date.now(),
+          },
+        ],
+        { waiting: [] }
+      );
+      return window.__demo.useTransfers.getState().loadInitial();
+    }, SRC);
+    await sleep(400);
+    t = await rowText(SRC.r1);
+    check("mid-auto-retry row renders the retrying text", t.includes("retrying in 5s (attempt 2/3)"), t.replace(/\n/g, " | "));
+    check(
+      "retrying text is styled as warning, not failure",
+      await drive(
+        (s) => !!window.__vt.row(s)?.querySelector(".text-warning"),
+        SRC.r1
+      )
+    );
+    await shot("7-retrying");
+
+    if (failures === 0) console.log("\n✅ all transfer-queue checks passed");
+    else console.log(`\n❌ ${failures} check(s) failed`);
+  } finally {
+    if (browser) await browser.close();
+    killTree(server);
+  }
+  process.exit(failures === 0 ? 0 : 1);
+}
+
+main().catch((e) => {
+  console.error("verify crashed:", e);
+  process.exit(3);
+});
diff --git a/src-tauri/Cargo.toml b/src-tauri/Cargo.toml
index 1f85e91..7d1fa24 100644
--- a/src-tauri/Cargo.toml
+++ b/src-tauri/Cargo.toml
@@ -167,6 +167,10 @@ windows = { version = "0.58", optional = true, features = [
     "Win32_Storage_CloudFilters",
 ] }
 
+[dev-dependencies]
+# Paused-clock tokio tests for the transfer token bucket (Plan 17).
+tokio = { version = "1", features = ["test-util"] }
+
 [features]
 default = ["custom-protocol"]
 custom-protocol = ["tauri/custom-protocol"]
diff --git a/src-tauri/src/transfer.rs b/src-tauri/src/transfer.rs
index cd9f714..3c74a4c 100644
--- a/src-tauri/src/transfer.rs
+++ b/src-tauri/src/transfer.rs
@@ -80,7 +80,8 @@ struct TokenBucket {
 
 struct BucketInner {
     tokens: f64,
-    last: Instant,
+    // tokio's Instant (not std's) so `start_paused` tests drive refills.
+    last: tokio::time::Instant,
 }
 
 impl TokenBucket {
@@ -88,7 +89,7 @@ impl TokenBucket {
         Self {
             inner: Mutex::new(BucketInner {
                 tokens: 0.0,
-                last: Instant::now(),
+                last: tokio::time::Instant::now(),
             }),
             rate_bps: AtomicU64::new(0),
         }
@@ -103,29 +104,31 @@ impl TokenBucket {
             .store(kbps.saturating_mul(1024), Ordering::Relaxed);
     }
 
-    /// Wait until `bytes` may flow under the cap. A grant is capped at one
-    /// second's worth of rate (min 64 KiB) so even a tiny rate always lets a
-    /// chunk through — a 1 KiB file never waits a full token window.
+    /// Wait until `bytes` may flow under the cap. Drawn in tranches capped at
+    /// one second's worth of rate (min 64 KiB) so a large chunk is charged in
+    /// full while a 1 KiB file never waits a whole token window.
     async fn acquire(&self, bytes: u64) {
-        loop {
+        let mut remaining = bytes as f64;
+        while remaining > 0.0 {
             let rate = self.rate_bps.load(Ordering::Relaxed);
             if rate == 0 {
                 return;
             }
             let wait = {
                 let mut g = self.inner.lock().await;
-                let now = Instant::now();
+                let now = tokio::time::Instant::now();
                 let elapsed = now.duration_since(g.last).as_secs_f64();
                 let rate_f = rate as f64;
                 let cap = rate_f.max(64.0 * 1024.0);
                 g.tokens = (g.tokens + elapsed * rate_f).min(cap);
                 g.last = now;
-                let grant = (bytes as f64).min(cap);
-                if g.tokens >= grant {
-                    g.tokens -= grant;
-                    return;
+                let tranche = remaining.min(cap);
+                if g.tokens >= tranche {
+                    g.tokens -= tranche;
+                    remaining -= tranche;
+                    continue;
                 }
-                Duration::from_secs_f64((grant - g.tokens) / rate_f)
+                Duration::from_secs_f64((tranche - g.tokens) / rate_f)
             };
             // Cap the sleep so a live rate change takes effect promptly.
             tokio::time::sleep(wait.min(Duration::from_millis(250))).await;
@@ -138,12 +141,14 @@ impl TokenBucket {
 #[derive(Debug, Clone)]
 pub struct PauseGate {
     tx: watch::Sender<bool>,
+    // Keeps the channel open: with zero receivers `send()` silently fails.
+    _rx: watch::Receiver<bool>,
 }
 
 impl PauseGate {
     fn new() -> Self {
-        let (tx, _) = watch::channel(false);
-        Self { tx }
+        let (tx, _rx) = watch::channel(false);
+        Self { tx, _rx }
     }
     fn is_paused(&self) -> bool {
         *self.tx.borrow()
@@ -406,7 +411,7 @@ impl TransferManager {
         let pauses = self.pauses.lock().await;
         let first_open = w
             .iter()
-            .position(|x| pauses.get(x).map_or(true, |g| !g.is_paused()));
+            .position(|x| pauses.get(x).is_none_or(|g| !g.is_paused()));
         first_open == w.iter().position(|x| x == id)
     }
 
@@ -2818,3 +2823,135 @@ async fn finalize(
     }
     mgr.tasks.lock().await.remove(id);
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // ---------- TokenBucket (Phase 4) ----------
+
+    #[tokio::test(start_paused = true)]
+    async fn token_bucket_caps_throughput() {
+        let bucket = TokenBucket::new();
+        bucket.set_rate_kbps(512); // 512 KiB/s
+        let start = tokio::time::Instant::now();
+        // 1 MiB at 512 KiB/s must take ~2s of (virtual) time, charged in full.
+        bucket.acquire(1024 * 1024).await;
+        let elapsed = start.elapsed();
+        assert!(elapsed >= Duration::from_millis(1900), "too fast: {elapsed:?}");
+        assert!(elapsed <= Duration::from_millis(2600), "too slow: {elapsed:?}");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn token_bucket_unlimited_is_instant() {
+        let bucket = TokenBucket::new(); // rate 0 = unlimited
+        let start = tokio::time::Instant::now();
+        bucket.acquire(10 * 1024 * 1024).await;
+        assert_eq!(start.elapsed(), Duration::ZERO);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn token_bucket_tiny_file_skips_full_window() {
+        let bucket = TokenBucket::new();
+        bucket.set_rate_kbps(64); // 64 KiB/s
+        let start = tokio::time::Instant::now();
+        bucket.acquire(1024).await; // 1 KiB → ~16ms, not a whole window
+        assert!(start.elapsed() < Duration::from_millis(100));
+    }
+
+    // ---------- PauseGate + checkpoint (Phase 2) ----------
+
+    #[tokio::test]
+    async fn pause_gate_parks_until_opened() {
+        let gate = PauseGate::new();
+        gate.set(true);
+        let g2 = gate.clone();
+        let handle = tokio::spawn(async move { g2.wait_open().await });
+        for _ in 0..10 {
+            tokio::task::yield_now().await;
+        }
+        assert!(!handle.is_finished());
+        gate.set(false);
+        handle.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn checkpoint_parks_then_signals_restart() {
+        let mgr = Arc::new(TransferManager::new());
+        mgr.pauses.lock().await.insert("t1".into(), PauseGate::new());
+        // Not paused → passes straight through.
+        mgr.checkpoint("t1", 128).await.unwrap();
+
+        mgr.pauses.lock().await.get("t1").unwrap().set(true);
+        let m2 = Arc::clone(&mgr);
+        let handle = tokio::spawn(async move { m2.checkpoint("t1", 128).await });
+        // The parked task cannot finish before the gate opens (Ok needs an
+        // open gate, Err needs the park loop to break) — so a finished handle
+        // here is impossible regardless of scheduling.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        assert!(!handle.is_finished());
+        mgr.pauses.lock().await.get("t1").unwrap().set(false);
+        let err = handle.await.unwrap().unwrap_err();
+        assert!(err.downcast_ref::<RestartFromPause>().is_some());
+    }
+
+    // ---------- FIFO admission (Phase 1) ----------
+
+    #[tokio::test]
+    async fn fifo_skips_paused_and_pause_all_blocks_everyone() {
+        let mgr = TransferManager::new();
+        {
+            let mut w = mgr.waiting.lock().await;
+            w.push_back("a".to_string());
+            w.push_back("b".to_string());
+            w.push_back("c".to_string());
+        }
+        mgr.pauses.lock().await.insert("a".into(), PauseGate::new());
+        mgr.pauses.lock().await.insert("b".into(), PauseGate::new());
+        {
+            let w = mgr.waiting.lock().await;
+            assert!(mgr.is_my_turn(&w, "a").await);
+            assert!(!mgr.is_my_turn(&w, "b").await);
+        }
+        // A paused front row never head-of-line blocks the queue.
+        mgr.pauses.lock().await.get("a").unwrap().set(true);
+        {
+            let w = mgr.waiting.lock().await;
+            assert!(!mgr.is_my_turn(&w, "a").await);
+            assert!(mgr.is_my_turn(&w, "b").await);
+            assert!(!mgr.is_my_turn(&w, "c").await);
+        }
+        // Pause-all blocks everyone, runnable or not.
+        mgr.pause_all.set(true);
+        {
+            let w = mgr.waiting.lock().await;
+            assert!(!mgr.is_my_turn(&w, "b").await);
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn concurrency_grows_and_shrinks() {
+        let mgr = TransferManager::new();
+        assert_eq!(mgr.semaphore.available_permits(), DEFAULT_CONCURRENCY);
+        mgr.set_concurrency(5);
+        assert_eq!(mgr.semaphore.available_permits(), 5);
+        mgr.set_concurrency(2);
+        for _ in 0..10 {
+            tokio::task::yield_now().await;
+        }
+        assert_eq!(mgr.semaphore.available_permits(), 2);
+    }
+
+    // ---------- Retry classification (Phase 3) ----------
+
+    #[test]
+    fn transient_errors_retry_permanent_ones_dont() {
+        assert!(is_transient(&anyhow::anyhow!("connection reset by peer")));
+        assert!(is_transient(&anyhow::anyhow!("operation timed out")));
+        assert!(!is_transient(&anyhow::anyhow!("authentication failed")));
+        assert!(!is_transient(&anyhow::anyhow!(
+            "Permission denied (os error 13)"
+        )));
+        assert!(!is_transient(&anyhow::anyhow!("No such file or directory")));
+    }
+}
diff --git a/src/mock/core.ts b/src/mock/core.ts
index 18f2240..78fff30 100644
--- a/src/mock/core.ts
+++ b/src/mock/core.ts
@@ -2,6 +2,7 @@
 // data so the whole UI renders populated without the Rust backend. Unhandled
 // commands resolve to null, which is harmless for the screens we capture.
 import * as data from "./data";
+import * as transfers from "./transfers";
 import { emit } from "./event";
 
 type Args = Record<string, any>;
@@ -175,7 +176,7 @@ async function dispatch(cmd: string, a: Args): Promise<unknown> {
 
     // ---- transfers ----
     case "list_transfers":
-      return data.transfers;
+      return transfers.listTransfers();
     case "start_download":
     case "start_upload":
     case "start_archive":
@@ -186,6 +187,20 @@ async function dispatch(cmd: string, a: Args): Promise<unknown> {
     case "cancel_transfer":
       return null;
 
+    // ---- transfer queue (Plan 17) ----
+    case "transfer_pause":
+    case "transfer_resume":
+    case "transfer_retry":
+    case "transfer_move":
+    case "transfer_pause_all":
+    case "transfer_resume_all":
+    case "transfer_set_concurrency":
+    case "transfer_set_throttle":
+      transfers.handle(cmd, a);
+      return null;
+    case "transfer_queue_state":
+      return transfers.queueState();
+
     // ---- importers ----
     case "importer_default_paths":
       return { openssh: "~/.ssh/config", filezilla: null, putty: null };
diff --git a/src/mock/demo.ts b/src/mock/demo.ts
index 72bca07..3dcef14 100644
--- a/src/mock/demo.ts
+++ b/src/mock/demo.ts
@@ -15,6 +15,7 @@ import { useBindings } from "@/stores/bindingsStore";
 import { useSync } from "@/stores/syncStore";
 import { useUpdater } from "@/stores/updaterStore";
 import { emit } from "./event";
+import { seed as seedTransfers, calls as transferCalls } from "./transfers";
 
 (window as any).__demo = {
   useLayout,
@@ -34,4 +35,8 @@ import { emit } from "./event";
     useConnections.getState().connect(profileId),
   // Push a mock Tauri event (e.g. a bridge://approval) for screenshots.
   emit,
+  // Plan 17 test hooks: replace the mock transfer world (the store re-reads
+  // via loadInitial) and inspect the transfer commands the UI invoked.
+  seedTransfers,
+  transferCalls,
 };
diff --git a/src/mock/transfers.ts b/src/mock/transfers.ts
new file mode 100644
index 0000000..fa9b3b9
--- /dev/null
+++ b/src/mock/transfers.ts
@@ -0,0 +1,114 @@
+// Mock transfer-queue engine (Plan 17). Mirrors the Rust backend's contract
+// closely enough for the headless verify script: keeps the transfer list plus
+// the queue state (waiting FIFO / pausedAll / concurrency / throttle), applies
+// the pause/resume/retry/move/pause-all/set-* commands, and emits the same
+// `transfer://updated` / `transfer://queue` events the real backend would.
+// Every mutating command is recorded in `calls` so the harness can assert the
+// UI invoked the right command with the right args.
+import type { Transfer, TransferQueueState } from "@/lib/types";
+import { emit } from "./event";
+import { transfers as canned } from "./data";
+
+export const calls: Array<{ cmd: string; args: Record<string, any> }> = [];
+
+let byId = new Map<string, Transfer>(canned.map((t) => [t.id, t]));
+let queue: TransferQueueState = {
+  waiting: canned.filter((t) => t.status === "queued").map((t) => t.id),
+  pausedAll: false,
+  concurrency: 3,
+  throttleKbps: 0,
+};
+
+export function listTransfers(): Transfer[] {
+  return [...byId.values()];
+}
+
+export function queueState(): TransferQueueState {
+  return { ...queue, waiting: [...queue.waiting] };
+}
+
+/** Test hook: replace the whole world. The store re-reads via loadInitial(). */
+export function seed(list: Transfer[], q?: Partial<TransferQueueState>) {
+  byId = new Map(list.map((t) => [t.id, t]));
+  queue = {
+    waiting: list.filter((t) => t.status === "queued").map((t) => t.id),
+    pausedAll: false,
+    concurrency: queue.concurrency,
+    throttleKbps: queue.throttleKbps,
+    ...q,
+  };
+}
+
+function update(t: Transfer) {
+  byId.set(t.id, t);
+  emit("transfer://updated", t);
+}
+
+function emitQueue() {
+  emit("transfer://queue", queueState());
+}
+
+export function handle(cmd: string, a: Record<string, any>): void {
+  calls.push({ cmd, args: a });
+  const t = byId.get(a.transferId);
+  switch (cmd) {
+    case "transfer_pause":
+      // Valid from queued or transferring. A queued row leaves the FIFO.
+      if (t && (t.status === "transferring" || t.status === "queued")) {
+        if (queue.waiting.includes(t.id)) {
+          queue.waiting = queue.waiting.filter((id) => id !== t.id);
+          emitQueue();
+        }
+        update({ ...t, status: "paused" });
+      }
+      break;
+    case "transfer_resume":
+      // Valid from paused; the backend re-runs from byte 0.
+      if (t && t.status === "paused") {
+        update({ ...t, status: "transferring", transferred: 0 });
+      }
+      break;
+    case "transfer_retry":
+      // Valid from error/canceled; re-enqueues at the FIFO tail.
+      if (t && (t.status === "error" || t.status === "canceled")) {
+        const next: Transfer = {
+          ...t,
+          status: "queued",
+          transferred: 0,
+          error: undefined,
+          retryAttempt: undefined,
+        };
+        queue.waiting = [...queue.waiting, t.id];
+        update(next);
+        emitQueue();
+      }
+      break;
+    case "transfer_move": {
+      const i = queue.waiting.indexOf(a.transferId);
+      const j = a.direction === "up" ? i - 1 : i + 1;
+      if (i >= 0 && j >= 0 && j < queue.waiting.length) {
+        const w = [...queue.waiting];
+        [w[i], w[j]] = [w[j], w[i]];
+        queue.waiting = w;
+        emitQueue();
+      }
+      break;
+    }
+    case "transfer_pause_all":
+      queue.pausedAll = true;
+      emitQueue();
+      break;
+    case "transfer_resume_all":
+      queue.pausedAll = false;
+      emitQueue();
+      break;
+    case "transfer_set_concurrency":
+      queue.concurrency = Math.max(1, Math.min(32, a.count | 0));
+      emitQueue();
+      break;
+    case "transfer_set_throttle":
+      queue.throttleKbps = Math.max(0, a.kbps | 0);
+      emitQueue();
+      break;
+  }
+}

From 073c8155123d5ef34b573d8623c1511dadf0adde Mon Sep 17 00:00:00 2001
From: Juan Denis <jhd3197@gmail.com>
Date: Sun, 2 Aug 2026 00:38:40 -0400
Subject: [PATCH 7/7] =?UTF-8?q?docs(plans):=20roadmap=20=E2=80=94=20plan?=
 =?UTF-8?q?=2017=20built=20(all=204=20phases)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 docs/plans/ROADMAP.md | 13 +++++++++++--
 1 file changed, 11 insertions(+), 2 deletions(-)

diff --git a/docs/plans/ROADMAP.md b/docs/plans/ROADMAP.md
index 2b29d6d..73a257a 100644
--- a/docs/plans/ROADMAP.md
+++ b/docs/plans/ROADMAP.md
@@ -52,7 +52,7 @@ thematic detail.
 | 14 | `14_iconify-brand-icons` | ✅ Phases 1–3 shipped + runtime-verified (Phase 4 consolidation deferred) | Additive brand/protocol logos, bundled offline. |
 | 15 | `15_keyboard-shortcuts` | ⬜ | Remappable shortcuts + Settings Keyboard tab + file-browser keys (F2 rename & friends). Builds on the Plan 12 settings substrate. Independent polish; do whenever. |
 | 16 | `16_app-updater-and-notifications` | ⬜ | In-app auto-updater (signed, GitHub Releases manifest) + desktop notifications + one-click per-user PATH install. Trust/UX fundamentals for a shipping app. |
-| 17 | `17_transfer-queue-depth` | ⬜ | Real queue (bounded concurrency), pause/resume, retry, bandwidth throttle. Turns the transfer list into an actual queue. |
+| 17 | `17_transfer-queue-depth` | ✅ built (all 4 phases; live-backend smoke run left) | Real queue (bounded concurrency), pause/resume, retry, bandwidth throttle. Turns the transfer list into an actual queue. |
 | 18 | `18_jump-hosts-proxyjump` | ⬜ | Jump hosts (ProxyJump) through the single `ssh_connect` choke point + optional cloudflared integration. Unlocks locked-down (IP-allowlisted / tunnel-fronted) servers. Consumes Plan 13's bastion E2E fixture. |
 | 20 | `20_hubspot-backend` | ✅ shipped (Phases 1–3, mock-verified; HubDB write-back future) | HubSpot portal as a connection, one private-app token over three surfaces: Design Manager as a real remote filesystem (Source Code API v3, draft/published roots), File Manager (Files API v3), HubDB tables as virtual CSV files (read-only). The Shopify recipe (`18_shopify-backend.md`, shipped), an even better fit. |
 | 21 | `21_dynamics-365-backend` | 🔄 Phase 1 shipped + mock-verified (client-credentials only; delegated OAuth + Phases 2–3 planned) | Dynamics 365/Dataverse environment as a connection: web resources are literally files in a table (`webresourceset` OData — path-like names, base64 content, publish-to-deploy). Client-credentials or delegated Entra auth (reuses `oauth.rs`). Phase 2: tables as virtual CSV + `faro-cli dynamics query` (the wp-cli-style db helper). The XrmToolBox gap-filler. |
@@ -415,7 +415,16 @@ concurrency (semaphore + FIFO, reorderable), per-transfer pause/resume
 (park at a chunk boundary, resume re-runs the file), manual + classified
 auto-retry (Plan 12 error kinds), and a global token-bucket bandwidth throttle.
 All checkpoints live in the *shared* copy loops so the 11 backends get it for
-free. Panel gains pause/retry row actions, pause-all, and a throttle input. ⬜
+free. Panel gains pause/retry row actions, pause-all, and a throttle input. ✅
+**Built** — all four phases: FIFO + semaphore admission (`transferConcurrency`,
+live-adjustable), per-transfer `PauseGate`s with chunk checkpoints in every
+copy loop (resume re-runs from byte 0), manual + auto retry (Network/Timeout,
+5s/20s backoff, classified via `classify_message`), global `TokenBucket`
+(`transferThrottleKbps`, live). Panel: pause-all, throttle input, row
+pause/resume/retry/↑↓ with queue positions; Settings → Transfers for the two
+defaults. Unit-tested (bucket pacing, FIFO skip-paused, checkpoint restart,
+retry classification); `scripts/verify-transfers.mjs` drives the panel
+headlessly. Remaining: a live-backend smoke run (real SFTP/S3 batch).
 
 ## Track R — Jump hosts & Zero-Trust connectivity (Plan 18)
 Locked-down servers (IP allowlists, Cloudflare Tunnels) are unreachable to