Add audio recording and upload capabilities

Introduces new modules for audio recording (`recorder.rs`) and asynchronous uploading (`uploader.rs`) with retry logic. The `recorder` module uses `ffmpeg` as a subprocess to capture audio and save it as M4A files. It handles starting, stopping, and reporting recording events. The `uploader` module manages a queue of recordings to be uploaded to the server. It supports persistent storage of pending uploads via sidecar JSON files, exponential backoff for retries, and emits events for UI feedback. New dependencies were added to `Cargo.toml` and `Cargo.lock` to support these features, including `reqwest`, `serde_json`, `tokio`, `uuid`, and `which`.
2026-04-17 15:49:55 +02:00
parent c441377749
commit 5f7e46256c
6 changed files with 693 additions and 0 deletions
@@ -1162,12 +1162,18 @@ dependencies = [
 "directories",
 "doctate-common",
 "eframe",
+ "reqwest",
 "serde",
+ "serde_json",
 "tempfile",
 "thiserror 1.0.69",
+ "tokio",
 "toml",
 "tracing",
 "tracing-subscriber",
+ "uuid",
+ "which",
+ "wiremock",
 ]

 [[package]]
@@ -1336,6 +1342,12 @@ dependencies = [
 "winit",
 ]

+[[package]]
+name = "either"
+version = "1.15.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "48c757948c5ede0e46177b7add2e67155f70e33c07fea8284df6576da70b3719"
+
 [[package]]
 name = "emath"
 version = "0.28.1"
@@ -1940,6 +1952,15 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "dfa686283ad6dd069f105e5ab091b04c62850d3e4cf5d67debad1933f55023df"

+[[package]]
+name = "home"
+version = "0.5.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc627f471c528ff0c4a49e1d5e60450c8f6461dd6d10ba9dcd3a61d3dff7728d"
+dependencies = [
+ "windows-sys 0.61.2",
+]
+
 [[package]]
 name = "http"
 version = "1.4.0"
@@ -5035,6 +5056,18 @@ dependencies = [
 "web-sys",
 ]

+[[package]]
+name = "which"
+version = "6.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b4ee928febd44d98f2f459a4a79bd4d928591333a494a10a868418ac1b39cf1f"
+dependencies = [
+ "either",
+ "home",
+ "rustix 0.38.44",
+ "winsafe",
+]
+
 [[package]]
 name = "widestring"
 version = "1.2.1"
@@ -5503,6 +5536,12 @@ dependencies = [
 "memchr",
 ]

+[[package]]
+name = "winsafe"
+version = "0.0.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d135d17ab770252ad95e9a872d365cf3090e3be864a34ab46f48555993efc904"
+
 [[package]]
 name = "wiremock"
 version = "0.6.5"
@@ -7,12 +7,19 @@ description = "Desktop dictation client for the doctate medical dictation system
 [dependencies]
 doctate-common = { path = "../doctate-common" }
 eframe = "0.28"
+reqwest = { workspace = true }
 serde = { workspace = true }
+serde_json = { workspace = true }
 toml = { workspace = true }
+tokio = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+uuid = { workspace = true }
 directories = "5"
 thiserror = "1"
+which = "6"

 [dev-dependencies]
 tempfile = "3"
+tokio = { workspace = true, features = ["test-util"] }
+wiremock = "0.6"
@@ -1,5 +1,11 @@
 #[allow(dead_code)]
 mod config;
+#[allow(dead_code)]
+mod paths;
+#[allow(dead_code)]
+mod recorder;
+#[allow(dead_code)]
+mod uploader;

 use eframe::egui;
 use tracing::info;
@@ -0,0 +1,23 @@
+//! OS-conventional paths for client-side storage.
+
+use std::path::PathBuf;
+
+use directories::ProjectDirs;
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum PathError {
+    #[error("cannot determine OS data directory (is $HOME set?)")]
+    NoDataDir,
+}
+
+/// Directory where recordings wait for upload. Persists across app
+/// restarts and crashes.
+///   Linux:   `~/.local/share/doctate/pending/`
+///   Windows: `%LOCALAPPDATA%\doctate\pending\`
+///   macOS:   `~/Library/Application Support/doctate/pending/`
+pub fn pending_dir() -> Result<PathBuf, PathError> {
+    ProjectDirs::from("", "", "doctate")
+        .map(|dirs| dirs.data_local_dir().join("pending"))
+        .ok_or(PathError::NoDataDir)
+}
@@ -0,0 +1,201 @@
+//! Audio recorder via an ffmpeg subprocess. Produces m4a/AAC files on disk.
+//!
+//! Portability note: this module is intentionally UI-agnostic and carries no
+//! `egui`/`eframe` deps so it can be lifted into a shared client-core crate
+//! once a second Rust client (Windows variant, etc.) arrives.
+//!
+//! Call `Recorder::start` inside a Tokio runtime context (entered with
+//! `Runtime::enter()` or via `#[tokio::main]`).
+
+use std::path::{Path, PathBuf};
+use std::process::Stdio;
+use std::time::Duration;
+
+use thiserror::Error;
+use tokio::io::AsyncWriteExt;
+use tokio::process::{Child, Command};
+use tokio::sync::{mpsc, oneshot};
+use tokio::time::timeout;
+use tracing::warn;
+
+/// Max time to wait for ffmpeg to finalize the MOOV atom after stop; past
+/// this we hard-kill and emit `Failed`.
+const STOP_GRACEFUL_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[derive(Debug)]
+pub enum RecorderEvent {
+    /// ffmpeg spawned successfully; audio capture has begun.
+    Started,
+    /// ffmpeg exited cleanly; the m4a file is ready.
+    Finished { output: PathBuf },
+    /// ffmpeg or I/O failed; output file may be missing or corrupt.
+    Failed { error: String },
+}
+
+#[derive(Debug, Error)]
+pub enum RecorderError {
+    #[error("ffmpeg not found in PATH")]
+    FfmpegMissing,
+    #[error("spawn ffmpeg failed: {0}")]
+    Spawn(#[from] std::io::Error),
+}
+
+/// Handle to a running recorder. Dropping it does NOT stop the recorder —
+/// call `stop()` explicitly (consumes self).
+pub struct Recorder {
+    stop_tx: oneshot::Sender<()>,
+}
+
+impl Recorder {
+    /// Spawn a recorder task that captures the system default audio input
+    /// to `output_path`. Returns a handle + event receiver. The task
+    /// continues until `stop()` is called, then finalizes the m4a.
+    pub fn start(
+        output_path: PathBuf,
+    ) -> Result<(Self, mpsc::UnboundedReceiver<RecorderEvent>), RecorderError> {
+        if which::which("ffmpeg").is_err() {
+            return Err(RecorderError::FfmpegMissing);
+        }
+
+        let (events_tx, events_rx) = mpsc::unbounded_channel();
+        let (stop_tx, stop_rx) = oneshot::channel();
+
+        let mut cmd = Command::new("ffmpeg");
+        cmd.args(ffmpeg_args(&output_path))
+            .stdin(Stdio::piped())
+            .stdout(Stdio::null())
+            .stderr(Stdio::null());
+
+        let child = cmd.spawn()?;
+
+        tokio::spawn(run_recorder_task(child, output_path, stop_rx, events_tx));
+
+        Ok((Self { stop_tx }, events_rx))
+    }
+
+    /// Request the recorder to finalize and exit. Consumes the handle —
+    /// no double-stop. Completion arrives as `Finished` (or `Failed`) on
+    /// the event receiver.
+    pub fn stop(self) {
+        let _ = self.stop_tx.send(());
+    }
+}
+
+async fn run_recorder_task(
+    mut child: Child,
+    output_path: PathBuf,
+    stop_rx: oneshot::Receiver<()>,
+    events_tx: mpsc::UnboundedSender<RecorderEvent>,
+) {
+    let _ = events_tx.send(RecorderEvent::Started);
+
+    tokio::select! {
+        exit_result = child.wait() => {
+            let event = match exit_result {
+                Ok(status) if status.success() => RecorderEvent::Finished { output: output_path },
+                Ok(status) => RecorderEvent::Failed {
+                    error: format!("ffmpeg exited unexpectedly: {status}"),
+                },
+                Err(e) => RecorderEvent::Failed {
+                    error: format!("wait ffmpeg: {e}"),
+                },
+            };
+            let _ = events_tx.send(event);
+        }
+        _ = stop_rx => {
+            finalize(&mut child, output_path, &events_tx).await;
+        }
+    }
+}
+
+/// Send 'q' to ffmpeg's stdin, wait up to `STOP_GRACEFUL_TIMEOUT` for a
+/// clean exit; hard-kill on timeout.
+async fn finalize(
+    child: &mut Child,
+    output_path: PathBuf,
+    events_tx: &mpsc::UnboundedSender<RecorderEvent>,
+) {
+    if let Some(stdin) = child.stdin.as_mut()
+        && let Err(e) = stdin.write_all(b"q\n").await
+    {
+        warn!(error = %e, "failed to write stop to ffmpeg stdin");
+    }
+
+    let event = match timeout(STOP_GRACEFUL_TIMEOUT, child.wait()).await {
+        Ok(Ok(status)) if status.success() => RecorderEvent::Finished { output: output_path },
+        Ok(Ok(status)) => RecorderEvent::Failed {
+            error: format!("ffmpeg exited with {status}"),
+        },
+        Ok(Err(e)) => RecorderEvent::Failed {
+            error: format!("wait ffmpeg: {e}"),
+        },
+        Err(_elapsed) => {
+            warn!("ffmpeg did not exit within timeout; killing");
+            let _ = child.kill().await;
+            RecorderEvent::Failed {
+                error: "ffmpeg stop timeout — output may be corrupt".into(),
+            }
+        }
+    };
+    let _ = events_tx.send(event);
+}
+
+fn ffmpeg_args(output: &Path) -> Vec<String> {
+    let mut args: Vec<String> = vec!["-loglevel".into(), "error".into()];
+    for s in audio_input_args() {
+        args.push(s.to_string());
+    }
+    args.extend([
+        "-c:a".into(),
+        "aac".into(),
+        "-b:a".into(),
+        "96k".into(),
+        "-ar".into(),
+        "48000".into(),
+        "-ac".into(),
+        "1".into(),
+        "-movflags".into(),
+        "+faststart".into(),
+        "-y".into(),
+        output.to_string_lossy().into_owned(),
+    ]);
+    args
+}
+
+#[cfg(target_os = "linux")]
+fn audio_input_args() -> Vec<&'static str> {
+    // PulseAudio interface (also works under PipeWire via pipewire-pulse).
+    vec!["-f", "pulse", "-i", "default"]
+}
+
+#[cfg(target_os = "windows")]
+fn audio_input_args() -> Vec<&'static str> {
+    // TODO(windows-client): enumerate dshow devices via
+    // `ffmpeg -list_devices true -f dshow -i dummy` and cache the first
+    // non-loopback capture device. "audio=default" below is a placeholder;
+    // most Windows systems do not have a device literally named "default",
+    // so the Windows client must resolve a real name before shipping.
+    vec!["-f", "dshow", "-i", "audio=default"]
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn ffmpeg_args_contain_output_path_and_encoder() {
+        let path = Path::new("/tmp/test.m4a");
+        let args = ffmpeg_args(path);
+        assert!(args.contains(&"/tmp/test.m4a".to_string()));
+        assert!(args.contains(&"-y".to_string()));
+        assert!(args.contains(&"aac".to_string()));
+        assert!(args.contains(&"+faststart".to_string()));
+    }
+
+    #[test]
+    fn ffmpeg_args_have_exactly_one_input() {
+        let args = ffmpeg_args(Path::new("/tmp/x.m4a"));
+        let input_flags = args.iter().filter(|s| *s == "-i").count();
+        assert_eq!(input_flags, 1);
+    }
+}
@@ -0,0 +1,417 @@
+//! Async uploader with persistent on-disk queue and exponential-backoff
+//! retry. Recordings live in the pending directory as `{stem}.m4a` plus
+//! a sidecar `{stem}.meta.json`. The worker replays anything it finds
+//! there at startup, then processes new submissions as they arrive.
+
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+use doctate_common::ack::{AckResponse, AckStatus};
+use doctate_common::constants::{
+    API_KEY_HEADER, CONTENT_TYPE_AUDIO_MP4, FIELD_AUDIO, FIELD_CASE_ID, FIELD_RECORDED_AT,
+    UPLOAD_PATH,
+};
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio::sync::mpsc;
+use tracing::{info, warn};
+use uuid::Uuid;
+
+use crate::config::Config;
+
+const INITIAL_BACKOFF: Duration = Duration::from_secs(2);
+const MAX_BACKOFF: Duration = Duration::from_secs(60);
+
+/// A recording waiting to be uploaded. The `file` path is derived from
+/// the sidecar JSON's location, never from the JSON content itself —
+/// that way the sidecar stays valid even if the pending directory is
+/// renamed.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PendingUpload {
+    pub case_id: Uuid,
+    pub recorded_at: String,
+    #[serde(skip, default)]
+    pub file: PathBuf,
+}
+
+/// Lifecycle events emitted to the UI for a single upload attempt chain.
+#[derive(Debug)]
+pub enum UploadEvent {
+    Started(Uuid),
+    Succeeded {
+        case_id: Uuid,
+        status: AckStatus,
+    },
+    Failed {
+        case_id: Uuid,
+        reason: String,
+        will_retry: bool,
+    },
+}
+
+#[derive(Debug, Error)]
+pub enum UploaderError {
+    #[error("I/O error: {0}")]
+    Io(#[from] std::io::Error),
+    #[error("serialize sidecar: {0}")]
+    Serialize(#[from] serde_json::Error),
+}
+
+/// Handle to the uploader's worker task. Drop does NOT shut down the
+/// worker; it stays alive as long as the `submit_tx` is kept.
+pub struct Uploader {
+    submit_tx: mpsc::UnboundedSender<PendingUpload>,
+}
+
+impl Uploader {
+    /// Spawn the worker task. Must be called inside a Tokio runtime
+    /// context (`Runtime::enter()` guard or `#[tokio::main]`).
+    ///
+    /// Re-scans `pending_dir` at startup so leftover recordings from
+    /// a prior run are replayed automatically.
+    pub fn spawn(
+        config: Config,
+        pending_dir: PathBuf,
+    ) -> (Self, mpsc::UnboundedReceiver<UploadEvent>) {
+        let (submit_tx, submit_rx) = mpsc::unbounded_channel();
+        let (events_tx, events_rx) = mpsc::unbounded_channel();
+        tokio::spawn(run_worker(config, pending_dir, submit_rx, events_tx));
+        (Self { submit_tx }, events_rx)
+    }
+
+    /// Submit a pending upload for processing. Non-blocking. Fails only
+    /// if the worker has been dropped (should not happen in normal
+    /// operation).
+    pub fn submit(
+        &self,
+        upload: PendingUpload,
+    ) -> Result<(), mpsc::error::SendError<PendingUpload>> {
+        self.submit_tx.send(upload)
+    }
+}
+
+/// Path of the sidecar JSON for a given m4a file — same directory,
+/// `{stem}.meta.json`.
+pub fn sidecar_path_for(m4a: &Path) -> PathBuf {
+    let stem = m4a
+        .file_stem()
+        .map(|s| s.to_string_lossy().into_owned())
+        .unwrap_or_default();
+    let dir = m4a.parent().unwrap_or_else(|| Path::new("."));
+    dir.join(format!("{stem}.meta.json"))
+}
+
+/// Persist a pending upload's metadata next to its m4a file.
+pub fn write_sidecar(upload: &PendingUpload) -> Result<(), UploaderError> {
+    let sidecar = sidecar_path_for(&upload.file);
+    if let Some(parent) = sidecar.parent() {
+        std::fs::create_dir_all(parent)?;
+    }
+    let json = serde_json::to_string_pretty(upload)?;
+    std::fs::write(&sidecar, json)?;
+    Ok(())
+}
+
+/// Scan `pending_dir` for m4a files with matching sidecars. Orphans
+/// (m4a without sidecar, or unreadable sidecar) are logged and skipped
+/// rather than propagated as errors — one broken file should not take
+/// down recovery of the rest.
+pub fn scan_pending_dir(pending_dir: &Path) -> Vec<PendingUpload> {
+    let entries = match std::fs::read_dir(pending_dir) {
+        Ok(e) => e,
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Vec::new(),
+        Err(e) => {
+            warn!(dir = ?pending_dir, error = %e, "scan pending dir failed");
+            return Vec::new();
+        }
+    };
+    let mut out = Vec::new();
+    for entry in entries.flatten() {
+        let path = entry.path();
+        if path.extension().and_then(|s| s.to_str()) != Some("m4a") {
+            continue;
+        }
+        match read_pending(&path) {
+            Ok(upload) => out.push(upload),
+            Err(e) => warn!(path = ?path, error = %e, "skipping pending entry"),
+        }
+    }
+    out
+}
+
+fn read_pending(m4a: &Path) -> Result<PendingUpload, UploaderError> {
+    let sidecar = sidecar_path_for(m4a);
+    let json = std::fs::read_to_string(&sidecar)?;
+    let mut upload: PendingUpload = serde_json::from_str(&json)?;
+    upload.file = m4a.to_path_buf();
+    Ok(upload)
+}
+
+async fn run_worker(
+    config: Config,
+    pending_dir: PathBuf,
+    mut submit_rx: mpsc::UnboundedReceiver<PendingUpload>,
+    events_tx: mpsc::UnboundedSender<UploadEvent>,
+) {
+    let client = reqwest::Client::new();
+
+    let startup = scan_pending_dir(&pending_dir);
+    if !startup.is_empty() {
+        info!(count = startup.len(), "replaying pending uploads");
+    }
+    for pending in startup {
+        upload_with_retry(&client, &config, pending, &events_tx).await;
+    }
+
+    while let Some(pending) = submit_rx.recv().await {
+        upload_with_retry(&client, &config, pending, &events_tx).await;
+    }
+}
+
+async fn upload_with_retry(
+    client: &reqwest::Client,
+    config: &Config,
+    upload: PendingUpload,
+    events_tx: &mpsc::UnboundedSender<UploadEvent>,
+) {
+    let mut backoff = INITIAL_BACKOFF;
+    loop {
+        let _ = events_tx.send(UploadEvent::Started(upload.case_id));
+        match attempt_upload(client, config, &upload).await {
+            Ok(ack) => {
+                let _ = tokio::fs::remove_file(&upload.file).await;
+                let _ = tokio::fs::remove_file(&sidecar_path_for(&upload.file)).await;
+                let _ = events_tx.send(UploadEvent::Succeeded {
+                    case_id: upload.case_id,
+                    status: ack.status,
+                });
+                return;
+            }
+            Err(ErrorKind::Terminal(reason)) => {
+                let _ = events_tx.send(UploadEvent::Failed {
+                    case_id: upload.case_id,
+                    reason,
+                    will_retry: false,
+                });
+                return;
+            }
+            Err(ErrorKind::Transient(reason)) => {
+                let _ = events_tx.send(UploadEvent::Failed {
+                    case_id: upload.case_id,
+                    reason,
+                    will_retry: true,
+                });
+                tokio::time::sleep(backoff).await;
+                backoff = (backoff * 2).min(MAX_BACKOFF);
+            }
+        }
+    }
+}
+
+enum ErrorKind {
+    Transient(String),
+    Terminal(String),
+}
+
+async fn attempt_upload(
+    client: &reqwest::Client,
+    config: &Config,
+    upload: &PendingUpload,
+) -> Result<AckResponse, ErrorKind> {
+    let audio = tokio::fs::read(&upload.file)
+        .await
+        .map_err(|e| ErrorKind::Terminal(format!("read audio file: {e}")))?;
+
+    let file_name = upload
+        .file
+        .file_name()
+        .map(|s| s.to_string_lossy().into_owned())
+        .unwrap_or_else(|| "recording.m4a".to_string());
+
+    let audio_part = reqwest::multipart::Part::bytes(audio)
+        .file_name(file_name)
+        .mime_str(CONTENT_TYPE_AUDIO_MP4)
+        .expect("static MIME string is valid");
+
+    let form = reqwest::multipart::Form::new()
+        .text(FIELD_CASE_ID, upload.case_id.to_string())
+        .text(FIELD_RECORDED_AT, upload.recorded_at.clone())
+        .part(FIELD_AUDIO, audio_part);
+
+    let url = format!("{}{}", config.server_url.trim_end_matches('/'), UPLOAD_PATH);
+
+    let resp = client
+        .post(&url)
+        .header(API_KEY_HEADER, &config.api_key)
+        .multipart(form)
+        .send()
+        .await
+        .map_err(|e| ErrorKind::Transient(format!("network: {e}")))?;
+
+    let status = resp.status();
+    if status.is_success() {
+        resp.json::<AckResponse>()
+            .await
+            .map_err(|e| ErrorKind::Terminal(format!("parse ack: {e}")))
+    } else if status.as_u16() == 408 || status.as_u16() == 429 || status.is_server_error() {
+        let body = resp.text().await.unwrap_or_default();
+        Err(ErrorKind::Transient(format!("HTTP {status}: {body}")))
+    } else {
+        let body = resp.text().await.unwrap_or_default();
+        Err(ErrorKind::Terminal(format!("HTTP {status}: {body}")))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+    use wiremock::matchers::{header, method, path as wpath};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
+
+    fn sample_pending(dir: &Path) -> PendingUpload {
+        let file = dir.join("sample.m4a");
+        std::fs::write(&file, b"dummy audio").unwrap();
+        PendingUpload {
+            case_id: Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap(),
+            recorded_at: "2026-04-15T10:32:00Z".into(),
+            file,
+        }
+    }
+
+    fn ack_body(status: AckStatus) -> AckResponse {
+        AckResponse {
+            case_id: "550e8400-e29b-41d4-a716-446655440000".into(),
+            recorded_at: "2026-04-15T10:32:00Z".into(),
+            status,
+        }
+    }
+
+    #[test]
+    fn sidecar_roundtrip_via_scan() {
+        let tmp = TempDir::new().unwrap();
+        let upload = sample_pending(tmp.path());
+        write_sidecar(&upload).unwrap();
+        let found = scan_pending_dir(tmp.path());
+        assert_eq!(found.len(), 1);
+        assert_eq!(found[0].case_id, upload.case_id);
+        assert_eq!(found[0].recorded_at, upload.recorded_at);
+        assert_eq!(found[0].file, upload.file);
+    }
+
+    #[test]
+    fn scan_ignores_m4a_without_sidecar() {
+        let tmp = TempDir::new().unwrap();
+        std::fs::write(tmp.path().join("orphan.m4a"), b"x").unwrap();
+        let found = scan_pending_dir(tmp.path());
+        assert!(found.is_empty());
+    }
+
+    #[test]
+    fn scan_returns_empty_for_missing_dir() {
+        let tmp = TempDir::new().unwrap();
+        let missing = tmp.path().join("does-not-exist");
+        assert!(scan_pending_dir(&missing).is_empty());
+    }
+
+    #[tokio::test]
+    async fn successful_upload_deletes_files_and_emits_events() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(wpath(UPLOAD_PATH))
+            .and(header(API_KEY_HEADER, "test-key"))
+            .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received)))
+            .mount(&server)
+            .await;
+
+        let tmp = TempDir::new().unwrap();
+        let upload = sample_pending(tmp.path());
+        write_sidecar(&upload).unwrap();
+        let config = Config {
+            server_url: server.uri(),
+            api_key: "test-key".into(),
+        };
+
+        let client = reqwest::Client::new();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        upload_with_retry(&client, &config, upload.clone(), &tx).await;
+
+        assert!(!upload.file.exists(), "m4a should be deleted after ack");
+        assert!(
+            !sidecar_path_for(&upload.file).exists(),
+            "sidecar should be deleted after ack"
+        );
+
+        assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_))));
+        let Some(UploadEvent::Succeeded { status, .. }) = rx.recv().await else {
+            panic!("expected Succeeded event");
+        };
+        assert_eq!(status, AckStatus::Received);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn transient_500_retries_then_succeeds() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(wpath(UPLOAD_PATH))
+            .respond_with(ResponseTemplate::new(500))
+            .up_to_n_times(1)
+            .mount(&server)
+            .await;
+        Mock::given(method("POST"))
+            .and(wpath(UPLOAD_PATH))
+            .respond_with(ResponseTemplate::new(200).set_body_json(ack_body(AckStatus::Received)))
+            .mount(&server)
+            .await;
+
+        let tmp = TempDir::new().unwrap();
+        let upload = sample_pending(tmp.path());
+        write_sidecar(&upload).unwrap();
+        let config = Config {
+            server_url: server.uri(),
+            api_key: "test-key".into(),
+        };
+
+        let client = reqwest::Client::new();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        upload_with_retry(&client, &config, upload.clone(), &tx).await;
+
+        assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_))));
+        let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else {
+            panic!("expected Failed event");
+        };
+        assert!(will_retry, "500 should be a transient failure");
+        assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_))));
+        assert!(matches!(rx.recv().await, Some(UploadEvent::Succeeded { .. })));
+    }
+
+    #[tokio::test]
+    async fn terminal_401_keeps_files_and_does_not_retry() {
+        let server = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(wpath(UPLOAD_PATH))
+            .respond_with(ResponseTemplate::new(401).set_body_string("Unauthorized"))
+            .mount(&server)
+            .await;
+
+        let tmp = TempDir::new().unwrap();
+        let upload = sample_pending(tmp.path());
+        write_sidecar(&upload).unwrap();
+        let config = Config {
+            server_url: server.uri(),
+            api_key: "bad-key".into(),
+        };
+
+        let client = reqwest::Client::new();
+        let (tx, mut rx) = mpsc::unbounded_channel();
+        upload_with_retry(&client, &config, upload.clone(), &tx).await;
+
+        assert!(upload.file.exists(), "m4a must be preserved on terminal failure");
+
+        assert!(matches!(rx.recv().await, Some(UploadEvent::Started(_))));
+        let Some(UploadEvent::Failed { will_retry, .. }) = rx.recv().await else {
+            panic!("expected Failed event");
+        };
+        assert!(!will_retry, "401 is terminal; must not retry");
+    }
+}