Skip to main content

suno_core/
executor.rs

1//! The download executor: it applies a reconcile [`Plan`] to disk through ports.
2//!
3//! Reconcile decides *what* to do; the executor does it. It is async and pure
4//! orchestration: every side effect goes through a port ([`Http`] for the
5//! network, [`Filesystem`] for disk, [`Ffmpeg`] for transcoding, [`Clock`] for
6//! waiting), so the whole pipeline is exercised in tests with in-memory doubles
7//! and no real IO, network, or sleeping.
8//!
9//! Safety is the point of this module. A wrong write or delete damages the
10//! user's library, so the executor:
11//!
12//! - writes only atomically (SYNC-13): a failed write leaves the prior file
13//!   intact, because the [`Filesystem`] adapter stages a temp file and renames;
14//! - verifies size (SYNC-14): a download whose body disagrees with the
15//!   provider's `Content-Length` is treated as truncated and retried, and a
16//!   written file whose on-disk size disagrees with the bytes written is a
17//!   failure, never a recorded success;
18//! - classifies errors (SYNC-17): an auth failure or a full disk stops the
19//!   account run (with an auth or disk-full status) and is never retried;
20//!   transient failures (timeouts, 5xx,
21//!   transport, 429) are retried a bounded number of times then recorded and
22//!   skipped; permanent failures are recorded and skipped; and a single clip's
23//!   failure never aborts the run;
24//! - backs off on rate limits (SYNC-16) through the injected [`Clock`], honouring
25//!   a `Retry-After` hint.
26//!
27//! The executor only ever sets the manifest's [`preserve`](ManifestEntry::preserve)
28//! marker on an entry it writes, and only deletes a path whose removal the
29//! [`Filesystem`] confirms. Higher-level safety (empty-listing abort, the
30//! destructive-sync confirmation, exit codes) is the caller's job.
31
32use std::collections::BTreeMap;
33use std::collections::BTreeSet;
34use std::collections::HashMap;
35use std::collections::HashSet;
36use std::sync::Mutex;
37use std::time::Duration;
38
39use futures_util::lock::Mutex as AsyncMutex;
40use futures_util::stream::{self, StreamExt};
41
42use crate::album_art::{AlbumArt, PlaylistState, set_album_artifact, set_playlist};
43use crate::backoff::{backoff_delay, retry_after};
44use crate::client::SunoClient;
45use crate::clock::Clock;
46use crate::config::{AudioFormat, StemFormat};
47use crate::error::Error;
48use crate::ffmpeg::{Ffmpeg, WebpEncodeSettings};
49use crate::fs::Filesystem;
50use crate::http::{Http, HttpRequest};
51use crate::lineage::LineageContext;
52use crate::lyrics::AlignedLyrics;
53use crate::manifest::{ArtifactState, Manifest, ManifestEntry};
54use crate::model::Clip;
55use crate::reconcile::{
56    Action, ArtifactKind, Desired, Plan, SourceMode, set_manifest_artifact, set_manifest_stem,
57};
58use crate::tag::{Cover, TrackMetadata, flac_picture_data_budget, tag_flac, tag_mp3, tag_wav};
59use crate::tag_alac::tag_alac;
60
61/// The shared Suno client behind an async mutex, so concurrent audio work can
62/// serialise its order-sensitive API calls (JWT refresh, adaptive limiter)
63/// without a runtime-specific lock. Held only for the brief WAV-render calls;
64/// the heavy CDN/transcode/tag work runs unlocked.
65type ClientLock<'a, C> = AsyncMutex<&'a SunoClient<C>>;
66
67/// Tunables for one [`execute`] run.
68#[derive(Debug, Clone)]
69pub struct ExecOptions {
70    /// How many times a transient failure is retried before record-and-skip.
71    pub max_retries: u32,
72    /// How many times to poll for a server-side WAV render before giving up.
73    pub wav_poll_attempts: u32,
74    /// How long to wait between WAV render polls.
75    pub wav_poll_interval: Duration,
76    /// How many clips' audio to fetch, transcode, and tag concurrently. Clamped
77    /// to at least one, so a zero collapses to sequential rather than stalling.
78    pub concurrency: u32,
79    /// Embed a bounded animated WebP as the audio file's front cover (in place of
80    /// the static JPEG) for clips that carry a video preview. Off leaves the
81    /// static JPEG embed unchanged.
82    pub embed_animated_cover: bool,
83    /// Settings used for animated WebP cover transcodes.
84    pub cover_webp: WebpEncodeSettings,
85}
86
87impl Default for ExecOptions {
88    fn default() -> Self {
89        Self {
90            max_retries: 3,
91            wav_poll_attempts: 24,
92            wav_poll_interval: Duration::from_secs(5),
93            concurrency: 4,
94            embed_animated_cover: false,
95            cover_webp: WebpEncodeSettings::default(),
96        }
97    }
98}
99
100/// How an [`execute`] run ended.
101#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
102pub enum RunStatus {
103    /// Every action was attempted; some may have failed and been skipped.
104    #[default]
105    Completed,
106    /// An auth failure stopped the run early; remaining actions were not tried.
107    AuthAborted,
108    /// The disk filled; the run stopped early rather than failing every
109    /// remaining clip. Remaining actions were not tried.
110    DiskFull,
111}
112
113/// One action that could not be applied, for the run summary and failure log.
114#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct Failure {
116    /// The clip the failed action concerned (or a path when no id applies).
117    pub clip_id: String,
118    /// A short, secret-free reason.
119    pub reason: String,
120}
121
122/// The result of applying a [`Plan`]: per-action counts and the failure list.
123#[derive(Debug, Clone, Default, PartialEq, Eq)]
124pub struct ExecOutcome {
125    pub downloaded: usize,
126    pub reformatted: usize,
127    pub retagged: usize,
128    pub renamed: usize,
129    pub deleted: usize,
130    pub skipped: usize,
131    pub artifacts_written: usize,
132    pub artifacts_deleted: usize,
133    /// Actions that failed and were skipped (auth, transient-exhausted, or
134    /// permanent). The run continued past each one unless it was an auth or
135    /// disk-full abort.
136    pub failures: Vec<Failure>,
137    /// How the run ended.
138    pub status: RunStatus,
139}
140
141impl ExecOutcome {
142    /// Number of failed actions.
143    pub fn failed(&self) -> usize {
144        self.failures.len()
145    }
146
147    fn record(&mut self, effect: Effect) {
148        match effect {
149            Effect::Downloaded => self.downloaded += 1,
150            Effect::Reformatted => self.reformatted += 1,
151            Effect::Retagged => self.retagged += 1,
152            Effect::Renamed => self.renamed += 1,
153            Effect::Deleted => self.deleted += 1,
154            Effect::Skipped => self.skipped += 1,
155            Effect::ArtifactWritten => self.artifacts_written += 1,
156            Effect::ArtifactDeleted => self.artifacts_deleted += 1,
157        }
158    }
159}
160
161/// The IO ports the executor drives, grouped so one value threads them through.
162///
163/// `client` performs the authenticated WAV render flow. The rest are shared
164/// references.
165pub struct Ports<'a, H, F, G, C> {
166    /// Performs the authenticated WAV render and poll flow.
167    pub client: &'a SunoClient<C>,
168    /// The public network port (CDN audio, rendered WAV, cover art).
169    pub http: &'a H,
170    /// The disk port.
171    pub fs: &'a F,
172    /// The transcode port (WAV to FLAC).
173    pub ffmpeg: &'a G,
174    /// The backoff and poll delay port.
175    pub clock: &'a C,
176}
177
178/// Apply `plan` to disk, updating `manifest` and `albums` in place, and return
179/// the outcome.
180///
181/// `desired` carries the per-clip metadata and art hashes plus the source modes
182/// that decide the [`preserve`](ManifestEntry::preserve) marker; it is indexed
183/// by clip id (and by target path, for renames) so each written entry records
184/// the right hashes and protection. `albums` is the album-art store, keyed by
185/// stable root id: folder-art writes and deletes record their state there rather
186/// than on the per-clip `manifest`. `ports` bundles the authenticated client and
187/// the network, disk, transcode, and backoff ports. A single clip's failure
188/// never aborts the run, except an auth failure or a full disk, which stop it
189/// with [`RunStatus::AuthAborted`] or [`RunStatus::DiskFull`].
190///
191/// Audio-producing ([`Download`](Action::Download) /
192/// [`Reformat`](Action::Reformat)), fetched-artifact
193/// ([`WriteArtifact`](Action::WriteArtifact) with no inline content), and stem
194/// ([`WriteStem`](Action::WriteStem)) actions all run their slow,
195/// side-effect-free work concurrently, bounded by
196/// [`ExecOptions::concurrency`]: WAV render + CDN download + transcode + tag for
197/// audio; CDN fetch + optional WebP transcode for artifacts; WAV render + CDN
198/// download for stems. Order-sensitive Suno API calls (WAV render initiation and
199/// poll) are serialised behind an async mutex over the shared [`SunoClient`],
200/// keeping the adaptive limiter and JWT refresh correct. The remaining actions
201/// (retag, rename, delete, artifact deletes, and inline artifact writes) run
202/// serially in plan order.
203///
204/// The outcome is deterministic regardless of completion order: all prepared
205/// results are committed to the manifest in plan-index order, so the same plan
206/// always yields the same manifest and counts whatever the concurrency level. A
207/// per-clip failure is recorded and the run continues; only an auth failure or a
208/// full disk aborts, and it does so promptly by stopping further concurrent work.
209///
210/// `synced` carries this run's fetched aligned (synced) lyrics keyed by clip id;
211/// it is the caller's IO result, not part of the pure plan. Audio tagging embeds
212/// a clip's entry as an MP3 `SYLT` frame and as the plain `USLT`/`LYRICS` text
213/// (FLAC), so a clip absent from the map (an instrumental, a WAV target, or a
214/// run with the feature off) is tagged exactly as before. The synced `.lrc`
215/// sidecar itself is a generated artifact whose body the caller has already
216/// resolved into the plan, so it is written like any other text sidecar.
217#[allow(clippy::too_many_arguments)]
218pub async fn execute<H, F, G, C>(
219    plan: &Plan,
220    manifest: &mut Manifest,
221    albums: &mut BTreeMap<String, AlbumArt>,
222    playlists: &mut BTreeMap<String, PlaylistState>,
223    desired: &[Desired],
224    synced: &HashMap<String, AlignedLyrics>,
225    ports: Ports<'_, H, F, G, C>,
226    opts: &ExecOptions,
227) -> ExecOutcome
228where
229    H: Http,
230    F: Filesystem,
231    G: Ffmpeg,
232    C: Clock,
233{
234    let Ports {
235        client,
236        http,
237        fs,
238        ffmpeg,
239        clock,
240    } = ports;
241    let by_id: HashMap<&str, &Desired> = desired.iter().map(|d| (d.clip.id.as_str(), d)).collect();
242    let by_path: HashMap<&str, &Desired> = desired.iter().map(|d| (d.path.as_str(), d)).collect();
243    // How many tracked artifact slots reference each path. The inline old-path
244    // cleanup removes a path only once nothing else holds it: each slot that
245    // moves away decrements its reference, and the removal fires only when the
246    // count reaches zero and no action writes the path this run. This keeps a
247    // live file a co-referencing slot still owns (a prior failed swap can leave
248    // two clips sharing a path) while letting the last slot to leave reclaim it,
249    // so nothing is orphaned either (#76).
250    let mut tracked_paths: HashMap<String, u32> = HashMap::new();
251    for (_, entry) in manifest.iter() {
252        for path in entry.artifact_paths() {
253            *tracked_paths.entry(path.to_owned()).or_default() += 1;
254        }
255    }
256    for art in albums.values() {
257        for state in [
258            art.folder_jpg.as_ref(),
259            art.folder_webp.as_ref(),
260            art.folder_mp4.as_ref(),
261        ]
262        .into_iter()
263        .flatten()
264        {
265            *tracked_paths.entry(state.path.clone()).or_default() += 1;
266        }
267    }
268    for playlist in playlists.values() {
269        *tracked_paths.entry(playlist.path.clone()).or_default() += 1;
270    }
271    // Static cover art is otherwise fetched twice per clip (#89): once to embed
272    // in the audio tag and once for the per-song `.jpg` sidecar, both from the
273    // same CDN URL. The audio producer caches each cover it embeds here, keyed by
274    // URL, and the sidecar write drains it rather than re-fetching. Only URLs a
275    // `CoverJpg` sidecar will fetch this run are cached, and the sidecar removes
276    // its entry on use, so the map holds at most the covers for the clips in
277    // flight (bounded by `concurrency`), never the whole library.
278    let cover_wanted: HashSet<&str> = plan
279        .actions
280        .iter()
281        .filter_map(|action| match action {
282            Action::WriteArtifact {
283                kind: ArtifactKind::CoverJpg,
284                source_url,
285                ..
286            } if !source_url.is_empty() => Some(source_url.as_str()),
287            _ => None,
288        })
289        .collect();
290    let cover_cache: Mutex<HashMap<String, Vec<u8>>> = Mutex::new(HashMap::new());
291    // The `both` video-cover retention keeps `cover.webp` (transcoded) and
292    // `cover.mp4` (raw) for an album from the SAME `video_cover_url`. Cache that
293    // source on its first fetch so the second folder artifact drains it rather
294    // than fetching the same MP4 twice (#90 reuses the #89 fetch-once path).
295    let mut folder_cover_uses: HashMap<&str, u32> = HashMap::new();
296    for action in &plan.actions {
297        if let Action::WriteArtifact {
298            kind: ArtifactKind::FolderWebp | ArtifactKind::FolderMp4,
299            source_url,
300            ..
301        } = action
302            && !source_url.is_empty()
303        {
304            *folder_cover_uses.entry(source_url.as_str()).or_default() += 1;
305        }
306    }
307    let shared_cover_urls: HashSet<&str> = folder_cover_uses
308        .into_iter()
309        .filter(|(_, uses)| *uses > 1)
310        .map(|(url, _)| url)
311        .collect();
312    let ctx = Ctx {
313        http,
314        fs,
315        ffmpeg,
316        clock,
317        opts,
318        by_id: &by_id,
319        by_path: &by_path,
320        synced,
321        cover_cache: &cover_cache,
322        cover_wanted: &cover_wanted,
323        shared_cover_urls: &shared_cover_urls,
324    };
325
326    let mut outcome = ExecOutcome::default();
327    // Destinations whose write has actually committed this run, gating old-path
328    // cleanup so a vacated sidecar/stem is kept only when a *successful* write
329    // also targets it (#142). Serial commit order makes this a clean prefix.
330    let mut committed: BTreeSet<String> = BTreeSet::new();
331
332    // Audio (Download/Reformat), fetched-artifact (WriteArtifact with no inline
333    // content), and stem (WriteStem) actions all split their work to maintain the
334    // CRITICAL DELETION-SAFETY INVARIANT: NO destination write, file removal, or
335    // manifest/album/playlist mutation happens off plan order:
336    //
337    // - concurrent preparers ([`prepare`](Ctx::prepare)) do only the slow,
338    //   side-effect-free work — fetch CDN/WAV bytes, transcode, tag — returning
339    //   bytes and the routing metadata the committer needs; and
340    // - a single serial committer below writes those bytes to the destination,
341    //   removes any superseded file, and records the manifest/album/playlist
342    //   entry, in strict plan-index order, interleaved with the remaining serial
343    //   actions.
344    //
345    // The shared client is the only `&mut` port and its API calls must stay
346    // ordered, so it rides behind an async mutex; each producer locks it only for
347    // the brief WAV-render calls and runs the heavy work unlocked. Prepares are
348    // yielded in plan order and bounded to `concurrency` in flight (and buffered),
349    // so at most about `concurrency` payloads are ever held in memory — never the
350    // whole library.
351    let client_lock = AsyncMutex::new(client);
352    let concurrency = opts.concurrency.max(1) as usize;
353    let ctx_ref = &ctx;
354    let client_lock_ref = &client_lock;
355    // Clip IDs already in the manifest before this plan runs. Per-clip
356    // artifacts and stems for these clips are prepared concurrently; for new
357    // clips (not yet in the manifest) the serial apply path handles them after
358    // the audio commit, so the owner-absent guard fires correctly.
359    let pre_clip_ids: HashSet<String> = manifest.entries.keys().cloned().collect();
360    // Clip IDs with a concurrent audio (Download/Reformat) action this run.
361    // Used to keep CoverJpg serial when its audio producer will cache the same
362    // cover URL (#89); preparing both concurrently races the remove vs insert.
363    let audio_clip_ids: HashSet<&str> = plan
364        .actions
365        .iter()
366        .filter_map(|action| match action {
367            Action::Download { clip, .. } | Action::Reformat { clip, .. } => Some(clip.id.as_str()),
368            _ => None,
369        })
370        .collect();
371    let mut prepares = stream::iter(
372        plan.actions
373            .iter()
374            .filter(|action| is_prepareable(action, &pre_clip_ids, &audio_clip_ids))
375            .map(|action| async move { ctx_ref.prepare(client_lock_ref, action).await }),
376    )
377    .buffered(concurrency);
378
379    for action in &plan.actions {
380        // Prepareable actions pull their pre-fetched bytes (yielded in plan order)
381        // and commit them here; every other action applies its own effect. Both the
382        // serial commit and the serial apply run in the same serial loop, so all
383        // destination and manifest effects keep the plan's order exactly.
384        let result = if is_prepareable(action, &pre_clip_ids, &audio_clip_ids) {
385            match prepares.next().await {
386                Some(Ok(Prepared::Audio(rendered))) => ctx.commit_audio(manifest, rendered),
387                Some(Ok(Prepared::Artifact(prepared))) => ctx.commit_artifact(
388                    manifest,
389                    albums,
390                    playlists,
391                    prepared,
392                    &mut tracked_paths,
393                    &committed,
394                ),
395                Some(Ok(Prepared::Stem(prepared))) => {
396                    ctx.commit_stem(manifest, prepared, &mut tracked_paths, &committed)
397                }
398                Some(Err(fail)) => Err(fail),
399                None => unreachable!("buffered yields one result per prepareable action"),
400            }
401        } else {
402            ctx.apply(
403                client_lock_ref,
404                action,
405                manifest,
406                albums,
407                playlists,
408                &mut tracked_paths,
409                &committed,
410            )
411            .await
412        };
413        match result {
414            Ok(effect) => {
415                outcome.record(effect);
416                // Record this action's destination now that its write succeeded.
417                // A later action vacating a path removes it only when no
418                // *committed* write also targets it; commit is strictly serial in
419                // plan order, so a planned-but-failed or not-yet-run write never
420                // protects a stale file from cleanup (#142).
421                if let Some(dest) = written_path(action) {
422                    committed.insert(dest.to_owned());
423                }
424            }
425            Err(fail) => {
426                let abort = abort_status(fail.class);
427                outcome.failures.push(Failure {
428                    clip_id: fail.clip_id,
429                    reason: fail.reason,
430                });
431                if let Some(status) = abort {
432                    // A systemic abort stops the run. Dropping the prepare stream
433                    // cancels any in-flight or completed-but-uncommitted producer;
434                    // because producers touch nothing on disk, the destination and
435                    // manifest are left exactly as the committed prefix wrote them,
436                    // with no untracked files and no removed-but-referenced file.
437                    outcome.status = status;
438                    break;
439                }
440            }
441        }
442    }
443    drop(prepares);
444
445    // Renames and deletes can leave an album directory empty; prune those ghost
446    // directories bottom-up. This runs on both the completed and the aborted
447    // paths, and is best-effort: a prune failure is only a missed tidy that the
448    // next run repeats, never a reason to fail the run.
449    let _ = fs.prune_empty_dirs("");
450    outcome
451}
452
453/// Whether an action has a slow, side-effect-free network or transcode phase
454/// that benefits from concurrent preparation. Audio actions (Download/Reformat)
455/// are always prepareable. A fetched artifact (WriteArtifact with no inline
456/// content) or stem write (WriteStem) is prepareable only when its owning clip
457/// was already in the manifest before this plan started: a new clip's sidecar
458/// cannot be prepared concurrently because its audio has not committed yet (the
459/// manifest entry doesn't exist at prepare time), so it falls through to the
460/// serial apply path which checks the manifest after the audio commits.
461///
462/// Two additional cases stay serial to preserve fetch-once dedup:
463///
464/// - [`FolderWebp`](ArtifactKind::FolderWebp) / [`FolderMp4`](ArtifactKind::FolderMp4):
465///   the `both` retention shares one `video_cover_url`; serial ordering lets
466///   the first fetch insert into `cover_cache` and the second drain it (#90).
467/// - [`CoverJpg`](ArtifactKind::CoverJpg) whose owner clip also has an audio
468///   action this run: the audio producer caches the cover bytes in `cover_cache`
469///   (#89); a concurrent CoverJpg drains the cache before the insert, causing a
470///   double fetch and a leaked entry.
471fn is_prepareable(
472    action: &Action,
473    pre_clip_ids: &HashSet<String>,
474    audio_clip_ids: &HashSet<&str>,
475) -> bool {
476    match action {
477        Action::Download { .. } | Action::Reformat { .. } => true,
478        Action::WriteArtifact {
479            kind,
480            owner_id,
481            content: None,
482            ..
483        } => {
484            if matches!(kind, ArtifactKind::FolderWebp | ArtifactKind::FolderMp4) {
485                return false;
486            }
487            if *kind == ArtifactKind::CoverJpg && audio_clip_ids.contains(owner_id.as_str()) {
488                return false;
489            }
490            !is_per_clip_kind(*kind) || pre_clip_ids.contains(owner_id.as_str())
491        }
492        Action::WriteStem { clip_id, .. } => pre_clip_ids.contains(clip_id.as_str()),
493        _ => false,
494    }
495}
496
497/// The destination path an action writes on success, or `None` for actions that
498/// write no file (skips, deletes). The serial committer records this once the
499/// action succeeds, so a later action vacating that same path keeps it rather
500/// than removing a freshly written file (#142, #76).
501fn written_path(action: &Action) -> Option<&str> {
502    match action {
503        Action::Download { path, .. }
504        | Action::Reformat { path, .. }
505        | Action::WriteArtifact { path, .. }
506        | Action::WriteStem { path, .. } => Some(path),
507        Action::Rename { to, .. }
508        | Action::MoveArtifact { to, .. }
509        | Action::MoveStem { to, .. } => Some(to),
510        _ => None,
511    }
512}
513
514/// A rendered-but-uncommitted audio result: the tagged bytes plus what the serial
515/// committer needs to place them. Produced concurrently and side-effect-free (no
516/// destination write, no removal, no manifest touch); [`commit_audio`] applies
517/// all of those in plan order.
518struct RenderedAudio {
519    clip_id: String,
520    path: String,
521    format: AudioFormat,
522    /// The superseded file to remove after the new one lands (a [`Reformat`]),
523    /// or `None` for a plain [`Download`].
524    from_path: Option<String>,
525    effect: Effect,
526    bytes: Vec<u8>,
527}
528
529/// A fetched-but-uncommitted artifact result: bytes for one
530/// [`WriteArtifact`](Action::WriteArtifact) with no inline content. Produced
531/// concurrently and side-effect-free; [`commit_artifact`](Ctx::commit_artifact)
532/// applies all filesystem and manifest/album/playlist effects in plan order.
533struct PreparedArtifact {
534    kind: ArtifactKind,
535    path: String,
536    hash: String,
537    owner_id: String,
538    bytes: Vec<u8>,
539}
540
541/// A fetched-but-uncommitted stem result: bytes for one
542/// [`WriteStem`](Action::WriteStem) action (including any WAV render + poll).
543/// Produced concurrently and side-effect-free; [`commit_stem`](Ctx::commit_stem)
544/// applies all filesystem and manifest effects in plan order.
545struct PreparedStem {
546    clip_id: String,
547    key: String,
548    path: String,
549    hash: String,
550    bytes: Vec<u8>,
551}
552
553/// The result of one concurrent preparation: audio, an artifact, or a stem.
554enum Prepared {
555    Audio(RenderedAudio),
556    Artifact(PreparedArtifact),
557    Stem(PreparedStem),
558}
559
560/// A cover image resolved for embedding: owned bytes plus their MIME type.
561struct EmbedCover {
562    bytes: Vec<u8>,
563    mime: &'static str,
564}
565
566impl EmbedCover {
567    /// Borrow as the [`Cover`] the taggers take.
568    fn as_cover(&self) -> Cover<'_> {
569        Cover {
570            bytes: &self.bytes,
571            mime: self.mime,
572        }
573    }
574}
575
576/// What an applied action did, for the outcome counters.
577enum Effect {
578    Downloaded,
579    Reformatted,
580    Retagged,
581    Renamed,
582    Deleted,
583    Skipped,
584    ArtifactWritten,
585    ArtifactDeleted,
586}
587
588/// How a failure should be handled (SYNC-17).
589#[derive(Debug, Clone, Copy)]
590enum Class {
591    /// Stop the account run; do not retry.
592    Auth,
593    /// Stop the account run: a full disk is systemic, like auth, so aborting
594    /// beats skipping every remaining clip (each of which would first burn a
595    /// server-side WAV-render budget before failing the same way).
596    Disk,
597    /// Retry a bounded number of times, then record and skip.
598    Transient,
599    /// Record and skip immediately.
600    Permanent,
601}
602
603/// A classified action failure attributed to a clip.
604struct Fail {
605    class: Class,
606    clip_id: String,
607    reason: String,
608}
609
610/// The run-ending status for a failure class, or `None` when the failure is
611/// per-clip and the run continues.
612fn abort_status(class: Class) -> Option<RunStatus> {
613    match class {
614        Class::Auth => Some(RunStatus::AuthAborted),
615        Class::Disk => Some(RunStatus::DiskFull),
616        Class::Transient | Class::Permanent => None,
617    }
618}
619
620fn auth_fail(clip_id: impl Into<String>, reason: impl Into<String>) -> Fail {
621    Fail {
622        class: Class::Auth,
623        clip_id: clip_id.into(),
624        reason: reason.into(),
625    }
626}
627
628fn transient_fail(clip_id: impl Into<String>, reason: impl Into<String>) -> Fail {
629    Fail {
630        class: Class::Transient,
631        clip_id: clip_id.into(),
632        reason: reason.into(),
633    }
634}
635
636fn permanent_fail(clip_id: impl Into<String>, reason: impl Into<String>) -> Fail {
637    Fail {
638        class: Class::Permanent,
639        clip_id: clip_id.into(),
640        reason: reason.into(),
641    }
642}
643
644fn disk_fail(clip_id: impl Into<String>, reason: impl Into<String>) -> Fail {
645    Fail {
646        class: Class::Disk,
647        clip_id: clip_id.into(),
648        reason: reason.into(),
649    }
650}
651
652/// Whether an artifact kind is album-scoped folder art (owned by a root id and
653/// recorded on the album store) rather than a per-clip sidecar (recorded on the
654/// manifest).
655fn is_album_kind(kind: ArtifactKind) -> bool {
656    matches!(
657        kind,
658        ArtifactKind::FolderJpg | ArtifactKind::FolderWebp | ArtifactKind::FolderMp4
659    )
660}
661
662/// True for the library-scoped playlist artifact, routed to the playlist store.
663fn is_playlist_kind(kind: ArtifactKind) -> bool {
664    matches!(kind, ArtifactKind::Playlist)
665}
666
667/// True for a per-song sidecar (`cover.jpg`/`cover.webp`), whose write requires
668/// the owning clip's manifest entry. Album and playlist kinds are keyed by a
669/// root/playlist id that is deliberately absent from the manifest.
670fn is_per_clip_kind(kind: ArtifactKind) -> bool {
671    matches!(
672        kind,
673        ArtifactKind::CoverJpg
674            | ArtifactKind::CoverWebp
675            | ArtifactKind::DetailsTxt
676            | ArtifactKind::LyricsTxt
677            | ArtifactKind::Lrc
678            | ArtifactKind::VideoMp4
679    )
680}
681
682/// Recover a playlist's display name from its `.m3u8` path's file stem.
683///
684/// The path is `<sanitised name>.m3u8` at the library root, so the stem is the
685/// sanitised name. Reconcile only ever reads a playlist's `path` and `hash`, so
686/// this recovered name is a convenience for humans and its lossiness (the
687/// sanitiser is not reversible) never affects a decision.
688fn playlist_name_from_path(path: &str) -> String {
689    std::path::Path::new(path)
690        .file_stem()
691        .map(|stem| stem.to_string_lossy().into_owned())
692        .unwrap_or_default()
693}
694
695/// A classified fetch failure, not yet attributed to a clip.
696struct FetchError {
697    class: Class,
698    reason: String,
699    retry_after: Option<Duration>,
700}
701
702impl FetchError {
703    fn transient(reason: impl Into<String>, retry_after: Option<Duration>) -> Self {
704        Self {
705            class: Class::Transient,
706            reason: reason.into(),
707            retry_after,
708        }
709    }
710
711    fn permanent(reason: impl Into<String>) -> Self {
712        Self {
713            class: Class::Permanent,
714            reason: reason.into(),
715            retry_after: None,
716        }
717    }
718
719    fn attribute(self, clip_id: &str) -> Fail {
720        Fail {
721            class: self.class,
722            clip_id: clip_id.to_owned(),
723            reason: self.reason,
724        }
725    }
726}
727
728/// The shared, read-only context threaded through every action handler.
729struct Ctx<'a, H, F, G, C> {
730    http: &'a H,
731    fs: &'a F,
732    ffmpeg: &'a G,
733    clock: &'a C,
734    opts: &'a ExecOptions,
735    by_id: &'a HashMap<&'a str, &'a Desired>,
736    by_path: &'a HashMap<&'a str, &'a Desired>,
737    /// This run's fetched aligned (synced) lyrics, keyed by clip id. Audio
738    /// tagging reads a clip's entry to embed an MP3 `SYLT` frame and the plain
739    /// lyric text; a clip absent here is tagged exactly as before. Populated by
740    /// the caller (the fetch is IO), so the engine stays free of direct IO.
741    synced: &'a HashMap<String, AlignedLyrics>,
742    /// Static cover art the audio producer already fetched to embed in the tag,
743    /// keyed by CDN URL, so the matching per-song `.jpg` sidecar reuses it rather
744    /// than fetching the same image again (#89). Only URLs a `CoverJpg` sidecar
745    /// will fetch are inserted (see `cover_wanted`) and each is removed on use, so
746    /// the map stays bounded to the clips in flight. A plain mutex guards it: the
747    /// concurrent producers only ever insert, and the lock is never held across an
748    /// await.
749    cover_cache: &'a Mutex<HashMap<String, Vec<u8>>>,
750    /// The cover URLs a `CoverJpg` sidecar will fetch this run. The producer caches
751    /// a cover only when its URL is here, so a clip whose cover is embedded but
752    /// never written as a sidecar leaves no bytes stranded in `cover_cache`.
753    cover_wanted: &'a HashSet<&'a str>,
754    /// Album video-cover source URLs fetched by more than one folder artifact
755    /// this run. The `both` retention derives `cover.webp` (transcoded) and
756    /// `cover.mp4` (raw) from the SAME `video_cover_url`; the first fetch caches
757    /// the raw source here so the sibling drains it instead of re-fetching (#90
758    /// reuses the #89 fetch-once path). `FolderWebp` sorts before `FolderMp4`, so
759    /// the raw source is always cached before the raw sidecar reads it.
760    shared_cover_urls: &'a HashSet<&'a str>,
761}
762
763impl<H, F, G, C> Ctx<'_, H, F, G, C>
764where
765    H: Http,
766    F: Filesystem,
767    G: Ffmpeg,
768    C: Clock,
769{
770    /// Apply one serial action, returning what it did or why it failed.
771    ///
772    /// Audio actions ([`Download`](Action::Download) and
773    /// [`Reformat`](Action::Reformat)) are always prepared concurrently and never
774    /// reach here. Fetched [`WriteArtifact`](Action::WriteArtifact) and
775    /// [`WriteStem`](Action::WriteStem) actions reach here only when their owning
776    /// clip was NOT in the manifest at plan start (new clips); those for existing
777    /// clips are prepared concurrently and commit through the stream path.
778    #[allow(clippy::too_many_arguments)]
779    async fn apply(
780        &self,
781        client_lock: &ClientLock<'_, C>,
782        action: &Action,
783        manifest: &mut Manifest,
784        albums: &mut BTreeMap<String, AlbumArt>,
785        playlists: &mut BTreeMap<String, PlaylistState>,
786        tracked_paths: &mut HashMap<String, u32>,
787        committed: &BTreeSet<String>,
788    ) -> Result<Effect, Fail> {
789        match action {
790            Action::Download { .. } | Action::Reformat { .. } => {
791                unreachable!("audio actions are prepared concurrently")
792            }
793            Action::Retag {
794                clip,
795                lineage,
796                path,
797            } => self.retag(manifest, clip, lineage, path).await,
798            Action::Rename { from, to } => self.rename(manifest, from, to),
799            Action::Delete { path, clip_id } => self.delete(manifest, path, clip_id),
800            Action::Skip { clip_id } => {
801                self.refresh_preserve(manifest, clip_id);
802                Ok(Effect::Skipped)
803            }
804            Action::WriteArtifact {
805                kind,
806                path,
807                source_url,
808                hash,
809                owner_id,
810                content,
811            } => {
812                // Inline text sidecars carry their body in the plan.
813                // Fetched artifacts for clips already in the manifest are prepared
814                // concurrently and never reach here. Fetched artifacts for new clips
815                // (owner not in the manifest at plan start) are handled here in the
816                // serial path, with the owner-absent guard fired before any fetch.
817                let bytes = match content.as_deref() {
818                    Some(text) => text.as_bytes().to_vec(),
819                    None => {
820                        if is_per_clip_kind(*kind) && manifest.get(owner_id).is_none() {
821                            // Owner never landed (audio failed or never existed).
822                            // Drain any stale cache entry so it doesn't outlive
823                            // this clip, then skip without fetching.
824                            self.cover_cache_lock().remove(source_url);
825                            return Ok(Effect::Skipped);
826                        }
827                        self.artifact_bytes(*kind, source_url, owner_id).await?
828                    }
829                };
830                self.commit_artifact(
831                    manifest,
832                    albums,
833                    playlists,
834                    PreparedArtifact {
835                        kind: *kind,
836                        path: path.clone(),
837                        hash: hash.clone(),
838                        owner_id: owner_id.clone(),
839                        bytes,
840                    },
841                    tracked_paths,
842                    committed,
843                )
844            }
845            Action::DeleteArtifact {
846                kind,
847                path,
848                owner_id,
849            } => self.delete_artifact(manifest, albums, playlists, *kind, path, owner_id),
850            Action::MoveArtifact {
851                kind,
852                from,
853                to,
854                source_url,
855                hash,
856                owner_id,
857            } => {
858                self.move_artifact(
859                    manifest,
860                    albums,
861                    playlists,
862                    *kind,
863                    from,
864                    to,
865                    source_url,
866                    hash,
867                    owner_id,
868                    tracked_paths,
869                    committed,
870                )
871                .await
872            }
873            Action::WriteStem {
874                clip_id,
875                key,
876                stem_id,
877                path,
878                source_url,
879                format,
880                hash,
881            } => {
882                // Stems for clips already in the manifest at plan start are
883                // prepared concurrently and never reach here. Stems for new
884                // clips (owner not yet in the manifest) are fetched here in the
885                // serial path, after the audio commit, with the same owner-absent
886                // guard as the old serial write_stem.
887                if manifest.get(clip_id).is_none() {
888                    return Ok(Effect::Skipped);
889                }
890                let bytes = self
891                    .fetch_stem_bytes(client_lock, clip_id, stem_id, source_url, *format)
892                    .await?;
893                self.commit_stem(
894                    manifest,
895                    PreparedStem {
896                        clip_id: clip_id.clone(),
897                        key: key.clone(),
898                        path: path.clone(),
899                        hash: hash.clone(),
900                        bytes,
901                    },
902                    tracked_paths,
903                    committed,
904                )
905            }
906            Action::DeleteStem { clip_id, key, path } => {
907                self.delete_stem(manifest, clip_id, key, path)
908            }
909            Action::MoveStem {
910                clip_id,
911                key,
912                stem_id,
913                from,
914                to,
915                source_url,
916                format,
917                hash,
918            } => {
919                self.move_stem(
920                    client_lock,
921                    manifest,
922                    clip_id,
923                    key,
924                    stem_id,
925                    from,
926                    to,
927                    source_url,
928                    *format,
929                    hash,
930                    tracked_paths,
931                    committed,
932                )
933                .await
934            }
935        }
936    }
937
938    /// Render one audio action's tagged bytes, side-effect-free.
939    ///
940    /// This is the concurrent part: it fetches, transcodes, and tags the file
941    /// (through shared ports, plus the client behind `client_lock`), then returns
942    /// the bytes and where they must go. It deliberately writes nothing, removes
943    /// nothing, and never touches `manifest`, so many run at once and an aborted
944    /// run can drop them with no destination or manifest effect. The serial
945    /// [`commit_audio`](Self::commit_audio) applies those effects in plan order.
946    async fn prepare_audio(
947        &self,
948        client_lock: &ClientLock<'_, C>,
949        action: &Action,
950    ) -> Result<RenderedAudio, Fail> {
951        match action {
952            Action::Download {
953                clip,
954                lineage,
955                path,
956                format,
957            } => {
958                let bytes = self
959                    .produce_audio(client_lock, clip, lineage, *format)
960                    .await?;
961                Ok(RenderedAudio {
962                    clip_id: clip.id.clone(),
963                    path: path.clone(),
964                    format: *format,
965                    from_path: None,
966                    effect: Effect::Downloaded,
967                    bytes,
968                })
969            }
970            Action::Reformat {
971                clip,
972                path,
973                from_path,
974                from: _,
975                to,
976            } => {
977                // A Reformat action carries no lineage, so recover it from the
978                // desired set (the same context that drove naming and the hash),
979                // falling back to a self-rooted context when the clip is not in
980                // the current selection.
981                let lineage = self
982                    .by_id
983                    .get(clip.id.as_str())
984                    .map(|d| d.lineage.clone())
985                    .unwrap_or_else(|| LineageContext::own_root(clip));
986                let bytes = self.produce_audio(client_lock, clip, &lineage, *to).await?;
987                Ok(RenderedAudio {
988                    clip_id: clip.id.clone(),
989                    path: path.clone(),
990                    format: *to,
991                    from_path: Some(from_path.clone()),
992                    effect: Effect::Reformatted,
993                    bytes,
994                })
995            }
996            _ => unreachable!("prepare_audio only handles audio actions"),
997        }
998    }
999
1000    /// Commit one rendered audio result serially, in plan order.
1001    ///
1002    /// Writes the tagged bytes to the destination, then, for a [`Reformat`], drops
1003    /// the superseded file, then records the manifest entry. Ordering the write
1004    /// before the removal keeps a crash from losing both copies; keeping all of
1005    /// this off the concurrent phase preserves the sequential executor's plan-order
1006    /// guarantee for every destination and manifest effect.
1007    fn commit_audio(
1008        &self,
1009        manifest: &mut Manifest,
1010        rendered: RenderedAudio,
1011    ) -> Result<Effect, Fail> {
1012        let RenderedAudio {
1013            clip_id,
1014            path,
1015            format,
1016            from_path,
1017            effect,
1018            bytes,
1019        } = rendered;
1020        let size = self.write_verify(&clip_id, &path, &bytes)?;
1021        if let Some(from) = from_path {
1022            // The new file is safely in place; only now drop the old rendering.
1023            self.fs.remove(&from).map_err(|err| {
1024                permanent_fail(&clip_id, format!("could not remove old file: {err}"))
1025            })?;
1026        }
1027        manifest.insert(clip_id.clone(), self.entry(&clip_id, &path, format, size));
1028        Ok(effect)
1029    }
1030
1031    /// Lock the cover cache, panicking on poison (uniform access point, no repeated magic string).
1032    fn cover_cache_lock(&self) -> std::sync::MutexGuard<'_, HashMap<String, Vec<u8>>> {
1033        self.cover_cache.lock().expect("cover cache mutex poisoned")
1034    }
1035
1036    /// Prepare one concurrent action side-effect-free, returning the bytes and
1037    /// routing metadata the serial committer needs. Only actions that pass
1038    /// [`is_prepareable`] reach here.
1039    async fn prepare(
1040        &self,
1041        client_lock: &ClientLock<'_, C>,
1042        action: &Action,
1043    ) -> Result<Prepared, Fail> {
1044        match action {
1045            Action::Download { .. } | Action::Reformat { .. } => self
1046                .prepare_audio(client_lock, action)
1047                .await
1048                .map(Prepared::Audio),
1049            Action::WriteArtifact {
1050                kind,
1051                path,
1052                source_url,
1053                hash,
1054                owner_id,
1055                content: None,
1056            } => {
1057                let bytes = self.artifact_bytes(*kind, source_url, owner_id).await?;
1058                Ok(Prepared::Artifact(PreparedArtifact {
1059                    kind: *kind,
1060                    path: path.clone(),
1061                    hash: hash.clone(),
1062                    owner_id: owner_id.clone(),
1063                    bytes,
1064                }))
1065            }
1066            Action::WriteStem {
1067                clip_id,
1068                key,
1069                stem_id,
1070                path,
1071                source_url,
1072                format,
1073                hash,
1074            } => {
1075                let bytes = self
1076                    .fetch_stem_bytes(client_lock, clip_id, stem_id, source_url, *format)
1077                    .await?;
1078                Ok(Prepared::Stem(PreparedStem {
1079                    clip_id: clip_id.clone(),
1080                    key: key.clone(),
1081                    path: path.clone(),
1082                    hash: hash.clone(),
1083                    bytes,
1084                }))
1085            }
1086            _ => unreachable!("prepare only handles prepareable actions"),
1087        }
1088    }
1089
1090    /// Commit one prepared artifact result serially, in plan order.
1091    ///
1092    /// Writes the pre-fetched bytes, removes any stale copy left at the previously
1093    /// tracked path (when the audio moved), then records the slot on the manifest,
1094    /// album, or playlist store. All filesystem and state effects are identical to
1095    /// what the former serial [`write_artifact`] did; moving the slow fetch (and
1096    /// optional transcode) into [`prepare`] is the only change.
1097    ///
1098    /// A per-clip sidecar is skipped when its owning clip's audio is absent from
1099    /// the manifest: the audio failed or never existed this run, so the sidecar
1100    /// must not land without an owner (the preparation was speculative).
1101    fn commit_artifact(
1102        &self,
1103        manifest: &mut Manifest,
1104        albums: &mut BTreeMap<String, AlbumArt>,
1105        playlists: &mut BTreeMap<String, PlaylistState>,
1106        prepared: PreparedArtifact,
1107        tracked_paths: &mut HashMap<String, u32>,
1108        committed: &BTreeSet<String>,
1109    ) -> Result<Effect, Fail> {
1110        let PreparedArtifact {
1111            kind,
1112            path,
1113            hash,
1114            owner_id,
1115            bytes,
1116        } = prepared;
1117        if is_per_clip_kind(kind) && manifest.get(&owner_id).is_none() {
1118            return Ok(Effect::Skipped);
1119        }
1120        let old_path = match kind {
1121            ArtifactKind::CoverJpg => manifest
1122                .get(&owner_id)
1123                .and_then(|e| e.cover_jpg.as_ref())
1124                .map(|s| s.path.clone()),
1125            ArtifactKind::CoverWebp => manifest
1126                .get(&owner_id)
1127                .and_then(|e| e.cover_webp.as_ref())
1128                .map(|s| s.path.clone()),
1129            ArtifactKind::DetailsTxt => manifest
1130                .get(&owner_id)
1131                .and_then(|e| e.details_txt.as_ref())
1132                .map(|s| s.path.clone()),
1133            ArtifactKind::LyricsTxt => manifest
1134                .get(&owner_id)
1135                .and_then(|e| e.lyrics_txt.as_ref())
1136                .map(|s| s.path.clone()),
1137            ArtifactKind::Lrc => manifest
1138                .get(&owner_id)
1139                .and_then(|e| e.lrc.as_ref())
1140                .map(|s| s.path.clone()),
1141            ArtifactKind::VideoMp4 => manifest
1142                .get(&owner_id)
1143                .and_then(|e| e.video_mp4.as_ref())
1144                .map(|s| s.path.clone()),
1145            ArtifactKind::FolderJpg | ArtifactKind::FolderWebp | ArtifactKind::FolderMp4 => albums
1146                .get(&owner_id)
1147                .and_then(|a| a.artifact(kind))
1148                .map(|s| s.path.clone()),
1149            ArtifactKind::Playlist => None,
1150        };
1151        self.write_verify(&owner_id, &path, &bytes)?;
1152        if let Some(old) = old_path.as_deref()
1153            && !old.is_empty()
1154            && old != path
1155        {
1156            let still_referenced = tracked_paths
1157                .get_mut(old)
1158                .map(|count| {
1159                    *count = count.saturating_sub(1);
1160                    *count > 0
1161                })
1162                .unwrap_or(false);
1163            if !still_referenced && !committed.contains(old) {
1164                self.fs.remove(old).map_err(|err| {
1165                    permanent_fail(
1166                        &owner_id,
1167                        format!("could not remove old sidecar {old}: {err}"),
1168                    )
1169                })?;
1170            }
1171        }
1172        if is_album_kind(kind) {
1173            set_album_artifact(
1174                albums,
1175                &owner_id,
1176                kind,
1177                Some(ArtifactState {
1178                    path: path.to_owned(),
1179                    hash: hash.to_owned(),
1180                }),
1181            );
1182        } else if is_playlist_kind(kind) {
1183            set_playlist(
1184                playlists,
1185                &owner_id,
1186                Some(PlaylistState {
1187                    name: playlist_name_from_path(&path),
1188                    path: path.to_owned(),
1189                    hash: hash.to_owned(),
1190                }),
1191            );
1192        } else if let Some(entry) = manifest.entries.get_mut(&owner_id) {
1193            set_manifest_artifact(
1194                entry,
1195                kind,
1196                Some(ArtifactState {
1197                    path: path.to_owned(),
1198                    hash: hash.to_owned(),
1199                }),
1200            );
1201        }
1202        Ok(Effect::ArtifactWritten)
1203    }
1204
1205    /// Commit one prepared stem result serially, in plan order.
1206    ///
1207    /// Writes the pre-fetched bytes (including any WAV render), removes any stale
1208    /// copy left at the previously tracked path, and records the stem slot.
1209    /// All filesystem and manifest effects are identical to what the former serial
1210    /// [`write_stem`] did; moving the slow fetch into [`prepare`] is the only change.
1211    ///
1212    /// Skipped when the owning clip's audio is absent from the manifest.
1213    fn commit_stem(
1214        &self,
1215        manifest: &mut Manifest,
1216        prepared: PreparedStem,
1217        tracked_paths: &mut HashMap<String, u32>,
1218        committed: &BTreeSet<String>,
1219    ) -> Result<Effect, Fail> {
1220        let PreparedStem {
1221            clip_id,
1222            key,
1223            path,
1224            hash,
1225            bytes,
1226        } = prepared;
1227        if manifest.get(&clip_id).is_none() {
1228            return Ok(Effect::Skipped);
1229        }
1230        let old_path = manifest
1231            .get(&clip_id)
1232            .and_then(|e| e.stems.get(&key))
1233            .map(|s| s.path.clone());
1234        self.write_verify(&clip_id, &path, &bytes)?;
1235        if let Some(old) = old_path.as_deref()
1236            && !old.is_empty()
1237            && old != path
1238        {
1239            let still_referenced = tracked_paths
1240                .get_mut(old)
1241                .map(|count| {
1242                    *count = count.saturating_sub(1);
1243                    *count > 0
1244                })
1245                .unwrap_or(false);
1246            if !still_referenced && !committed.contains(old) {
1247                self.fs.remove(old).map_err(|err| {
1248                    permanent_fail(&clip_id, format!("could not remove old stem {old}: {err}"))
1249                })?;
1250            }
1251        }
1252        if let Some(entry) = manifest.entries.get_mut(&clip_id) {
1253            set_manifest_stem(
1254                entry,
1255                &key,
1256                Some(ArtifactState {
1257                    path: path.to_owned(),
1258                    hash: hash.to_owned(),
1259                }),
1260            );
1261        }
1262        Ok(Effect::ArtifactWritten)
1263    }
1264
1265    /// Re-tag the existing file in place to match current metadata and art.
1266    async fn retag(
1267        &self,
1268        manifest: &mut Manifest,
1269        clip: &Clip,
1270        lineage: &LineageContext,
1271        path: &str,
1272    ) -> Result<Effect, Fail> {
1273        let Some(format) = manifest.get(&clip.id).map(|entry| entry.format) else {
1274            return Err(permanent_fail(
1275                &clip.id,
1276                "retag target missing from manifest",
1277            ));
1278        };
1279
1280        if format == AudioFormat::Wav {
1281            let (meta, synced) = self.track_meta(clip, lineage);
1282            let cover = self.resolve_cover(clip, format).await?;
1283            let existing = self.fs.read(path).map_err(|err| {
1284                permanent_fail(&clip.id, format!("could not read for retag: {err}"))
1285            })?;
1286            let tagged = tag_wav(
1287                &existing,
1288                &meta,
1289                cover.as_ref().map(EmbedCover::as_cover),
1290                synced,
1291            )
1292            .map_err(|err| permanent_fail(&clip.id, err.to_string()))?;
1293            let size = self.write_verify(&clip.id, path, &tagged)?;
1294            self.refresh_hashes(manifest, &clip.id, Some(size));
1295            return Ok(Effect::Retagged);
1296        }
1297
1298        let (meta, synced) = self.track_meta(clip, lineage);
1299        let cover = self.resolve_cover(clip, format).await?;
1300        let cover = cover.as_ref().map(EmbedCover::as_cover);
1301        let existing = self
1302            .fs
1303            .read(path)
1304            .map_err(|err| permanent_fail(&clip.id, format!("could not read for retag: {err}")))?;
1305        let tagged = match format {
1306            AudioFormat::Mp3 => tag_mp3(&existing, &meta, cover, synced),
1307            AudioFormat::Flac => tag_flac(&existing, &meta, cover),
1308            AudioFormat::Alac => tag_alac(&existing, &meta, cover),
1309            AudioFormat::Wav => unreachable!("WAV handled above"),
1310        }
1311        .map_err(|err| permanent_fail(&clip.id, err.to_string()))?;
1312        let size = self.write_verify(&clip.id, path, &tagged)?;
1313        self.refresh_hashes(manifest, &clip.id, Some(size));
1314        Ok(Effect::Retagged)
1315    }
1316
1317    /// Move the file and update the entry's path (and protection).
1318    fn rename(&self, manifest: &mut Manifest, from: &str, to: &str) -> Result<Effect, Fail> {
1319        let label = self
1320            .by_path
1321            .get(to)
1322            .map(|d| d.clip.id.clone())
1323            .unwrap_or_else(|| to.to_owned());
1324        self.fs.rename(from, to).map_err(|err| {
1325            if err.is_out_of_space() {
1326                disk_fail(label, "disk full: no space left to rename")
1327            } else {
1328                permanent_fail(label, format!("rename failed: {err}"))
1329            }
1330        })?;
1331
1332        let clip_id = self.by_path.get(to).map(|d| d.clip.id.clone()).or_else(|| {
1333            manifest
1334                .entries
1335                .iter()
1336                .find(|(_, entry)| entry.path == from)
1337                .map(|(id, _)| id.clone())
1338        });
1339        if let Some(id) = clip_id
1340            && let Some(entry) = manifest.entries.get_mut(&id)
1341        {
1342            entry.path = to.to_owned();
1343            if let Some(d) = self.by_path.get(to) {
1344                entry.preserve = preserve_for(d);
1345            }
1346        }
1347        Ok(Effect::Renamed)
1348    }
1349
1350    /// Remove the file and drop the manifest entry.
1351    fn delete(&self, manifest: &mut Manifest, path: &str, clip_id: &str) -> Result<Effect, Fail> {
1352        self.fs
1353            .remove(path)
1354            .map_err(|err| permanent_fail(clip_id, format!("delete failed: {err}")))?;
1355        manifest.remove(clip_id);
1356        Ok(Effect::Deleted)
1357    }
1358
1359    /// Relocate a fetched per-clip sidecar with a local rename, falling back to a
1360    /// fetch-and-write when the move is unsafe or the old file has vanished.
1361    ///
1362    /// Reconcile downgrades a pure path drift (same bytes, new path, old file
1363    /// present, fetched kind) to a `MoveArtifact`, so a retitle renames the file
1364    /// rather than re-downloading a cover or re-transcoding an animated WebP
1365    /// (#141). The in-place rename is taken only when `from` is this slot's alone
1366    /// to give up (no other tracked slot references it and no committed write has
1367    /// placed a file there); otherwise, or if the rename fails, fresh bytes are
1368    /// fetched and [`commit_artifact`](Self::commit_artifact) runs the gated
1369    /// old-path cleanup, so a swap or co-reference is handled exactly as before.
1370    #[allow(clippy::too_many_arguments)]
1371    async fn move_artifact(
1372        &self,
1373        manifest: &mut Manifest,
1374        albums: &mut BTreeMap<String, AlbumArt>,
1375        playlists: &mut BTreeMap<String, PlaylistState>,
1376        kind: ArtifactKind,
1377        from: &str,
1378        to: &str,
1379        source_url: &str,
1380        hash: &str,
1381        owner_id: &str,
1382        tracked_paths: &mut HashMap<String, u32>,
1383        committed: &BTreeSet<String>,
1384    ) -> Result<Effect, Fail> {
1385        // A per-clip sidecar needs its owning clip's audio present.
1386        if is_per_clip_kind(kind) && manifest.get(owner_id).is_none() {
1387            return Ok(Effect::Skipped);
1388        }
1389        // Relocate in place only when `from` is ours alone to give up: no other
1390        // tracked slot still references it (a prior failed swap can share a path)
1391        // and no committed write this run has already placed a file there.
1392        // Otherwise the fetch-and-write fallback copies fresh bytes and runs the
1393        // gated old-path cleanup.
1394        let exclusive =
1395            tracked_paths.get(from).is_none_or(|count| *count <= 1) && !committed.contains(from);
1396        if from != to && exclusive {
1397            match self.fs.rename(from, to) {
1398                Ok(()) => {
1399                    if let Some(count) = tracked_paths.get_mut(from) {
1400                        *count = count.saturating_sub(1);
1401                    }
1402                    if let Some(entry) = manifest.entries.get_mut(owner_id) {
1403                        set_manifest_artifact(
1404                            entry,
1405                            kind,
1406                            Some(ArtifactState {
1407                                path: to.to_owned(),
1408                                hash: hash.to_owned(),
1409                            }),
1410                        );
1411                    }
1412                    return Ok(Effect::Renamed);
1413                }
1414                Err(err) if err.is_out_of_space() => {
1415                    return Err(disk_fail(
1416                        owner_id,
1417                        "disk full: no space left to move sidecar",
1418                    ));
1419                }
1420                // The old file has vanished, or the rename is unsupported: fall
1421                // through to a fetch-and-write at `to`.
1422                Err(_) => {}
1423            }
1424        }
1425        let bytes = self.artifact_bytes(kind, source_url, owner_id).await?;
1426        self.commit_artifact(
1427            manifest,
1428            albums,
1429            playlists,
1430            PreparedArtifact {
1431                kind,
1432                path: to.to_owned(),
1433                hash: hash.to_owned(),
1434                owner_id: owner_id.to_owned(),
1435                bytes,
1436            },
1437            tracked_paths,
1438            committed,
1439        )
1440    }
1441    ///
1442    /// An animated cover — a per-clip [`CoverWebp`](ArtifactKind::CoverWebp) or an
1443    /// album [`FolderWebp`](ArtifactKind::FolderWebp) — fetches the clip's
1444    /// `video_cover` MP4 preview and transcodes it to an animated WebP through the
1445    /// ffmpeg port; every other kind is the fetched source verbatim (the static
1446    /// [`CoverJpg`](ArtifactKind::CoverJpg) / album [`FolderJpg`](ArtifactKind::FolderJpg)
1447    /// image, or the raw album [`FolderMp4`](ArtifactKind::FolderMp4) whose
1448    /// `video_cover_url` is kept untranscoded). A fetch or transcode failure
1449    /// is attributed to the owning clip and is a per-clip [`Fail`], except a
1450    /// disk-full transcode, which aborts the run like the audio FLAC path.
1451    async fn artifact_bytes(
1452        &self,
1453        kind: ArtifactKind,
1454        source_url: &str,
1455        owner_id: &str,
1456    ) -> Result<Vec<u8>, Fail> {
1457        // Reuse the cover the audio producer already fetched for the embedded tag
1458        // when it cached this exact URL (#89); otherwise fetch it now. The guard
1459        // is taken and dropped in its own statement so it never spans the await.
1460        let cached = self.cover_cache_lock().remove(source_url);
1461        let source = match cached {
1462            Some(bytes) => bytes,
1463            None => {
1464                let fetched = self
1465                    .fetch_bytes(source_url)
1466                    .await
1467                    .map_err(|err| err.attribute(owner_id))?;
1468                // Cache the raw source when a sibling folder artifact will fetch
1469                // the same URL (the `both` retention: cover.webp + cover.mp4), so
1470                // it is fetched exactly once. Bounded to shared URLs and drained
1471                // on the sibling's use.
1472                if self.shared_cover_urls.contains(source_url) {
1473                    self.cover_cache_lock()
1474                        .insert(source_url.to_owned(), fetched.clone());
1475                }
1476                fetched
1477            }
1478        };
1479        match kind {
1480            ArtifactKind::CoverWebp | ArtifactKind::FolderWebp => self
1481                .ffmpeg
1482                .mp4_to_webp(&source, self.opts.cover_webp)
1483                .await
1484                .map_err(|err| {
1485                    if err.is_out_of_space() {
1486                        disk_fail(owner_id, "disk full: no space left to transcode")
1487                    } else {
1488                        permanent_fail(owner_id, format!("cover transcode failed: {err}"))
1489                    }
1490                }),
1491            // The text sidecars are generated and always carry inline content, so
1492            // `write_artifact` never reaches this fetch path for them. Guard it so
1493            // a future miswiring fails loudly rather than fetching a URL.
1494            ArtifactKind::DetailsTxt | ArtifactKind::LyricsTxt | ArtifactKind::Lrc => Err(
1495                permanent_fail(owner_id, "text sidecar requires inline content"),
1496            ),
1497            ArtifactKind::CoverJpg
1498            | ArtifactKind::FolderJpg
1499            | ArtifactKind::FolderMp4
1500            | ArtifactKind::Playlist
1501            | ArtifactKind::VideoMp4 => Ok(source),
1502        }
1503    }
1504
1505    /// Remove a sidecar file and clear its slot on the owning manifest entry.
1506    ///
1507    /// `remove` is idempotent, so an already-absent sidecar is not a failure.
1508    /// When the owning entry is already gone (its audio was deleted earlier this
1509    /// run, co-deleting the sidecar), there is no slot to clear and that is fine.
1510    ///
1511    /// Folder art is album-scoped: its slot is cleared on the album store keyed by
1512    /// the album's root id, not on a manifest clip.
1513    ///
1514    /// The audio `Delete` is applied before its sidecar `DeleteArtifact`. If the
1515    /// sidecar removal fails after the audio is already gone, the sidecar lingers
1516    /// untracked, but the design stays convergent rather than transactional: the
1517    /// next run re-plans the same removal and retries, and any directory it would
1518    /// have emptied is pruned once the file finally clears.
1519    fn delete_artifact(
1520        &self,
1521        manifest: &mut Manifest,
1522        albums: &mut BTreeMap<String, AlbumArt>,
1523        playlists: &mut BTreeMap<String, PlaylistState>,
1524        kind: ArtifactKind,
1525        path: &str,
1526        owner_id: &str,
1527    ) -> Result<Effect, Fail> {
1528        self.fs
1529            .remove(path)
1530            .map_err(|err| permanent_fail(owner_id, format!("artifact delete failed: {err}")))?;
1531        if is_album_kind(kind) {
1532            set_album_artifact(albums, owner_id, kind, None);
1533        } else if is_playlist_kind(kind) {
1534            set_playlist(playlists, owner_id, None);
1535        } else if let Some(entry) = manifest.entries.get_mut(owner_id) {
1536            set_manifest_artifact(entry, kind, None);
1537        }
1538        Ok(Effect::ArtifactDeleted)
1539    }
1540
1541    /// Relocate a stem with a local rename, falling back to a fetch-and-write
1542    /// when the move is unsafe or the old file has vanished (#141).
1543    ///
1544    /// Reconcile downgrades a pure stem path drift to a `MoveStem`, so a retitle
1545    /// renames the raw stem rather than re-rendering a WAV through `convert_wav`
1546    /// or re-fetching an MP3. The in-place rename is taken only when `from` is
1547    /// this slot's alone to give up (no other tracked slot references it — two
1548    /// same-base clips can share a stem path after a partially-failed swap — and
1549    /// no committed write this run already holds it); otherwise the
1550    /// fetch-and-write fallback re-fetches the correct bytes at `to`, so a
1551    /// co-referenced shared stem is never renamed away with mismatched content.
1552    #[allow(clippy::too_many_arguments)]
1553    async fn move_stem(
1554        &self,
1555        client_lock: &ClientLock<'_, C>,
1556        manifest: &mut Manifest,
1557        clip_id: &str,
1558        key: &str,
1559        stem_id: &str,
1560        from: &str,
1561        to: &str,
1562        source_url: &str,
1563        format: StemFormat,
1564        hash: &str,
1565        tracked_paths: &mut HashMap<String, u32>,
1566        committed: &BTreeSet<String>,
1567    ) -> Result<Effect, Fail> {
1568        if manifest.get(clip_id).is_none() {
1569            return Ok(Effect::Skipped);
1570        }
1571        let exclusive =
1572            tracked_paths.get(from).is_none_or(|count| *count <= 1) && !committed.contains(from);
1573        if from != to && exclusive {
1574            match self.fs.rename(from, to) {
1575                Ok(()) => {
1576                    if let Some(count) = tracked_paths.get_mut(from) {
1577                        *count = count.saturating_sub(1);
1578                    }
1579                    if let Some(entry) = manifest.entries.get_mut(clip_id) {
1580                        set_manifest_stem(
1581                            entry,
1582                            key,
1583                            Some(ArtifactState {
1584                                path: to.to_owned(),
1585                                hash: hash.to_owned(),
1586                            }),
1587                        );
1588                    }
1589                    return Ok(Effect::Renamed);
1590                }
1591                Err(err) if err.is_out_of_space() => {
1592                    return Err(disk_fail(clip_id, "disk full: no space left to move stem"));
1593                }
1594                // The old file has vanished, or the rename is unsupported: fall
1595                // through to a fetch-and-write at `to`.
1596                Err(_) => {}
1597            }
1598        }
1599        let bytes = self
1600            .fetch_stem_bytes(client_lock, clip_id, stem_id, source_url, format)
1601            .await?;
1602        self.commit_stem(
1603            manifest,
1604            PreparedStem {
1605                clip_id: clip_id.to_owned(),
1606                key: key.to_owned(),
1607                path: to.to_owned(),
1608                hash: hash.to_owned(),
1609                bytes,
1610            },
1611            tracked_paths,
1612            committed,
1613        )
1614    }
1615
1616    /// Resolve a stem's RAW bytes in its native container, never transcoding.
1617    ///
1618    /// A `Wav` stem renders the stem clip's lossless WAV through the very same
1619    /// free `convert_wav` + poll flow the main FLAC/WAV audio uses
1620    /// ([`resolve_wav_url`](Self::resolve_wav_url)), keyed on the stem's own
1621    /// `stem_id`, then downloads that WAV. An `Mp3` stem (or a degenerate `Wav`
1622    /// stem with no id to render) downloads its public CDN url directly. Stems
1623    /// are the deliberate exception to the source format: the bytes are returned
1624    /// exactly as delivered and are never re-encoded to FLAC.
1625    async fn fetch_stem_bytes(
1626        &self,
1627        client_lock: &ClientLock<'_, C>,
1628        clip_id: &str,
1629        stem_id: &str,
1630        source_url: &str,
1631        format: StemFormat,
1632    ) -> Result<Vec<u8>, Fail> {
1633        let url = match format {
1634            StemFormat::Wav if !stem_id.is_empty() => {
1635                match self.resolve_wav_url(client_lock, stem_id).await? {
1636                    Some(url) => url,
1637                    None => return Err(transient_fail(clip_id, "stem WAV render was not ready")),
1638                }
1639            }
1640            // Mp3, or a Wav stem with no id to render, downloads the CDN mp3.
1641            _ => source_url.to_owned(),
1642        };
1643        self.fetch_bytes(&url)
1644            .await
1645            .map_err(|err| err.attribute(clip_id))
1646    }
1647
1648    /// Remove one stem file and clear its slot in the owning clip's stem map.
1649    ///
1650    /// `remove` is idempotent, so an already-absent stem is not a failure. When
1651    /// the owning entry is already gone (its audio was deleted earlier this run,
1652    /// co-deleting the stem), there is no slot to clear and that is fine; the
1653    /// emptied `.stems` folder is pruned by the end-of-run directory sweep.
1654    fn delete_stem(
1655        &self,
1656        manifest: &mut Manifest,
1657        clip_id: &str,
1658        key: &str,
1659        path: &str,
1660    ) -> Result<Effect, Fail> {
1661        self.fs
1662            .remove(path)
1663            .map_err(|err| permanent_fail(clip_id, format!("stem delete failed: {err}")))?;
1664        if let Some(entry) = manifest.entries.get_mut(clip_id) {
1665            set_manifest_stem(entry, key, None);
1666        }
1667        Ok(Effect::ArtifactDeleted)
1668    }
1669
1670    /// Download (and transcode/tag) the audio for `clip` in `format`.
1671    async fn produce_audio(
1672        &self,
1673        client_lock: &ClientLock<'_, C>,
1674        clip: &Clip,
1675        lineage: &LineageContext,
1676        format: AudioFormat,
1677    ) -> Result<Vec<u8>, Fail> {
1678        let (meta, synced) = self.track_meta(clip, lineage);
1679        match format {
1680            AudioFormat::Mp3 => {
1681                let url = clip.mp3_url();
1682                let audio = self
1683                    .fetch_bytes(&url)
1684                    .await
1685                    .map_err(|err| err.attribute(&clip.id))?;
1686                let cover = self.resolve_cover(clip, format).await?;
1687                tag_mp3(
1688                    &audio,
1689                    &meta,
1690                    cover.as_ref().map(EmbedCover::as_cover),
1691                    synced,
1692                )
1693                .map_err(|err| permanent_fail(&clip.id, err.to_string()))
1694            }
1695            AudioFormat::Flac | AudioFormat::Alac => {
1696                let wav = self.fetch_wav(client_lock, clip).await?;
1697                let audio = self
1698                    .ffmpeg
1699                    .wav_to_lossless(&wav, format)
1700                    .await
1701                    .map_err(|err| {
1702                        if err.is_out_of_space() {
1703                            disk_fail(&clip.id, "disk full: no space left to transcode")
1704                        } else {
1705                            permanent_fail(&clip.id, format!("transcode failed: {err}"))
1706                        }
1707                    })?;
1708                let cover = self.resolve_cover(clip, format).await?;
1709                let cover = cover.as_ref().map(EmbedCover::as_cover);
1710                let tagged = match format {
1711                    AudioFormat::Alac => tag_alac(&audio, &meta, cover),
1712                    _ => tag_flac(&audio, &meta, cover),
1713                };
1714                tagged.map_err(|err| permanent_fail(&clip.id, err.to_string()))
1715            }
1716            AudioFormat::Wav => {
1717                let wav = self.fetch_wav(client_lock, clip).await?;
1718                let cover = self.resolve_cover(clip, format).await?;
1719                tag_wav(
1720                    &wav,
1721                    &meta,
1722                    cover.as_ref().map(EmbedCover::as_cover),
1723                    synced,
1724                )
1725                .map_err(|err| permanent_fail(&clip.id, err.to_string()))
1726            }
1727        }
1728    }
1729
1730    /// This run's non-empty aligned lyrics for a clip, if any were fetched.
1731    fn synced_for(&self, clip_id: &str) -> Option<&AlignedLyrics> {
1732        self.synced
1733            .get(clip_id)
1734            .filter(|aligned| !aligned.is_empty())
1735    }
1736
1737    /// The track metadata for a clip, paired with its synced lyrics (if any).
1738    ///
1739    /// The feed omits per-clip lyrics, so when this run fetched aligned lyrics
1740    /// for the clip the plain text is folded into `lyrics` here, which the MP3
1741    /// `USLT` and FLAC `LYRICS` tags then carry. The returned [`AlignedLyrics`]
1742    /// is passed on to [`tag_mp3`] for the word-level `SYLT` frame.
1743    fn track_meta<'m>(
1744        &'m self,
1745        clip: &Clip,
1746        lineage: &LineageContext,
1747    ) -> (TrackMetadata, Option<&'m AlignedLyrics>) {
1748        let synced = self.synced_for(&clip.id);
1749        let mut meta = TrackMetadata::from_clip(clip, lineage);
1750        if let Some(aligned) = synced {
1751            meta.lyrics = aligned.plain_text();
1752        }
1753        (meta, synced)
1754    }
1755
1756    /// Resolve the rendered WAV URL and download it.
1757    async fn fetch_wav(
1758        &self,
1759        client_lock: &ClientLock<'_, C>,
1760        clip: &Clip,
1761    ) -> Result<Vec<u8>, Fail> {
1762        let url = match self.resolve_wav_url(client_lock, &clip.id).await? {
1763            Some(url) => url,
1764            None => return Err(transient_fail(&clip.id, "WAV render was not ready")),
1765        };
1766        self.fetch_bytes(&url)
1767            .await
1768            .map_err(|err| err.attribute(&clip.id))
1769    }
1770
1771    /// Read the WAV URL, requesting a render and polling if it is not ready.
1772    ///
1773    /// `None` means the render did not become ready within the poll budget; the
1774    /// caller treats that as a non-fatal transient failure, never a silent skip.
1775    ///
1776    /// Each client call briefly locks `client_lock`; the poll waits happen
1777    /// unlocked, so concurrent clips interleave their WAV renders rather than
1778    /// serialising behind one clip's whole poll budget.
1779    async fn resolve_wav_url(
1780        &self,
1781        client_lock: &ClientLock<'_, C>,
1782        id: &str,
1783    ) -> Result<Option<String>, Fail> {
1784        if let Some(url) = self.wav_url_retrying(client_lock, id).await? {
1785            return Ok(Some(url));
1786        }
1787        self.request_wav_retrying(client_lock, id).await?;
1788        for _ in 0..self.opts.wav_poll_attempts {
1789            self.clock.sleep(self.opts.wav_poll_interval).await;
1790            if let Some(url) = self.wav_url_retrying(client_lock, id).await? {
1791                return Ok(Some(url));
1792            }
1793        }
1794        Ok(None)
1795    }
1796
1797    /// Read the rendered WAV URL, retrying transient API failures with backoff
1798    /// (SYNC-16/17), so the default FLAC path is as resilient as the CDN path.
1799    async fn wav_url_retrying(
1800        &self,
1801        client_lock: &ClientLock<'_, C>,
1802        id: &str,
1803    ) -> Result<Option<String>, Fail> {
1804        let mut attempt: u32 = 0;
1805        loop {
1806            let result = {
1807                let client = client_lock.lock().await;
1808                client.wav_url(self.http, id).await
1809            };
1810            match result {
1811                Ok(url) => return Ok(url),
1812                Err(err) => match self.retry_core(id, err, &mut attempt).await {
1813                    Some(fail) => return Err(fail),
1814                    None => continue,
1815                },
1816            }
1817        }
1818    }
1819
1820    /// Ask Suno to render a WAV, retrying transient API failures with backoff.
1821    async fn request_wav_retrying(
1822        &self,
1823        client_lock: &ClientLock<'_, C>,
1824        id: &str,
1825    ) -> Result<(), Fail> {
1826        let mut attempt: u32 = 0;
1827        loop {
1828            let result = {
1829                let client = client_lock.lock().await;
1830                client.request_wav(self.http, id).await
1831            };
1832            match result {
1833                Ok(()) => return Ok(()),
1834                Err(err) => match self.retry_core(id, err, &mut attempt).await {
1835                    Some(fail) => return Err(fail),
1836                    None => continue,
1837                },
1838            }
1839        }
1840    }
1841
1842    /// Classify a core error from the authenticated WAV flow. On a transient
1843    /// class within budget, back off through the [`Clock`] and return `None` to
1844    /// retry; otherwise return the terminal [`Fail`].
1845    async fn retry_core(&self, id: &str, err: Error, attempt: &mut u32) -> Option<Fail> {
1846        let fail = classify_core(id, err);
1847        if matches!(fail.class, Class::Transient) && *attempt < self.opts.max_retries {
1848            self.clock.sleep(backoff_delay(*attempt, None)).await;
1849            *attempt += 1;
1850            None
1851        } else {
1852            Some(fail)
1853        }
1854    }
1855
1856    /// GET `url`, retrying transient failures with backoff, verifying size.
1857    async fn fetch_bytes(&self, url: &str) -> Result<Vec<u8>, FetchError> {
1858        let mut attempt: u32 = 0;
1859        loop {
1860            let result = self.http.send(HttpRequest::get(url)).await;
1861            match classify_response(result) {
1862                Ok(body) => return Ok(body),
1863                Err(err) => {
1864                    if matches!(err.class, Class::Transient) && attempt < self.opts.max_retries {
1865                        let delay = backoff_delay(attempt, err.retry_after);
1866                        self.clock.sleep(delay).await;
1867                        attempt += 1;
1868                        continue;
1869                    }
1870                    return Err(err);
1871                }
1872            }
1873        }
1874    }
1875
1876    /// Download cover art, trying each candidate URL in order; `None` is fine.
1877    async fn fetch_cover(&self, clip: &Clip) -> Option<Vec<u8>> {
1878        for url in clip.cover_candidates() {
1879            if let Ok(response) = self.http.send(HttpRequest::get(url)).await
1880                && (200..=299).contains(&response.status)
1881                && !response.body.is_empty()
1882            {
1883                // A `CoverJpg` sidecar will fetch this exact URL this run; keep the
1884                // bytes so its write reuses them instead of fetching again (#89).
1885                // The lock guards only the insert, never the await above.
1886                if self.cover_wanted.contains(url) {
1887                    self.cover_cache_lock()
1888                        .insert(url.to_owned(), response.body.clone());
1889                }
1890                return Some(response.body);
1891            }
1892        }
1893        None
1894    }
1895
1896    /// Resolve the cover to embed in `clip`'s audio for `format`.
1897    ///
1898    /// When animated covers are enabled, the container can embed WebP
1899    /// ([`AudioFormat::embeds_animated_cover`]), and the clip has a
1900    /// `video_cover_url`, this fetches that MP4 preview, transcodes it to a
1901    /// bounded animated WebP, and — if the result fits the FLAC picture budget —
1902    /// embeds it as `image/webp`. It falls back to the static JPEG (exactly what
1903    /// a coverless clip embeds today) when the feature is off, the clip has no
1904    /// preview, the container is ALAC, the encode overflows the budget, or the
1905    /// fetch/transcode fails for any non-systemic reason. A disk-full transcode
1906    /// aborts the run, like the audio transcode path.
1907    async fn resolve_cover(
1908        &self,
1909        clip: &Clip,
1910        format: AudioFormat,
1911    ) -> Result<Option<EmbedCover>, Fail> {
1912        if self.opts.embed_animated_cover
1913            && format.embeds_animated_cover()
1914            && !clip.video_cover_url.is_empty()
1915        {
1916            match self.animated_cover_webp(clip).await {
1917                Ok(webp) if webp.len() <= flac_picture_data_budget("image/webp") => {
1918                    return Ok(Some(EmbedCover {
1919                        bytes: webp,
1920                        mime: "image/webp",
1921                    }));
1922                }
1923                // Oversized encode: keep the file valid by embedding the static
1924                // JPEG instead (the intent hash is unchanged, so this does not
1925                // churn; a settings change that makes it fit re-embeds).
1926                Ok(_) => {}
1927                // A full scratch disk is systemic: abort like the audio path.
1928                Err(fail) if matches!(fail.class, Class::Disk) => return Err(fail),
1929                // Any other fetch/transcode failure is best-effort, exactly like a
1930                // failed static-cover fetch: fall back to the JPEG.
1931                Err(_) => {}
1932            }
1933        }
1934        Ok(self.fetch_cover(clip).await.map(|bytes| EmbedCover {
1935            bytes,
1936            mime: "image/jpeg",
1937        }))
1938    }
1939
1940    /// Fetch the clip's MP4 preview and transcode it to an animated WebP.
1941    ///
1942    /// A disk-full transcode is classified [`Class::Disk`] so [`resolve_cover`]
1943    /// can abort the run; every other failure is per-clip and triggers the JPEG
1944    /// fallback.
1945    async fn animated_cover_webp(&self, clip: &Clip) -> Result<Vec<u8>, Fail> {
1946        let mp4 = self
1947            .fetch_bytes(&clip.video_cover_url)
1948            .await
1949            .map_err(|err| err.attribute(&clip.id))?;
1950        self.ffmpeg
1951            .mp4_to_webp(&mp4, self.opts.cover_webp)
1952            .await
1953            .map_err(|err| {
1954                if err.is_out_of_space() {
1955                    disk_fail(&clip.id, "disk full: no space left to transcode cover")
1956                } else {
1957                    permanent_fail(&clip.id, format!("cover transcode failed: {err}"))
1958                }
1959            })
1960    }
1961
1962    /// Write `bytes` atomically, then confirm the on-disk size (SYNC-13/14).
1963    fn write_verify(&self, clip_id: &str, path: &str, bytes: &[u8]) -> Result<u64, Fail> {
1964        self.fs.write_atomic(path, bytes).map_err(|err| {
1965            if err.is_out_of_space() {
1966                disk_fail(clip_id, format!("disk full: no space left to write {path}"))
1967            } else {
1968                permanent_fail(clip_id, format!("write failed: {err}"))
1969            }
1970        })?;
1971        match self.fs.metadata(path) {
1972            Some(stat) if stat.size == bytes.len() as u64 => Ok(stat.size),
1973            Some(stat) => Err(permanent_fail(
1974                clip_id,
1975                format!("wrote {} bytes, expected {}", stat.size, bytes.len()),
1976            )),
1977            None => Ok(bytes.len() as u64),
1978        }
1979    }
1980
1981    /// Build the manifest entry for a freshly written file.
1982    fn entry(&self, clip_id: &str, path: &str, format: AudioFormat, size: u64) -> ManifestEntry {
1983        match self.by_id.get(clip_id) {
1984            Some(d) => manifest_entry(d, size),
1985            None => ManifestEntry {
1986                path: path.to_owned(),
1987                format,
1988                size,
1989                ..ManifestEntry::default()
1990            },
1991        }
1992    }
1993
1994    /// Refresh an existing entry's hashes, protection, and (optionally) size.
1995    fn refresh_hashes(&self, manifest: &mut Manifest, clip_id: &str, size: Option<u64>) {
1996        let desired = self.by_id.get(clip_id).copied();
1997        if let Some(entry) = manifest.entries.get_mut(clip_id) {
1998            if let Some(d) = desired {
1999                entry.meta_hash = d.meta_hash.clone();
2000                entry.art_hash = d.art_hash.clone();
2001                entry.preserve = preserve_for(d);
2002            }
2003            if let Some(size) = size {
2004                entry.size = size;
2005            }
2006        }
2007    }
2008
2009    /// Refresh only an entry's preserve marker from the current desired state.
2010    ///
2011    /// A clip can gain or lose copy/private protection with no file change, which
2012    /// reconcile emits as a [`Skip`](Action::Skip). Refreshing here keeps the
2013    /// persisted marker a faithful image of live protection, so the cross-run
2014    /// delete guard (SYNC-8) never reads it stale.
2015    fn refresh_preserve(&self, manifest: &mut Manifest, clip_id: &str) {
2016        if let Some(d) = self.by_id.get(clip_id).copied()
2017            && let Some(entry) = manifest.entries.get_mut(clip_id)
2018        {
2019            entry.preserve = preserve_for(d);
2020        }
2021    }
2022}
2023
2024/// Build a manifest entry from the desired record (SYNC-8 preserve rule).
2025fn manifest_entry(d: &Desired, size: u64) -> ManifestEntry {
2026    ManifestEntry {
2027        path: d.path.clone(),
2028        format: d.format,
2029        meta_hash: d.meta_hash.clone(),
2030        art_hash: d.art_hash.clone(),
2031        size,
2032        preserve: preserve_for(d),
2033        ..Default::default()
2034    }
2035}
2036
2037/// Whether a written entry must be preserved across runs: held by any copy
2038/// source, or private. The reconcile delete guard reads this marker later.
2039fn preserve_for(d: &Desired) -> bool {
2040    d.private || d.modes.contains(&SourceMode::Copy)
2041}
2042
2043/// Classify one HTTP result into bytes or a [`FetchError`] (SYNC-14/17).
2044fn classify_response(
2045    result: Result<crate::http::HttpResponse, crate::http::TransportError>,
2046) -> Result<Vec<u8>, FetchError> {
2047    let response = match result {
2048        Ok(response) => response,
2049        Err(err) => {
2050            return Err(FetchError::transient(
2051                format!("transport error: {err}"),
2052                None,
2053            ));
2054        }
2055    };
2056    match response.status {
2057        200..=299 => {
2058            if let Some(expected) = content_length(&response) {
2059                let actual = response.body.len() as u64;
2060                if actual != expected {
2061                    return Err(FetchError::transient(
2062                        format!("truncated download: {actual} of {expected} bytes"),
2063                        None,
2064                    ));
2065                }
2066            }
2067            Ok(response.body)
2068        }
2069        401 | 403 => Err(FetchError::transient(
2070            format!("download rejected: status {}", response.status),
2071            None,
2072        )),
2073        408 => Err(FetchError::transient("request timed out", None)),
2074        429 => Err(FetchError::transient(
2075            "rate limited",
2076            retry_after(&response),
2077        )),
2078        500..=599 => Err(FetchError::transient(
2079            format!("server error {}", response.status),
2080            None,
2081        )),
2082        status => Err(FetchError::permanent(format!(
2083            "download failed: status {status}"
2084        ))),
2085    }
2086}
2087
2088/// Map a core [`Error`] from the authenticated WAV flow to a [`Fail`].
2089fn classify_core(id: &str, err: Error) -> Fail {
2090    let reason = err.to_string();
2091    match err {
2092        Error::Auth(_) => auth_fail(id, reason),
2093        Error::RateLimited { .. } | Error::Connection(_) => transient_fail(id, reason),
2094        Error::Api(_)
2095        | Error::BadRequest(_)
2096        | Error::NotFound(_)
2097        | Error::Tag(_)
2098        | Error::Config(_)
2099        | Error::Refused(_) => permanent_fail(id, reason),
2100    }
2101}
2102
2103/// The provider-reported body size from `Content-Length`, if present and valid.
2104fn content_length(response: &crate::http::HttpResponse) -> Option<u64> {
2105    response.header("content-length")?.trim().parse().ok()
2106}
2107
2108#[cfg(test)]
2109mod tests {
2110    use super::*;
2111    use crate::ClerkAuth;
2112    use crate::http::HttpResponse;
2113    use crate::testutil::{MemFs, RecordingClock, Reply, ScriptedHttp, StubFfmpeg};
2114
2115    fn clip(id: &str) -> Clip {
2116        Clip {
2117            id: id.to_owned(),
2118            title: "Song".to_owned(),
2119            audio_url: format!("https://cdn1.suno.ai/{id}.mp3"),
2120            ..Default::default()
2121        }
2122    }
2123
2124    fn art_clip(id: &str) -> Clip {
2125        Clip {
2126            image_large_url: format!("https://art.suno.ai/{id}/large.jpg"),
2127            image_url: format!("https://art.suno.ai/{id}/small.jpg"),
2128            ..clip(id)
2129        }
2130    }
2131
2132    fn desired(clip: Clip, format: AudioFormat) -> Desired {
2133        Desired {
2134            path: format!("{}.{}", clip.id, format.ext()),
2135            lineage: LineageContext::own_root(&clip),
2136            clip,
2137            format,
2138            meta_hash: "m".to_owned(),
2139            art_hash: "art".to_owned(),
2140            modes: vec![SourceMode::Mirror],
2141            trashed: false,
2142            private: false,
2143            artifacts: Vec::new(),
2144            stems: None,
2145        }
2146    }
2147
2148    fn entry(path: &str, format: AudioFormat) -> ManifestEntry {
2149        ManifestEntry {
2150            path: path.to_owned(),
2151            format,
2152            meta_hash: "old".to_owned(),
2153            art_hash: "old-art".to_owned(),
2154            size: 8,
2155            preserve: false,
2156            ..Default::default()
2157        }
2158    }
2159
2160    #[allow(clippy::too_many_arguments)]
2161    fn run<G: Ffmpeg>(
2162        plan: &Plan,
2163        manifest: &mut Manifest,
2164        desired: &[Desired],
2165        http: &ScriptedHttp,
2166        fs: &MemFs,
2167        ffmpeg: &G,
2168        clock: &RecordingClock,
2169        opts: &ExecOptions,
2170    ) -> ExecOutcome {
2171        let mut albums = BTreeMap::new();
2172        run_with_albums(
2173            plan,
2174            manifest,
2175            &mut albums,
2176            desired,
2177            http,
2178            fs,
2179            ffmpeg,
2180            clock,
2181            opts,
2182        )
2183    }
2184
2185    #[allow(clippy::too_many_arguments)]
2186    fn run_with_albums<G: Ffmpeg>(
2187        plan: &Plan,
2188        manifest: &mut Manifest,
2189        albums: &mut BTreeMap<String, AlbumArt>,
2190        desired: &[Desired],
2191        http: &ScriptedHttp,
2192        fs: &MemFs,
2193        ffmpeg: &G,
2194        clock: &RecordingClock,
2195        opts: &ExecOptions,
2196    ) -> ExecOutcome {
2197        let mut playlists = BTreeMap::new();
2198        run_full(
2199            plan,
2200            manifest,
2201            albums,
2202            &mut playlists,
2203            desired,
2204            http,
2205            fs,
2206            ffmpeg,
2207            clock,
2208            opts,
2209        )
2210    }
2211
2212    #[allow(clippy::too_many_arguments)]
2213    fn run_full<G: Ffmpeg>(
2214        plan: &Plan,
2215        manifest: &mut Manifest,
2216        albums: &mut BTreeMap<String, AlbumArt>,
2217        playlists: &mut BTreeMap<String, PlaylistState>,
2218        desired: &[Desired],
2219        http: &ScriptedHttp,
2220        fs: &MemFs,
2221        ffmpeg: &G,
2222        clock: &RecordingClock,
2223        opts: &ExecOptions,
2224    ) -> ExecOutcome {
2225        let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
2226        let synced = HashMap::new();
2227        pollster::block_on(execute(
2228            plan,
2229            manifest,
2230            albums,
2231            playlists,
2232            desired,
2233            &synced,
2234            Ports {
2235                client: &client,
2236                http,
2237                fs,
2238                ffmpeg,
2239                clock,
2240            },
2241            opts,
2242        ))
2243    }
2244
2245    fn small_poll() -> ExecOptions {
2246        ExecOptions {
2247            max_retries: 3,
2248            wav_poll_attempts: 2,
2249            wav_poll_interval: Duration::from_secs(5),
2250            concurrency: 4,
2251            embed_animated_cover: false,
2252            cover_webp: WebpEncodeSettings::default(),
2253        }
2254    }
2255
2256    // ── Download: MP3 ───────────────────────────────────────────────
2257
2258    #[test]
2259    fn download_mp3_writes_tagged_file_and_records_manifest() {
2260        let c = art_clip("a");
2261        let d = desired(c.clone(), AudioFormat::Mp3);
2262        let plan = Plan {
2263            actions: vec![Action::Download {
2264                clip: c.clone(),
2265                lineage: LineageContext::own_root(&c),
2266                path: d.path.clone(),
2267                format: AudioFormat::Mp3,
2268            }],
2269        };
2270        let http = ScriptedHttp::new()
2271            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
2272            .route("a/large.jpg", Reply::ok(b"art-bytes".to_vec()));
2273        let fs = MemFs::new();
2274        let ffmpeg = StubFfmpeg::flac();
2275        let clock = RecordingClock::new();
2276        let mut manifest = Manifest::new();
2277
2278        let outcome = run(
2279            &plan,
2280            &mut manifest,
2281            &[d],
2282            &http,
2283            &fs,
2284            &ffmpeg,
2285            &clock,
2286            &ExecOptions::default(),
2287        );
2288
2289        assert_eq!(outcome.downloaded, 1);
2290        assert_eq!(outcome.failed(), 0);
2291        assert_eq!(outcome.status, RunStatus::Completed);
2292        let written = fs.read_file("a.mp3").unwrap();
2293        assert_eq!(&written[..3], b"ID3");
2294        assert!(written.ends_with(b"mp3-body"));
2295        let entry = manifest.get("a").unwrap();
2296        assert_eq!(entry.path, "a.mp3");
2297        assert_eq!(entry.format, AudioFormat::Mp3);
2298        assert_eq!(entry.meta_hash, "m");
2299        assert_eq!(entry.art_hash, "art");
2300        assert_eq!(entry.size, written.len() as u64);
2301        assert!(!entry.preserve);
2302    }
2303
2304    #[test]
2305    fn download_mp3_embeds_sylt_and_lyrics_from_synced_map() {
2306        // A clip whose alignment was fetched this run gets a word-level SYLT frame
2307        // and its plain lyric text embedded (USLT), end to end through execute.
2308        let c = art_clip("a");
2309        let d = desired(c.clone(), AudioFormat::Mp3);
2310        let plan = Plan {
2311            actions: vec![Action::Download {
2312                clip: c.clone(),
2313                lineage: LineageContext::own_root(&c),
2314                path: d.path.clone(),
2315                format: AudioFormat::Mp3,
2316            }],
2317        };
2318        let http = ScriptedHttp::new()
2319            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
2320            .route("a/large.jpg", Reply::ok(b"art-bytes".to_vec()));
2321        let fs = MemFs::new();
2322        let ffmpeg = StubFfmpeg::flac();
2323        let clock = RecordingClock::new();
2324        let mut manifest = Manifest::new();
2325        let mut albums = BTreeMap::new();
2326        let mut playlists = BTreeMap::new();
2327        let mut synced = HashMap::new();
2328        synced.insert(
2329            "a".to_string(),
2330            AlignedLyrics::from_json(&serde_json::json!({
2331                "aligned_words": [],
2332                "aligned_lyrics": [
2333                    {"text": "hi there", "start_s": 0.5, "end_s": 1.2, "section": "Verse 1",
2334                     "words": [
2335                         {"text": "hi", "start_s": 0.5, "end_s": 0.8},
2336                         {"text": "there", "start_s": 0.9, "end_s": 1.2}
2337                     ]}
2338                ]
2339            })),
2340        );
2341        let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
2342        let outcome = pollster::block_on(execute(
2343            &plan,
2344            &mut manifest,
2345            &mut albums,
2346            &mut playlists,
2347            &[d],
2348            &synced,
2349            Ports {
2350                client: &client,
2351                http: &http,
2352                fs: &fs,
2353                ffmpeg: &ffmpeg,
2354                clock: &clock,
2355            },
2356            &ExecOptions::default(),
2357        ));
2358
2359        assert_eq!(outcome.downloaded, 1);
2360        let written = fs.read_file("a.mp3").unwrap();
2361        let tag = id3::Tag::read_from2(std::io::Cursor::new(written)).unwrap();
2362        assert_eq!(
2363            tag.synchronised_lyrics().count(),
2364            1,
2365            "a SYLT frame is embedded"
2366        );
2367        // The plain lyric text is populated from the alignment for the USLT frame.
2368        assert_eq!(
2369            tag.lyrics().next().map(|frame| frame.text.as_str()),
2370            Some("hi there")
2371        );
2372    }
2373
2374    #[test]
2375    fn download_mp3_embeds_no_sylt_when_synced_map_empty() {
2376        // The synced map is empty when the feature is off (no alignment fetched),
2377        // so no SYLT frame and no lyric text are embedded.
2378        let c = art_clip("a");
2379        let d = desired(c.clone(), AudioFormat::Mp3);
2380        let plan = Plan {
2381            actions: vec![Action::Download {
2382                clip: c.clone(),
2383                lineage: LineageContext::own_root(&c),
2384                path: d.path.clone(),
2385                format: AudioFormat::Mp3,
2386            }],
2387        };
2388        let http = ScriptedHttp::new()
2389            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
2390            .route("a/large.jpg", Reply::ok(b"art-bytes".to_vec()));
2391        let fs = MemFs::new();
2392        let ffmpeg = StubFfmpeg::flac();
2393        let clock = RecordingClock::new();
2394        let mut manifest = Manifest::new();
2395        let mut albums = BTreeMap::new();
2396        let mut playlists = BTreeMap::new();
2397        let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
2398        let outcome = pollster::block_on(execute(
2399            &plan,
2400            &mut manifest,
2401            &mut albums,
2402            &mut playlists,
2403            &[d],
2404            &HashMap::new(),
2405            Ports {
2406                client: &client,
2407                http: &http,
2408                fs: &fs,
2409                ffmpeg: &ffmpeg,
2410                clock: &clock,
2411            },
2412            &ExecOptions::default(),
2413        ));
2414        assert_eq!(outcome.downloaded, 1);
2415        let written = fs.read_file("a.mp3").unwrap();
2416        let tag = id3::Tag::read_from2(std::io::Cursor::new(written)).unwrap();
2417        assert_eq!(tag.synchronised_lyrics().count(), 0);
2418        assert_eq!(tag.lyrics().count(), 0);
2419    }
2420
2421    #[test]
2422    fn download_mp3_uses_cdn_fallback_when_audio_url_empty() {
2423        let mut c = clip("a");
2424        c.audio_url = String::new();
2425        let d = desired(c.clone(), AudioFormat::Mp3);
2426        let plan = Plan {
2427            actions: vec![Action::Download {
2428                clip: c.clone(),
2429                lineage: LineageContext::own_root(&c),
2430                path: d.path.clone(),
2431                format: AudioFormat::Mp3,
2432            }],
2433        };
2434        let http = ScriptedHttp::new().route("cdn1.suno.ai/a.mp3", Reply::ok(b"body".to_vec()));
2435        let fs = MemFs::new();
2436        let mut manifest = Manifest::new();
2437        let outcome = run(
2438            &plan,
2439            &mut manifest,
2440            &[d],
2441            &http,
2442            &fs,
2443            &StubFfmpeg::flac(),
2444            &RecordingClock::new(),
2445            &ExecOptions::default(),
2446        );
2447        assert_eq!(outcome.downloaded, 1);
2448        assert_eq!(http.count("cdn1.suno.ai/a.mp3"), 1);
2449    }
2450
2451    // ── Download: FLAC render + transcode ───────────────────────────
2452
2453    #[test]
2454    fn download_flac_renders_transcodes_and_records() {
2455        let c = clip("b");
2456        let d = desired(c.clone(), AudioFormat::Flac);
2457        let plan = Plan {
2458            actions: vec![Action::Download {
2459                clip: c.clone(),
2460                lineage: LineageContext::own_root(&c),
2461                path: d.path.clone(),
2462                format: AudioFormat::Flac,
2463            }],
2464        };
2465        let http = ScriptedHttp::new()
2466            .with_auth()
2467            .route(
2468                "/wav_file/",
2469                Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/b.wav"}"#),
2470            )
2471            .route("b.wav", Reply::ok(b"wav-bytes".to_vec()));
2472        let fs = MemFs::new();
2473        let clock = RecordingClock::new();
2474        let mut manifest = Manifest::new();
2475
2476        let outcome = run(
2477            &plan,
2478            &mut manifest,
2479            &[d],
2480            &http,
2481            &fs,
2482            &StubFfmpeg::flac(),
2483            &clock,
2484            &ExecOptions::default(),
2485        );
2486
2487        assert_eq!(outcome.downloaded, 1);
2488        assert_eq!(outcome.failed(), 0);
2489        let written = fs.read_file("b.flac").unwrap();
2490        assert_eq!(&written[..4], b"fLaC");
2491        assert_eq!(manifest.get("b").unwrap().format, AudioFormat::Flac);
2492        // The URL was ready immediately, so no render request and no polling.
2493        assert_eq!(http.count("/convert_wav/"), 0);
2494        assert!(clock.sleeps().is_empty());
2495    }
2496
2497    #[test]
2498    fn download_flac_requests_render_then_polls_until_ready() {
2499        let c = clip("c");
2500        let d = desired(c.clone(), AudioFormat::Flac);
2501        let plan = Plan {
2502            actions: vec![Action::Download {
2503                clip: c.clone(),
2504                lineage: LineageContext::own_root(&c),
2505                path: d.path.clone(),
2506                format: AudioFormat::Flac,
2507            }],
2508        };
2509        let http = ScriptedHttp::new()
2510            .with_auth()
2511            .route_seq(
2512                "/wav_file/",
2513                vec![
2514                    Reply::json("{}"),
2515                    Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/c.wav"}"#),
2516                ],
2517            )
2518            .route("/convert_wav/", Reply::status(200))
2519            .route("c.wav", Reply::ok(b"wav".to_vec()));
2520        let clock = RecordingClock::new();
2521        let mut manifest = Manifest::new();
2522
2523        let outcome = run(
2524            &plan,
2525            &mut manifest,
2526            &[d],
2527            &http,
2528            &fs_new(),
2529            &StubFfmpeg::flac(),
2530            &clock,
2531            &small_poll(),
2532        );
2533
2534        assert_eq!(outcome.downloaded, 1);
2535        assert_eq!(http.count("/convert_wav/"), 1);
2536        assert_eq!(clock.sleeps(), vec![Duration::from_secs(5)]);
2537    }
2538
2539    #[test]
2540    fn download_flac_unavailable_render_is_a_nonfatal_failure() {
2541        let c = clip("d");
2542        let d = desired(c.clone(), AudioFormat::Flac);
2543        let plan = Plan {
2544            actions: vec![Action::Download {
2545                clip: c.clone(),
2546                lineage: LineageContext::own_root(&c),
2547                path: d.path.clone(),
2548                format: AudioFormat::Flac,
2549            }],
2550        };
2551        let http = ScriptedHttp::new()
2552            .with_auth()
2553            .route("/wav_file/", Reply::json("{}"))
2554            .route("/convert_wav/", Reply::status(200));
2555        let fs = MemFs::new();
2556        let clock = RecordingClock::new();
2557        let mut manifest = Manifest::new();
2558
2559        let outcome = run(
2560            &plan,
2561            &mut manifest,
2562            &[d],
2563            &http,
2564            &fs,
2565            &StubFfmpeg::flac(),
2566            &clock,
2567            &small_poll(),
2568        );
2569
2570        assert_eq!(outcome.downloaded, 0);
2571        assert_eq!(outcome.failed(), 1);
2572        assert_eq!(outcome.failures[0].clip_id, "d");
2573        assert_eq!(outcome.status, RunStatus::Completed);
2574        assert!(!fs.exists("d.flac"));
2575        assert_eq!(clock.sleeps().len(), 2);
2576    }
2577
2578    #[test]
2579    fn flac_transcode_failure_is_recorded_and_skipped() {
2580        let c = clip("t");
2581        let d = desired(c.clone(), AudioFormat::Flac);
2582        let plan = Plan {
2583            actions: vec![Action::Download {
2584                clip: c.clone(),
2585                lineage: LineageContext::own_root(&c),
2586                path: d.path.clone(),
2587                format: AudioFormat::Flac,
2588            }],
2589        };
2590        let http = ScriptedHttp::new()
2591            .with_auth()
2592            .route(
2593                "/wav_file/",
2594                Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/t.wav"}"#),
2595            )
2596            .route("t.wav", Reply::ok(b"wav".to_vec()));
2597        let fs = MemFs::new();
2598        let mut manifest = Manifest::new();
2599
2600        let outcome = run(
2601            &plan,
2602            &mut manifest,
2603            &[d],
2604            &http,
2605            &fs,
2606            &StubFfmpeg::failing(),
2607            &RecordingClock::new(),
2608            &ExecOptions::default(),
2609        );
2610
2611        assert_eq!(outcome.downloaded, 0);
2612        assert_eq!(outcome.failed(), 1);
2613        assert!(!fs.exists("t.flac"));
2614        assert!(manifest.get("t").is_none());
2615    }
2616
2617    // ── Cover fallback ──────────────────────────────────────────────
2618
2619    #[test]
2620    fn cover_falls_back_when_large_image_is_missing() {
2621        let c = art_clip("e");
2622        let d = desired(c.clone(), AudioFormat::Mp3);
2623        let plan = Plan {
2624            actions: vec![Action::Download {
2625                clip: c.clone(),
2626                lineage: LineageContext::own_root(&c),
2627                path: d.path.clone(),
2628                format: AudioFormat::Mp3,
2629            }],
2630        };
2631        let http = ScriptedHttp::new()
2632            .route("e.mp3", Reply::ok(b"body".to_vec()))
2633            .route("e/large.jpg", Reply::status(404))
2634            .route("e/small.jpg", Reply::ok(b"the-art".to_vec()));
2635        let fs = MemFs::new();
2636        let mut manifest = Manifest::new();
2637
2638        let outcome = run(
2639            &plan,
2640            &mut manifest,
2641            &[d],
2642            &http,
2643            &fs,
2644            &StubFfmpeg::flac(),
2645            &RecordingClock::new(),
2646            &ExecOptions::default(),
2647        );
2648
2649        assert_eq!(outcome.downloaded, 1);
2650        let calls = http.calls();
2651        let large = calls
2652            .iter()
2653            .position(|u| u.contains("e/large.jpg"))
2654            .unwrap();
2655        let small = calls
2656            .iter()
2657            .position(|u| u.contains("e/small.jpg"))
2658            .unwrap();
2659        assert!(large < small, "large art tried before small");
2660    }
2661
2662    // ── Cover reuse: embed + sidecar share one fetch (#89) ──────────
2663
2664    #[test]
2665    fn download_reuses_the_embedded_cover_for_the_jpg_sidecar() {
2666        // The embedded tag and the `.jpg` sidecar want the same cover URL; it is
2667        // fetched once and the bytes serve both.
2668        let c = art_clip("a");
2669        let d = desired(c.clone(), AudioFormat::Mp3);
2670        let plan = Plan {
2671            actions: vec![
2672                Action::Download {
2673                    clip: c.clone(),
2674                    lineage: LineageContext::own_root(&c),
2675                    path: d.path.clone(),
2676                    format: AudioFormat::Mp3,
2677                },
2678                Action::WriteArtifact {
2679                    kind: ArtifactKind::CoverJpg,
2680                    path: "a/cover.jpg".to_owned(),
2681                    source_url: c.selected_image_url().unwrap().to_owned(),
2682                    hash: "art".to_owned(),
2683                    owner_id: "a".to_owned(),
2684                    content: None,
2685                },
2686            ],
2687        };
2688        let http = ScriptedHttp::new()
2689            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
2690            .route("a/large.jpg", Reply::ok(b"the-art".to_vec()));
2691        let fs = MemFs::new();
2692        let mut manifest = Manifest::new();
2693
2694        let outcome = run(
2695            &plan,
2696            &mut manifest,
2697            &[d],
2698            &http,
2699            &fs,
2700            &StubFfmpeg::flac(),
2701            &RecordingClock::new(),
2702            &ExecOptions::default(),
2703        );
2704
2705        assert_eq!(outcome.downloaded, 1);
2706        assert_eq!(outcome.artifacts_written, 1);
2707        assert_eq!(outcome.failed(), 0);
2708        // Fetched once, not twice.
2709        assert_eq!(http.count("a/large.jpg"), 1);
2710        // The sidecar carries the fetched bytes, and the audio was tagged.
2711        assert_eq!(fs.read_file("a/cover.jpg").unwrap(), b"the-art");
2712        assert_eq!(&fs.read_file("a.mp3").unwrap()[..3], b"ID3");
2713    }
2714
2715    #[test]
2716    fn concurrent_downloads_reuse_each_clips_own_cover() {
2717        // Two clips render concurrently; each `.jpg` sidecar gets its own cover
2718        // (no cross-contamination) and each cover URL is fetched exactly once.
2719        let a = art_clip("a");
2720        let b = art_clip("b");
2721        let da = desired(a.clone(), AudioFormat::Mp3);
2722        let db = desired(b.clone(), AudioFormat::Mp3);
2723        let plan = Plan {
2724            actions: vec![
2725                Action::Download {
2726                    clip: a.clone(),
2727                    lineage: LineageContext::own_root(&a),
2728                    path: da.path.clone(),
2729                    format: AudioFormat::Mp3,
2730                },
2731                Action::WriteArtifact {
2732                    kind: ArtifactKind::CoverJpg,
2733                    path: "a/cover.jpg".to_owned(),
2734                    source_url: a.selected_image_url().unwrap().to_owned(),
2735                    hash: "art".to_owned(),
2736                    owner_id: "a".to_owned(),
2737                    content: None,
2738                },
2739                Action::Download {
2740                    clip: b.clone(),
2741                    lineage: LineageContext::own_root(&b),
2742                    path: db.path.clone(),
2743                    format: AudioFormat::Mp3,
2744                },
2745                Action::WriteArtifact {
2746                    kind: ArtifactKind::CoverJpg,
2747                    path: "b/cover.jpg".to_owned(),
2748                    source_url: b.selected_image_url().unwrap().to_owned(),
2749                    hash: "art".to_owned(),
2750                    owner_id: "b".to_owned(),
2751                    content: None,
2752                },
2753            ],
2754        };
2755        let http = ScriptedHttp::new()
2756            .route("a.mp3", Reply::ok(b"a-mp3".to_vec()))
2757            .route("b.mp3", Reply::ok(b"b-mp3".to_vec()))
2758            .route("a/large.jpg", Reply::ok(b"art-a".to_vec()))
2759            .route("b/large.jpg", Reply::ok(b"art-b".to_vec()));
2760        let fs = MemFs::new();
2761        let mut manifest = Manifest::new();
2762
2763        let outcome = run(
2764            &plan,
2765            &mut manifest,
2766            &[da, db],
2767            &http,
2768            &fs,
2769            &StubFfmpeg::flac(),
2770            &RecordingClock::new(),
2771            &small_poll(),
2772        );
2773
2774        assert_eq!(outcome.downloaded, 2);
2775        assert_eq!(outcome.artifacts_written, 2);
2776        assert_eq!(http.count("a/large.jpg"), 1);
2777        assert_eq!(http.count("b/large.jpg"), 1);
2778        assert_eq!(fs.read_file("a/cover.jpg").unwrap(), b"art-a");
2779        assert_eq!(fs.read_file("b/cover.jpg").unwrap(), b"art-b");
2780    }
2781
2782    #[test]
2783    fn cover_sidecar_refetches_when_embed_fell_back_to_another_url() {
2784        // The large image 404s so the embed falls back to the small image; the
2785        // sidecar still wants the (dead) large URL and must NOT be handed the
2786        // small bytes. Reuse is keyed on the exact URL, so nothing is cached and
2787        // the sidecar fetches the large URL itself (then fails on the 404).
2788        let c = art_clip("e");
2789        let d = desired(c.clone(), AudioFormat::Mp3);
2790        let plan = Plan {
2791            actions: vec![
2792                Action::Download {
2793                    clip: c.clone(),
2794                    lineage: LineageContext::own_root(&c),
2795                    path: d.path.clone(),
2796                    format: AudioFormat::Mp3,
2797                },
2798                Action::WriteArtifact {
2799                    kind: ArtifactKind::CoverJpg,
2800                    path: "e/cover.jpg".to_owned(),
2801                    source_url: "https://art.suno.ai/e/large.jpg".to_owned(),
2802                    hash: "art".to_owned(),
2803                    owner_id: "e".to_owned(),
2804                    content: None,
2805                },
2806            ],
2807        };
2808        let http = ScriptedHttp::new()
2809            .route("e.mp3", Reply::ok(b"body".to_vec()))
2810            .route("e/large.jpg", Reply::status(404))
2811            .route("e/small.jpg", Reply::ok(b"small-art".to_vec()));
2812        let fs = MemFs::new();
2813        let mut manifest = Manifest::new();
2814
2815        let outcome = run(
2816            &plan,
2817            &mut manifest,
2818            &[d],
2819            &http,
2820            &fs,
2821            &StubFfmpeg::flac(),
2822            &RecordingClock::new(),
2823            &ExecOptions::default(),
2824        );
2825
2826        assert_eq!(outcome.downloaded, 1);
2827        // The small image was fetched once (the embed fallback) and never reused
2828        // for the large-keyed sidecar; the sidecar went to the network itself.
2829        assert_eq!(http.count("e/small.jpg"), 1);
2830        assert!(
2831            http.count("e/large.jpg") >= 2,
2832            "sidecar refetched the large URL"
2833        );
2834        assert_eq!(manifest.get("e").unwrap().cover_jpg, None);
2835        assert!(!fs.exists("e/cover.jpg"));
2836    }
2837
2838    // ── Atomic write and size verification (SYNC-13/14) ─────────────
2839
2840    #[test]
2841    fn failed_write_leaves_the_prior_file_intact() {
2842        let c = clip("f");
2843        let d = desired(c.clone(), AudioFormat::Mp3);
2844        let plan = Plan {
2845            actions: vec![Action::Download {
2846                clip: c.clone(),
2847                lineage: LineageContext::own_root(&c),
2848                path: d.path.clone(),
2849                format: AudioFormat::Mp3,
2850            }],
2851        };
2852        let http = ScriptedHttp::new().route("f.mp3", Reply::ok(b"new-body".to_vec()));
2853        let fs = MemFs::new()
2854            .with_file("f.mp3", b"OLD-CONTENT".to_vec())
2855            .fail_write("f.mp3");
2856        let mut manifest = Manifest::new();
2857
2858        let outcome = run(
2859            &plan,
2860            &mut manifest,
2861            &[d],
2862            &http,
2863            &fs,
2864            &StubFfmpeg::flac(),
2865            &RecordingClock::new(),
2866            &ExecOptions::default(),
2867        );
2868
2869        assert_eq!(outcome.downloaded, 0);
2870        assert_eq!(outcome.failed(), 1);
2871        assert_eq!(fs.read_file("f.mp3").unwrap(), b"OLD-CONTENT");
2872        assert!(manifest.get("f").is_none());
2873    }
2874
2875    #[test]
2876    fn size_mismatch_after_write_is_a_failure() {
2877        let c = clip("g");
2878        let d = desired(c.clone(), AudioFormat::Mp3);
2879        let plan = Plan {
2880            actions: vec![Action::Download {
2881                clip: c.clone(),
2882                lineage: LineageContext::own_root(&c),
2883                path: d.path.clone(),
2884                format: AudioFormat::Mp3,
2885            }],
2886        };
2887        let http = ScriptedHttp::new().route("g.mp3", Reply::ok(b"body".to_vec()));
2888        let fs = MemFs::new().corrupt_write("g.mp3");
2889        let mut manifest = Manifest::new();
2890
2891        let outcome = run(
2892            &plan,
2893            &mut manifest,
2894            &[d],
2895            &http,
2896            &fs,
2897            &StubFfmpeg::flac(),
2898            &RecordingClock::new(),
2899            &ExecOptions::default(),
2900        );
2901
2902        assert_eq!(outcome.downloaded, 0);
2903        assert_eq!(outcome.failed(), 1);
2904        assert!(outcome.failures[0].reason.contains("expected"));
2905        assert!(manifest.get("g").is_none());
2906    }
2907
2908    // ── Reliability policy (SYNC-16/17) ─────────────────────────────
2909
2910    #[test]
2911    fn transient_failure_is_retried_then_skipped() {
2912        let c = clip("h");
2913        let d = desired(c.clone(), AudioFormat::Mp3);
2914        let plan = Plan {
2915            actions: vec![Action::Download {
2916                clip: c.clone(),
2917                lineage: LineageContext::own_root(&c),
2918                path: d.path.clone(),
2919                format: AudioFormat::Mp3,
2920            }],
2921        };
2922        let http = ScriptedHttp::new().route("h.mp3", Reply::status(500));
2923        let fs = MemFs::new();
2924        let clock = RecordingClock::new();
2925        let opts = ExecOptions {
2926            max_retries: 2,
2927            ..ExecOptions::default()
2928        };
2929        let mut manifest = Manifest::new();
2930
2931        let outcome = run(
2932            &plan,
2933            &mut manifest,
2934            &[d],
2935            &http,
2936            &fs,
2937            &StubFfmpeg::flac(),
2938            &clock,
2939            &opts,
2940        );
2941
2942        assert_eq!(outcome.downloaded, 0);
2943        assert_eq!(outcome.failed(), 1);
2944        assert_eq!(http.count("h.mp3"), 3);
2945        assert_eq!(clock.sleeps().len(), 2);
2946    }
2947
2948    #[test]
2949    fn truncated_download_is_retried_then_succeeds() {
2950        let c = clip("i");
2951        let d = desired(c.clone(), AudioFormat::Mp3);
2952        let plan = Plan {
2953            actions: vec![Action::Download {
2954                clip: c.clone(),
2955                lineage: LineageContext::own_root(&c),
2956                path: d.path.clone(),
2957                format: AudioFormat::Mp3,
2958            }],
2959        };
2960        let http = ScriptedHttp::new().route_seq(
2961            "i.mp3",
2962            vec![
2963                Reply::ok(b"short".to_vec()).with_content_length(999),
2964                Reply::ok(b"good-body".to_vec()),
2965            ],
2966        );
2967        let fs = MemFs::new();
2968        let clock = RecordingClock::new();
2969        let mut manifest = Manifest::new();
2970
2971        let outcome = run(
2972            &plan,
2973            &mut manifest,
2974            &[d],
2975            &http,
2976            &fs,
2977            &StubFfmpeg::flac(),
2978            &clock,
2979            &ExecOptions::default(),
2980        );
2981
2982        assert_eq!(outcome.downloaded, 1);
2983        assert_eq!(http.count("i.mp3"), 2);
2984        assert_eq!(clock.sleeps().len(), 1);
2985    }
2986
2987    #[test]
2988    fn rate_limit_backs_off_using_retry_after() {
2989        let c = clip("j");
2990        let d = desired(c.clone(), AudioFormat::Mp3);
2991        let plan = Plan {
2992            actions: vec![Action::Download {
2993                clip: c.clone(),
2994                lineage: LineageContext::own_root(&c),
2995                path: d.path.clone(),
2996                format: AudioFormat::Mp3,
2997            }],
2998        };
2999        let http = ScriptedHttp::new().route_seq(
3000            "j.mp3",
3001            vec![
3002                Reply::status(429).with_retry_after(7),
3003                Reply::ok(b"body".to_vec()),
3004            ],
3005        );
3006        let fs = MemFs::new();
3007        let clock = RecordingClock::new();
3008        let mut manifest = Manifest::new();
3009
3010        let outcome = run(
3011            &plan,
3012            &mut manifest,
3013            &[d],
3014            &http,
3015            &fs,
3016            &StubFfmpeg::flac(),
3017            &clock,
3018            &ExecOptions::default(),
3019        );
3020
3021        assert_eq!(outcome.downloaded, 1);
3022        assert_eq!(clock.sleeps(), vec![Duration::from_secs(7)]);
3023    }
3024
3025    #[test]
3026    fn auth_failure_aborts_the_run() {
3027        let c1 = clip("k1");
3028        let c2 = clip("k2");
3029        let d1 = desired(c1.clone(), AudioFormat::Flac);
3030        let d2 = desired(c2.clone(), AudioFormat::Flac);
3031        let plan = Plan {
3032            actions: vec![
3033                Action::Download {
3034                    clip: c1.clone(),
3035                    lineage: LineageContext::own_root(&c1),
3036                    path: d1.path.clone(),
3037                    format: AudioFormat::Flac,
3038                },
3039                Action::Download {
3040                    clip: c2.clone(),
3041                    lineage: LineageContext::own_root(&c2),
3042                    path: d2.path.clone(),
3043                    format: AudioFormat::Flac,
3044                },
3045            ],
3046        };
3047        // The authenticated WAV-render endpoint rejects auth even after a JWT
3048        // refresh: that is a bad token, so the whole run aborts rather than
3049        // hammering every clip. A CDN media rejection, by contrast, does not.
3050        let http = ScriptedHttp::new()
3051            .with_auth()
3052            .route("/wav_file/", Reply::status(401));
3053        let fs = MemFs::new();
3054        let mut manifest = Manifest::new();
3055
3056        let outcome = run(
3057            &plan,
3058            &mut manifest,
3059            &[d1, d2],
3060            &http,
3061            &fs,
3062            &StubFfmpeg::flac(),
3063            &RecordingClock::new(),
3064            &small_poll(),
3065        );
3066
3067        assert_eq!(outcome.status, RunStatus::AuthAborted);
3068        assert_eq!(outcome.failed(), 1);
3069        assert_eq!(outcome.failures[0].clip_id, "k1");
3070        assert_eq!(outcome.downloaded, 0);
3071    }
3072
3073    // ── Disk-full aborts the run (issue #17) ────────────────────────
3074
3075    #[test]
3076    fn disk_full_primary_write_aborts_the_run() {
3077        // Two MP3 downloads; the first write is out of space. That is systemic,
3078        // so the run aborts before the second is even attempted: exactly one
3079        // failure is recorded and its reason names the disk-full cause.
3080        let c1 = clip("d1");
3081        let c2 = clip("d2");
3082        let d1 = desired(c1.clone(), AudioFormat::Mp3);
3083        let d2 = desired(c2.clone(), AudioFormat::Mp3);
3084        let plan = Plan {
3085            actions: vec![
3086                Action::Download {
3087                    clip: c1.clone(),
3088                    lineage: LineageContext::own_root(&c1),
3089                    path: d1.path.clone(),
3090                    format: AudioFormat::Mp3,
3091                },
3092                Action::Download {
3093                    clip: c2.clone(),
3094                    lineage: LineageContext::own_root(&c2),
3095                    path: d2.path.clone(),
3096                    format: AudioFormat::Mp3,
3097                },
3098            ],
3099        };
3100        let http = ScriptedHttp::new()
3101            .route("d1.mp3", Reply::ok(b"body-1".to_vec()))
3102            .route("d2.mp3", Reply::ok(b"body-2".to_vec()));
3103        let fs = MemFs::new().fail_write_out_of_space("d1.mp3");
3104        let mut manifest = Manifest::new();
3105
3106        let outcome = run(
3107            &plan,
3108            &mut manifest,
3109            &[d1, d2],
3110            &http,
3111            &fs,
3112            &StubFfmpeg::flac(),
3113            &RecordingClock::new(),
3114            &ExecOptions::default(),
3115        );
3116
3117        assert_eq!(outcome.status, RunStatus::DiskFull);
3118        assert_eq!(outcome.failed(), 1);
3119        assert_eq!(outcome.failures[0].clip_id, "d1");
3120        assert!(outcome.failures[0].reason.contains("disk full"));
3121        assert_eq!(outcome.downloaded, 0);
3122        // The second clip was never fetched: the run aborted first.
3123        assert_eq!(http.count("d2.mp3"), 0);
3124        assert!(!fs.exists("d2.mp3"));
3125    }
3126
3127    #[test]
3128    fn disk_full_flac_transcode_aborts_the_run() {
3129        // The scratch disk fills during the FLAC re-encode; a WAV rendered, but
3130        // there is nowhere to stage the transcode, so the run aborts.
3131        let c1 = clip("d1");
3132        let c2 = clip("d2");
3133        let d1 = desired(c1.clone(), AudioFormat::Flac);
3134        let d2 = desired(c2.clone(), AudioFormat::Flac);
3135        let plan = Plan {
3136            actions: vec![
3137                Action::Download {
3138                    clip: c1.clone(),
3139                    lineage: LineageContext::own_root(&c1),
3140                    path: d1.path.clone(),
3141                    format: AudioFormat::Flac,
3142                },
3143                Action::Download {
3144                    clip: c2.clone(),
3145                    lineage: LineageContext::own_root(&c2),
3146                    path: d2.path.clone(),
3147                    format: AudioFormat::Flac,
3148                },
3149            ],
3150        };
3151        let http = ScriptedHttp::new()
3152            .with_auth()
3153            .route(
3154                "/wav_file/",
3155                Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/d1.wav"}"#),
3156            )
3157            .route(".wav", Reply::ok(b"wav".to_vec()));
3158        let fs = MemFs::new();
3159        let mut manifest = Manifest::new();
3160
3161        let outcome = run(
3162            &plan,
3163            &mut manifest,
3164            &[d1, d2],
3165            &http,
3166            &fs,
3167            &StubFfmpeg::out_of_space(),
3168            &RecordingClock::new(),
3169            &ExecOptions::default(),
3170        );
3171
3172        assert_eq!(outcome.status, RunStatus::DiskFull);
3173        assert_eq!(outcome.failed(), 1);
3174        assert_eq!(outcome.failures[0].clip_id, "d1");
3175        assert!(outcome.failures[0].reason.contains("disk full"));
3176        assert_eq!(outcome.downloaded, 0);
3177    }
3178
3179    #[test]
3180    fn disk_full_artifact_write_aborts_the_run() {
3181        // A sidecar write (not a primary download) also aborts on a full disk:
3182        // the owning audio is present, the cover fetch succeeds, but the sidecar
3183        // cannot be written.
3184        let mut manifest = Manifest::new();
3185        manifest.insert("a", entry("a.mp3", AudioFormat::Mp3));
3186        let plan = Plan {
3187            actions: vec![Action::WriteArtifact {
3188                kind: ArtifactKind::CoverJpg,
3189                path: "a/cover.jpg".to_owned(),
3190                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
3191                hash: "h1".to_owned(),
3192                owner_id: "a".to_owned(),
3193                content: None,
3194            }],
3195        };
3196        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"jpg-bytes".to_vec()));
3197        let fs = MemFs::new().fail_write_out_of_space("a/cover.jpg");
3198
3199        let outcome = run(
3200            &plan,
3201            &mut manifest,
3202            &[],
3203            &http,
3204            &fs,
3205            &StubFfmpeg::flac(),
3206            &RecordingClock::new(),
3207            &ExecOptions::default(),
3208        );
3209
3210        assert_eq!(outcome.status, RunStatus::DiskFull);
3211        assert_eq!(outcome.failed(), 1);
3212        assert!(outcome.failures[0].reason.contains("disk full"));
3213        assert_eq!(outcome.artifacts_written, 0);
3214        // The sidecar slot was never recorded: the write failed before it.
3215        assert_eq!(manifest.get("a").unwrap().cover_jpg, None);
3216    }
3217
3218    #[test]
3219    fn disk_full_leaves_the_failed_clips_manifest_entry_unchanged() {
3220        // write_verify fails before any manifest insert, so a re-download that
3221        // hits a full disk leaves the prior entry (and file) exactly as it was.
3222        let c = clip("m");
3223        let d = desired(c.clone(), AudioFormat::Mp3);
3224        let plan = Plan {
3225            actions: vec![Action::Download {
3226                clip: c.clone(),
3227                lineage: LineageContext::own_root(&c),
3228                path: d.path.clone(),
3229                format: AudioFormat::Mp3,
3230            }],
3231        };
3232        let http = ScriptedHttp::new().route("m.mp3", Reply::ok(b"new-body".to_vec()));
3233        let fs = MemFs::new()
3234            .with_file("m.mp3", b"OLD-CONTENT".to_vec())
3235            .fail_write_out_of_space("m.mp3");
3236        let mut manifest = Manifest::new();
3237        let before = entry("m.mp3", AudioFormat::Mp3);
3238        manifest.insert("m", before.clone());
3239
3240        let outcome = run(
3241            &plan,
3242            &mut manifest,
3243            &[d],
3244            &http,
3245            &fs,
3246            &StubFfmpeg::flac(),
3247            &RecordingClock::new(),
3248            &ExecOptions::default(),
3249        );
3250
3251        assert_eq!(outcome.status, RunStatus::DiskFull);
3252        assert_eq!(manifest.get("m"), Some(&before));
3253        assert_eq!(fs.read_file("m.mp3").unwrap(), b"OLD-CONTENT");
3254    }
3255
3256    #[test]
3257    fn cdn_download_rejection_skips_the_clip_without_aborting() {
3258        let c1 = clip("k1");
3259        let c2 = clip("k2");
3260        let d1 = desired(c1.clone(), AudioFormat::Mp3);
3261        let d2 = desired(c2.clone(), AudioFormat::Mp3);
3262        let plan = Plan {
3263            actions: vec![
3264                Action::Download {
3265                    clip: c1.clone(),
3266                    lineage: LineageContext::own_root(&c1),
3267                    path: d1.path.clone(),
3268                    format: AudioFormat::Mp3,
3269                },
3270                Action::Download {
3271                    clip: c2.clone(),
3272                    lineage: LineageContext::own_root(&c2),
3273                    path: d2.path.clone(),
3274                    format: AudioFormat::Mp3,
3275                },
3276            ],
3277        };
3278        // A CDN media fetch is unauthenticated, so a 403 is a per-asset
3279        // rejection (often transient), not a bad token: the clip is retried
3280        // then recorded and skipped, and the run carries on to the rest.
3281        let http = ScriptedHttp::new()
3282            .route("k1.mp3", Reply::status(403))
3283            .route("k2.mp3", Reply::ok(b"body".to_vec()));
3284        let fs = MemFs::new();
3285        let mut manifest = Manifest::new();
3286
3287        let outcome = run(
3288            &plan,
3289            &mut manifest,
3290            &[d1, d2],
3291            &http,
3292            &fs,
3293            &StubFfmpeg::flac(),
3294            &RecordingClock::new(),
3295            &ExecOptions::default(),
3296        );
3297
3298        assert_ne!(outcome.status, RunStatus::AuthAborted);
3299        assert_eq!(outcome.downloaded, 1);
3300        assert_eq!(outcome.failed(), 1);
3301        assert_eq!(outcome.failures[0].clip_id, "k1");
3302    }
3303
3304    #[test]
3305    fn one_clip_failure_does_not_abort_the_run() {
3306        let c1 = clip("l1");
3307        let c2 = clip("l2");
3308        let d1 = desired(c1.clone(), AudioFormat::Mp3);
3309        let d2 = desired(c2.clone(), AudioFormat::Mp3);
3310        let plan = Plan {
3311            actions: vec![
3312                Action::Download {
3313                    clip: c1.clone(),
3314                    lineage: LineageContext::own_root(&c1),
3315                    path: d1.path.clone(),
3316                    format: AudioFormat::Mp3,
3317                },
3318                Action::Download {
3319                    clip: c2.clone(),
3320                    lineage: LineageContext::own_root(&c2),
3321                    path: d2.path.clone(),
3322                    format: AudioFormat::Mp3,
3323                },
3324            ],
3325        };
3326        let http = ScriptedHttp::new()
3327            .route("l1.mp3", Reply::status(404))
3328            .route("l2.mp3", Reply::ok(b"body".to_vec()));
3329        let fs = MemFs::new();
3330        let mut manifest = Manifest::new();
3331
3332        let outcome = run(
3333            &plan,
3334            &mut manifest,
3335            &[d1, d2],
3336            &http,
3337            &fs,
3338            &StubFfmpeg::flac(),
3339            &RecordingClock::new(),
3340            &ExecOptions::default(),
3341        );
3342
3343        assert_eq!(outcome.status, RunStatus::Completed);
3344        assert_eq!(outcome.downloaded, 1);
3345        assert_eq!(outcome.failed(), 1);
3346        assert_eq!(outcome.failures[0].clip_id, "l1");
3347        assert!(fs.exists("l2.mp3"));
3348        assert!(manifest.get("l2").is_some());
3349        assert!(manifest.get("l1").is_none());
3350    }
3351
3352    // ── preserve marker (SYNC-8) ────────────────────────────────────
3353
3354    #[test]
3355    fn preserve_is_set_for_copy_held_and_private_clips() {
3356        let mut mirror = desired(clip("m1"), AudioFormat::Mp3);
3357        mirror.modes = vec![SourceMode::Mirror];
3358        let mut copy_held = desired(clip("m2"), AudioFormat::Mp3);
3359        copy_held.modes = vec![SourceMode::Mirror, SourceMode::Copy];
3360        let mut private = desired(clip("m3"), AudioFormat::Mp3);
3361        private.private = true;
3362
3363        let plan = Plan {
3364            actions: vec![
3365                Action::Download {
3366                    clip: mirror.clip.clone(),
3367                    lineage: LineageContext::own_root(&mirror.clip),
3368                    path: mirror.path.clone(),
3369                    format: AudioFormat::Mp3,
3370                },
3371                Action::Download {
3372                    clip: copy_held.clip.clone(),
3373                    lineage: LineageContext::own_root(&copy_held.clip),
3374                    path: copy_held.path.clone(),
3375                    format: AudioFormat::Mp3,
3376                },
3377                Action::Download {
3378                    clip: private.clip.clone(),
3379                    lineage: LineageContext::own_root(&private.clip),
3380                    path: private.path.clone(),
3381                    format: AudioFormat::Mp3,
3382                },
3383            ],
3384        };
3385        let http = ScriptedHttp::new()
3386            .route("m1.mp3", Reply::ok(b"a".to_vec()))
3387            .route("m2.mp3", Reply::ok(b"b".to_vec()))
3388            .route("m3.mp3", Reply::ok(b"c".to_vec()));
3389        let fs = MemFs::new();
3390        let mut manifest = Manifest::new();
3391
3392        let outcome = run(
3393            &plan,
3394            &mut manifest,
3395            &[mirror, copy_held, private],
3396            &http,
3397            &fs,
3398            &StubFfmpeg::flac(),
3399            &RecordingClock::new(),
3400            &ExecOptions::default(),
3401        );
3402
3403        assert_eq!(outcome.downloaded, 3);
3404        assert!(!manifest.get("m1").unwrap().preserve);
3405        assert!(manifest.get("m2").unwrap().preserve);
3406        assert!(manifest.get("m3").unwrap().preserve);
3407    }
3408
3409    // ── Reformat / Retag / Rename / Delete / Skip ───────────────────
3410
3411    #[test]
3412    fn reformat_writes_new_format_and_removes_old_file() {
3413        let c = clip("n");
3414        let d = desired(c.clone(), AudioFormat::Mp3);
3415        let plan = Plan {
3416            actions: vec![Action::Reformat {
3417                clip: c.clone(),
3418                path: "n.mp3".to_owned(),
3419                from_path: "n.flac".to_owned(),
3420                from: AudioFormat::Flac,
3421                to: AudioFormat::Mp3,
3422            }],
3423        };
3424        let http = ScriptedHttp::new().route("n.mp3", Reply::ok(b"body".to_vec()));
3425        let fs = MemFs::new().with_file("n.flac", b"OLD-FLAC".to_vec());
3426        let mut manifest = Manifest::new();
3427        manifest.insert("n", entry("n.flac", AudioFormat::Flac));
3428
3429        let outcome = run(
3430            &plan,
3431            &mut manifest,
3432            &[d],
3433            &http,
3434            &fs,
3435            &StubFfmpeg::flac(),
3436            &RecordingClock::new(),
3437            &ExecOptions::default(),
3438        );
3439
3440        assert_eq!(outcome.reformatted, 1);
3441        assert!(fs.exists("n.mp3"));
3442        assert!(!fs.exists("n.flac"));
3443        let updated = manifest.get("n").unwrap();
3444        assert_eq!(updated.path, "n.mp3");
3445        assert_eq!(updated.format, AudioFormat::Mp3);
3446        assert_eq!(updated.meta_hash, "m");
3447    }
3448
3449    #[test]
3450    fn retag_rewrites_file_and_updates_hashes() {
3451        let c = clip("o");
3452        let mut d = desired(c.clone(), AudioFormat::Mp3);
3453        d.meta_hash = "new".to_owned();
3454        d.art_hash = "new-art".to_owned();
3455        let existing = tag_mp3(
3456            b"audio",
3457            &TrackMetadata::from_clip(&c, &LineageContext::own_root(&c)),
3458            None,
3459            None,
3460        )
3461        .unwrap();
3462        let fs = MemFs::new().with_file("o.mp3", existing.clone());
3463        let mut manifest = Manifest::new();
3464        let mut start = entry("o.mp3", AudioFormat::Mp3);
3465        start.size = existing.len() as u64;
3466        manifest.insert("o", start);
3467        let plan = Plan {
3468            actions: vec![Action::Retag {
3469                clip: c.clone(),
3470                lineage: LineageContext::own_root(&c),
3471                path: "o.mp3".to_owned(),
3472            }],
3473        };
3474
3475        let outcome = run(
3476            &plan,
3477            &mut manifest,
3478            &[d],
3479            &ScriptedHttp::new(),
3480            &fs,
3481            &StubFfmpeg::flac(),
3482            &RecordingClock::new(),
3483            &ExecOptions::default(),
3484        );
3485
3486        assert_eq!(outcome.retagged, 1);
3487        let updated = manifest.get("o").unwrap();
3488        assert_eq!(updated.meta_hash, "new");
3489        assert_eq!(updated.art_hash, "new-art");
3490        assert_eq!(&fs.read_file("o.mp3").unwrap()[..3], b"ID3");
3491    }
3492
3493    #[test]
3494    fn rename_moves_file_and_updates_manifest_path() {
3495        let c = clip("p");
3496        let mut d = desired(c.clone(), AudioFormat::Mp3);
3497        d.path = "new/p.mp3".to_owned();
3498        let fs = MemFs::new().with_file("old/p.mp3", b"DATA".to_vec());
3499        let mut manifest = Manifest::new();
3500        manifest.insert("p", entry("old/p.mp3", AudioFormat::Mp3));
3501        let plan = Plan {
3502            actions: vec![Action::Rename {
3503                from: "old/p.mp3".to_owned(),
3504                to: "new/p.mp3".to_owned(),
3505            }],
3506        };
3507
3508        let outcome = run(
3509            &plan,
3510            &mut manifest,
3511            &[d],
3512            &ScriptedHttp::new(),
3513            &fs,
3514            &StubFfmpeg::flac(),
3515            &RecordingClock::new(),
3516            &ExecOptions::default(),
3517        );
3518
3519        assert_eq!(outcome.renamed, 1);
3520        assert!(fs.exists("new/p.mp3"));
3521        assert!(!fs.exists("old/p.mp3"));
3522        assert_eq!(manifest.get("p").unwrap().path, "new/p.mp3");
3523    }
3524
3525    #[test]
3526    fn disk_full_rename_aborts_the_run() {
3527        // A move onto a full disk is systemic like a full-disk write: the run
3528        // aborts with DiskFull and the source file is left untouched.
3529        let c = clip("p");
3530        let mut d = desired(c.clone(), AudioFormat::Mp3);
3531        d.path = "new/p.mp3".to_owned();
3532        let fs = MemFs::new()
3533            .with_file("old/p.mp3", b"DATA".to_vec())
3534            .fail_rename_out_of_space("new/p.mp3");
3535        let mut manifest = Manifest::new();
3536        manifest.insert("p", entry("old/p.mp3", AudioFormat::Mp3));
3537        let plan = Plan {
3538            actions: vec![Action::Rename {
3539                from: "old/p.mp3".to_owned(),
3540                to: "new/p.mp3".to_owned(),
3541            }],
3542        };
3543
3544        let outcome = run(
3545            &plan,
3546            &mut manifest,
3547            &[d],
3548            &ScriptedHttp::new(),
3549            &fs,
3550            &StubFfmpeg::flac(),
3551            &RecordingClock::new(),
3552            &ExecOptions::default(),
3553        );
3554
3555        assert_eq!(outcome.status, RunStatus::DiskFull);
3556        assert_eq!(outcome.renamed, 0);
3557        assert_eq!(outcome.failed(), 1);
3558        assert!(outcome.failures[0].reason.contains("disk full"));
3559        // The source is untouched: the move never happened.
3560        assert!(fs.exists("old/p.mp3"));
3561        assert!(!fs.exists("new/p.mp3"));
3562        assert_eq!(manifest.get("p").unwrap().path, "old/p.mp3");
3563    }
3564
3565    #[test]
3566    fn delete_removes_file_and_manifest_entry() {
3567        let fs = MemFs::new().with_file("q.mp3", b"DATA".to_vec());
3568        let mut manifest = Manifest::new();
3569        manifest.insert("q", entry("q.mp3", AudioFormat::Mp3));
3570        let plan = Plan {
3571            actions: vec![Action::Delete {
3572                path: "q.mp3".to_owned(),
3573                clip_id: "q".to_owned(),
3574            }],
3575        };
3576
3577        let outcome = run(
3578            &plan,
3579            &mut manifest,
3580            &[],
3581            &ScriptedHttp::new(),
3582            &fs,
3583            &StubFfmpeg::flac(),
3584            &RecordingClock::new(),
3585            &ExecOptions::default(),
3586        );
3587
3588        assert_eq!(outcome.deleted, 1);
3589        assert!(!fs.exists("q.mp3"));
3590        assert!(manifest.get("q").is_none());
3591    }
3592
3593    #[test]
3594    fn failed_delete_keeps_the_manifest_entry() {
3595        let fs = MemFs::new()
3596            .with_file("s.mp3", b"DATA".to_vec())
3597            .fail_remove("s.mp3");
3598        let mut manifest = Manifest::new();
3599        manifest.insert("s", entry("s.mp3", AudioFormat::Mp3));
3600        let plan = Plan {
3601            actions: vec![Action::Delete {
3602                path: "s.mp3".to_owned(),
3603                clip_id: "s".to_owned(),
3604            }],
3605        };
3606
3607        let outcome = run(
3608            &plan,
3609            &mut manifest,
3610            &[],
3611            &ScriptedHttp::new(),
3612            &fs,
3613            &StubFfmpeg::flac(),
3614            &RecordingClock::new(),
3615            &ExecOptions::default(),
3616        );
3617
3618        assert_eq!(outcome.deleted, 0);
3619        assert_eq!(outcome.failed(), 1);
3620        assert!(manifest.get("s").is_some());
3621        assert!(fs.exists("s.mp3"));
3622    }
3623
3624    #[test]
3625    fn skip_is_a_noop() {
3626        let mut manifest = Manifest::new();
3627        let plan = Plan {
3628            actions: vec![Action::Skip {
3629                clip_id: "r".to_owned(),
3630            }],
3631        };
3632        let outcome = run(
3633            &plan,
3634            &mut manifest,
3635            &[],
3636            &ScriptedHttp::new(),
3637            &MemFs::new(),
3638            &StubFfmpeg::flac(),
3639            &RecordingClock::new(),
3640            &ExecOptions::default(),
3641        );
3642        assert_eq!(outcome.skipped, 1);
3643        assert_eq!(outcome.failed(), 0);
3644    }
3645
3646    // ── Pure helpers ────────────────────────────────────────────────
3647
3648    #[test]
3649    fn header_helpers_parse_or_ignore() {
3650        let resp = HttpResponse {
3651            status: 200,
3652            headers: vec![("Content-Length".to_owned(), "42".to_owned())],
3653            body: Vec::new(),
3654        };
3655        assert_eq!(content_length(&resp), Some(42));
3656
3657        let bare = HttpResponse {
3658            status: 200,
3659            headers: Vec::new(),
3660            body: Vec::new(),
3661        };
3662        assert_eq!(content_length(&bare), None);
3663    }
3664
3665    #[test]
3666    fn preserve_rule_covers_copy_and_private() {
3667        let base = desired(clip("x"), AudioFormat::Mp3);
3668        assert!(!preserve_for(&base));
3669        let mut copy_held = base.clone();
3670        copy_held.modes = vec![SourceMode::Copy];
3671        assert!(preserve_for(&copy_held));
3672        let mut private = base.clone();
3673        private.private = true;
3674        assert!(preserve_for(&private));
3675    }
3676
3677    fn fs_new() -> MemFs {
3678        MemFs::new()
3679    }
3680
3681    // ── Skip refreshes the preserve marker (SYNC-8 cross-run) ────────
3682
3683    #[test]
3684    fn skip_sets_preserve_when_a_clip_becomes_copy_held() {
3685        let c = clip("s1");
3686        let mut d = desired(c.clone(), AudioFormat::Mp3);
3687        d.modes = vec![SourceMode::Copy];
3688        let plan = Plan {
3689            actions: vec![Action::Skip {
3690                clip_id: "s1".to_owned(),
3691            }],
3692        };
3693        let mut manifest = Manifest::new();
3694        manifest.insert("s1".to_owned(), entry("s1.mp3", AudioFormat::Mp3));
3695        assert!(!manifest.get("s1").unwrap().preserve);
3696
3697        let outcome = run(
3698            &plan,
3699            &mut manifest,
3700            &[d],
3701            &ScriptedHttp::new(),
3702            &fs_new(),
3703            &StubFfmpeg::flac(),
3704            &RecordingClock::new(),
3705            &ExecOptions::default(),
3706        );
3707
3708        assert_eq!(outcome.skipped, 1);
3709        assert!(
3710            manifest.get("s1").unwrap().preserve,
3711            "a copy-held skip must mark the entry preserved"
3712        );
3713    }
3714
3715    #[test]
3716    fn skip_clears_stale_preserve_when_a_clip_returns_to_mirror_only() {
3717        let c = clip("s2");
3718        let d = desired(c.clone(), AudioFormat::Mp3);
3719        let plan = Plan {
3720            actions: vec![Action::Skip {
3721                clip_id: "s2".to_owned(),
3722            }],
3723        };
3724        let mut manifest = Manifest::new();
3725        let mut stale = entry("s2.mp3", AudioFormat::Mp3);
3726        stale.preserve = true;
3727        manifest.insert("s2".to_owned(), stale);
3728
3729        run(
3730            &plan,
3731            &mut manifest,
3732            &[d],
3733            &ScriptedHttp::new(),
3734            &fs_new(),
3735            &StubFfmpeg::flac(),
3736            &RecordingClock::new(),
3737            &ExecOptions::default(),
3738        );
3739
3740        assert!(
3741            !manifest.get("s2").unwrap().preserve,
3742            "a mirror-only skip must clear a stale preserve marker"
3743        );
3744    }
3745
3746    #[test]
3747    fn flac_render_retries_a_rate_limited_wav_lookup() {
3748        let c = clip("rl");
3749        let d = desired(c.clone(), AudioFormat::Flac);
3750        let plan = Plan {
3751            actions: vec![Action::Download {
3752                clip: c.clone(),
3753                lineage: LineageContext::own_root(&c),
3754                path: d.path.clone(),
3755                format: AudioFormat::Flac,
3756            }],
3757        };
3758        let http = ScriptedHttp::new()
3759            .with_auth()
3760            .route_seq(
3761                "/wav_file/",
3762                vec![
3763                    Reply::status(429),
3764                    Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/rl.wav"}"#),
3765                ],
3766            )
3767            .route("rl.wav", Reply::ok(b"wav".to_vec()));
3768        let clock = RecordingClock::new();
3769        let mut manifest = Manifest::new();
3770
3771        let outcome = run(
3772            &plan,
3773            &mut manifest,
3774            &[d],
3775            &http,
3776            &fs_new(),
3777            &StubFfmpeg::flac(),
3778            &clock,
3779            &small_poll(),
3780        );
3781
3782        assert_eq!(outcome.downloaded, 1);
3783        assert_eq!(outcome.failed(), 0);
3784        // The render was ready on retry, so no fresh convert_wav was needed.
3785        assert_eq!(http.count("/convert_wav/"), 0);
3786        // One transient backoff (1s base), not the 5s poll interval.
3787        assert_eq!(clock.sleeps(), vec![Duration::from_secs(1)]);
3788    }
3789
3790    // ── Phase 6: artifact actions ───────────────────────────────────
3791
3792    #[test]
3793    fn write_artifact_fetches_writes_and_updates_manifest() {
3794        // The owning entry exists (its audio was kept this run); WriteArtifact
3795        // fetches the source, writes the sidecar, and records it on the entry.
3796        let mut manifest = Manifest::new();
3797        manifest.insert("a", entry("a.mp3", AudioFormat::Mp3));
3798        let plan = Plan {
3799            actions: vec![Action::WriteArtifact {
3800                kind: ArtifactKind::CoverJpg,
3801                path: "a/cover.jpg".to_owned(),
3802                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
3803                hash: "h1".to_owned(),
3804                owner_id: "a".to_owned(),
3805                content: None,
3806            }],
3807        };
3808        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"jpg-bytes".to_vec()));
3809        let fs = MemFs::new();
3810
3811        let outcome = run(
3812            &plan,
3813            &mut manifest,
3814            &[],
3815            &http,
3816            &fs,
3817            &StubFfmpeg::flac(),
3818            &RecordingClock::new(),
3819            &ExecOptions::default(),
3820        );
3821
3822        assert_eq!(outcome.artifacts_written, 1);
3823        assert_eq!(outcome.failed(), 0);
3824        assert_eq!(outcome.status, RunStatus::Completed);
3825        assert_eq!(fs.read_file("a/cover.jpg").unwrap(), b"jpg-bytes");
3826        assert_eq!(
3827            manifest.get("a").unwrap().cover_jpg,
3828            Some(ArtifactState {
3829                path: "a/cover.jpg".to_owned(),
3830                hash: "h1".to_owned(),
3831            })
3832        );
3833    }
3834
3835    #[test]
3836    fn write_text_sidecar_records_slot_with_no_network_fetch() {
3837        // A generated text sidecar carries its body inline, so it is written
3838        // verbatim with NO HTTP fetch and the details slot records its state.
3839        let mut manifest = Manifest::new();
3840        manifest.insert("a", entry("a.mp3", AudioFormat::Mp3));
3841        let plan = Plan {
3842            actions: vec![Action::WriteArtifact {
3843                kind: ArtifactKind::DetailsTxt,
3844                path: "a.details.txt".to_owned(),
3845                source_url: String::new(),
3846                hash: "dh".to_owned(),
3847                owner_id: "a".to_owned(),
3848                content: Some("Title: A\n".to_owned()),
3849            }],
3850        };
3851        // An empty HTTP script: any fetch would fail, proving none happens.
3852        let http = ScriptedHttp::new();
3853        let fs = MemFs::new();
3854
3855        let outcome = run(
3856            &plan,
3857            &mut manifest,
3858            &[],
3859            &http,
3860            &fs,
3861            &StubFfmpeg::flac(),
3862            &RecordingClock::new(),
3863            &ExecOptions::default(),
3864        );
3865
3866        assert_eq!(outcome.artifacts_written, 1);
3867        assert_eq!(outcome.failed(), 0);
3868        assert_eq!(fs.read_file("a.details.txt").unwrap(), b"Title: A\n");
3869        assert_eq!(
3870            manifest.get("a").unwrap().details_txt,
3871            Some(ArtifactState {
3872                path: "a.details.txt".to_owned(),
3873                hash: "dh".to_owned(),
3874            })
3875        );
3876    }
3877
3878    #[test]
3879    fn write_lyrics_sidecar_relocation_removes_old_file() {
3880        // The audio moved, so the lyrics sidecar is re-emitted at the new path;
3881        // the executor writes the new file and prunes the stale one.
3882        let mut manifest = Manifest::new();
3883        let mut e = entry("old/a.flac", AudioFormat::Flac);
3884        e.lyrics_txt = Some(ArtifactState {
3885            path: "old/a.lyrics.txt".to_owned(),
3886            hash: "lh".to_owned(),
3887        });
3888        manifest.insert("a", e);
3889        let fs = MemFs::new()
3890            .with_file("old/a.flac", b"AUDIO".to_vec())
3891            .with_file("old/a.lyrics.txt", b"old words\n".to_vec());
3892        let plan = Plan {
3893            actions: vec![Action::WriteArtifact {
3894                kind: ArtifactKind::LyricsTxt,
3895                path: "new/a.lyrics.txt".to_owned(),
3896                source_url: String::new(),
3897                hash: "lh".to_owned(),
3898                owner_id: "a".to_owned(),
3899                content: Some("new words\n".to_owned()),
3900            }],
3901        };
3902
3903        let outcome = run(
3904            &plan,
3905            &mut manifest,
3906            &[],
3907            &ScriptedHttp::new(),
3908            &fs,
3909            &StubFfmpeg::flac(),
3910            &RecordingClock::new(),
3911            &ExecOptions::default(),
3912        );
3913
3914        assert_eq!(outcome.failed(), 0);
3915        assert_eq!(fs.read_file("new/a.lyrics.txt").unwrap(), b"new words\n");
3916        assert!(!fs.exists("old/a.lyrics.txt"));
3917        assert_eq!(
3918            manifest.get("a").unwrap().lyrics_txt.as_ref().unwrap().path,
3919            "new/a.lyrics.txt"
3920        );
3921    }
3922
3923    #[test]
3924    fn sidecar_path_swap_never_deletes_a_file_written_this_run() {
3925        // Two clips swap sidecar paths in one run (A: x -> y while B: y -> x).
3926        // Each write's inline old-path cleanup must skip a path another action
3927        // writes this run, or the second write would delete the first's freshly
3928        // written file (issue #76). The guard is kind-agnostic; lyrics stands in
3929        // for every sidecar, including the .mp4 video.
3930        let mut manifest = Manifest::new();
3931        let mut a = entry("a.flac", AudioFormat::Flac);
3932        a.lyrics_txt = Some(ArtifactState {
3933            path: "x.lyrics.txt".to_owned(),
3934            hash: "ah".to_owned(),
3935        });
3936        manifest.insert("a", a);
3937        let mut b = entry("b.flac", AudioFormat::Flac);
3938        b.lyrics_txt = Some(ArtifactState {
3939            path: "y.lyrics.txt".to_owned(),
3940            hash: "bh".to_owned(),
3941        });
3942        manifest.insert("b", b);
3943        let fs = MemFs::new()
3944            .with_file("a.flac", b"A".to_vec())
3945            .with_file("b.flac", b"B".to_vec())
3946            .with_file("x.lyrics.txt", b"A words\n".to_vec())
3947            .with_file("y.lyrics.txt", b"B words\n".to_vec());
3948        // A moves its sidecar x -> y; B moves its sidecar y -> x (the swap).
3949        let plan = Plan {
3950            actions: vec![
3951                Action::WriteArtifact {
3952                    kind: ArtifactKind::LyricsTxt,
3953                    path: "y.lyrics.txt".to_owned(),
3954                    source_url: String::new(),
3955                    hash: "ah".to_owned(),
3956                    owner_id: "a".to_owned(),
3957                    content: Some("A words\n".to_owned()),
3958                },
3959                Action::WriteArtifact {
3960                    kind: ArtifactKind::LyricsTxt,
3961                    path: "x.lyrics.txt".to_owned(),
3962                    source_url: String::new(),
3963                    hash: "bh".to_owned(),
3964                    owner_id: "b".to_owned(),
3965                    content: Some("B words\n".to_owned()),
3966                },
3967            ],
3968        };
3969
3970        let outcome = run(
3971            &plan,
3972            &mut manifest,
3973            &[],
3974            &ScriptedHttp::new(),
3975            &fs,
3976            &StubFfmpeg::flac(),
3977            &RecordingClock::new(),
3978            &ExecOptions::default(),
3979        );
3980
3981        assert_eq!(outcome.failed(), 0);
3982        // Both freshly written files survive; neither cleanup clobbered the other.
3983        assert_eq!(fs.read_file("y.lyrics.txt").unwrap(), b"A words\n");
3984        assert_eq!(fs.read_file("x.lyrics.txt").unwrap(), b"B words\n");
3985        assert_eq!(
3986            manifest.get("a").unwrap().lyrics_txt.as_ref().unwrap().path,
3987            "y.lyrics.txt"
3988        );
3989        assert_eq!(
3990            manifest.get("b").unwrap().lyrics_txt.as_ref().unwrap().path,
3991            "x.lyrics.txt"
3992        );
3993    }
3994
3995    #[test]
3996    fn old_sidecar_kept_when_another_clip_still_references_it() {
3997        // A prior failed swap can leave two clips pointing at one path (A -> y and
3998        // B -> y). When B now moves y -> x, its cleanup must not delete y, which is
3999        // still A's live file (#76). tracked_paths counts two references to y, so
4000        // the removal is skipped even though y is not a write target this run.
4001        let mut manifest = Manifest::new();
4002        let mut a = entry("a.flac", AudioFormat::Flac);
4003        a.lyrics_txt = Some(ArtifactState {
4004            path: "y.lyrics.txt".to_owned(),
4005            hash: "ah".to_owned(),
4006        });
4007        manifest.insert("a", a);
4008        let mut b = entry("b.flac", AudioFormat::Flac);
4009        b.lyrics_txt = Some(ArtifactState {
4010            path: "y.lyrics.txt".to_owned(),
4011            hash: "bh".to_owned(),
4012        });
4013        manifest.insert("b", b);
4014        let fs = MemFs::new()
4015            .with_file("a.flac", b"A".to_vec())
4016            .with_file("b.flac", b"B".to_vec())
4017            .with_file("y.lyrics.txt", b"A words\n".to_vec());
4018        // Only B moves this run: y -> x. A is stable, so y is not a write target;
4019        // the tracked-reference count is what protects A's file.
4020        let plan = Plan {
4021            actions: vec![Action::WriteArtifact {
4022                kind: ArtifactKind::LyricsTxt,
4023                path: "x.lyrics.txt".to_owned(),
4024                source_url: String::new(),
4025                hash: "bh".to_owned(),
4026                owner_id: "b".to_owned(),
4027                content: Some("B words\n".to_owned()),
4028            }],
4029        };
4030
4031        let outcome = run(
4032            &plan,
4033            &mut manifest,
4034            &[],
4035            &ScriptedHttp::new(),
4036            &fs,
4037            &StubFfmpeg::flac(),
4038            &RecordingClock::new(),
4039            &ExecOptions::default(),
4040        );
4041
4042        assert_eq!(outcome.failed(), 0);
4043        assert!(
4044            fs.exists("y.lyrics.txt"),
4045            "A's live sidecar must not be deleted"
4046        );
4047        assert_eq!(fs.read_file("x.lyrics.txt").unwrap(), b"B words\n");
4048    }
4049
4050    #[test]
4051    fn shared_old_path_is_reclaimed_when_every_referencing_clip_moves_away() {
4052        // Two clips share one path (A -> s and B -> s, from a prior failed swap).
4053        // When BOTH move away this run, the path is no longer live, so the last
4054        // mover must reclaim it: it is neither kept as an orphan nor deleted while
4055        // still referenced. The dynamic reference count drops to zero only after
4056        // both moves, so exactly the final cleanup removes it (#76).
4057        let mut manifest = Manifest::new();
4058        let mut a = entry("a.flac", AudioFormat::Flac);
4059        a.lyrics_txt = Some(ArtifactState {
4060            path: "s.lyrics.txt".to_owned(),
4061            hash: "ah".to_owned(),
4062        });
4063        manifest.insert("a", a);
4064        let mut b = entry("b.flac", AudioFormat::Flac);
4065        b.lyrics_txt = Some(ArtifactState {
4066            path: "s.lyrics.txt".to_owned(),
4067            hash: "bh".to_owned(),
4068        });
4069        manifest.insert("b", b);
4070        let fs = MemFs::new()
4071            .with_file("a.flac", b"A".to_vec())
4072            .with_file("b.flac", b"B".to_vec())
4073            .with_file("s.lyrics.txt", b"shared\n".to_vec());
4074        let plan = Plan {
4075            actions: vec![
4076                Action::WriteArtifact {
4077                    kind: ArtifactKind::LyricsTxt,
4078                    path: "pa.lyrics.txt".to_owned(),
4079                    source_url: String::new(),
4080                    hash: "ah".to_owned(),
4081                    owner_id: "a".to_owned(),
4082                    content: Some("A words\n".to_owned()),
4083                },
4084                Action::WriteArtifact {
4085                    kind: ArtifactKind::LyricsTxt,
4086                    path: "pb.lyrics.txt".to_owned(),
4087                    source_url: String::new(),
4088                    hash: "bh".to_owned(),
4089                    owner_id: "b".to_owned(),
4090                    content: Some("B words\n".to_owned()),
4091                },
4092            ],
4093        };
4094
4095        let outcome = run(
4096            &plan,
4097            &mut manifest,
4098            &[],
4099            &ScriptedHttp::new(),
4100            &fs,
4101            &StubFfmpeg::flac(),
4102            &RecordingClock::new(),
4103            &ExecOptions::default(),
4104        );
4105
4106        assert_eq!(outcome.failed(), 0);
4107        assert_eq!(fs.read_file("pa.lyrics.txt").unwrap(), b"A words\n");
4108        assert_eq!(fs.read_file("pb.lyrics.txt").unwrap(), b"B words\n");
4109        assert!(
4110            !fs.exists("s.lyrics.txt"),
4111            "the vacated shared path must be reclaimed, not orphaned"
4112        );
4113    }
4114
4115    #[test]
4116    fn write_text_sidecar_skipped_when_owner_audio_absent() {
4117        // A text sidecar for a clip with no manifest entry (its audio download
4118        // failed) must be skipped, never writing an untracked file.
4119        let plan = Plan {
4120            actions: vec![Action::WriteArtifact {
4121                kind: ArtifactKind::DetailsTxt,
4122                path: "gone.details.txt".to_owned(),
4123                source_url: String::new(),
4124                hash: "dh".to_owned(),
4125                owner_id: "gone".to_owned(),
4126                content: Some("Title: Gone\n".to_owned()),
4127            }],
4128        };
4129        let fs = MemFs::new();
4130        let mut manifest = Manifest::new();
4131
4132        let outcome = run(
4133            &plan,
4134            &mut manifest,
4135            &[],
4136            &ScriptedHttp::new(),
4137            &fs,
4138            &StubFfmpeg::flac(),
4139            &RecordingClock::new(),
4140            &ExecOptions::default(),
4141        );
4142
4143        assert_eq!(outcome.artifacts_written, 0);
4144        assert_eq!(outcome.skipped, 1);
4145        assert!(!fs.exists("gone.details.txt"));
4146        assert!(manifest.get("gone").is_none());
4147    }
4148
4149    #[test]
4150    fn delete_artifact_removes_file_and_clears_slot() {
4151        let fs = MemFs::new().with_file("a/cover.jpg", b"jpg".to_vec());
4152        let mut manifest = Manifest::new();
4153        let mut e = entry("a.mp3", AudioFormat::Mp3);
4154        e.cover_jpg = Some(ArtifactState {
4155            path: "a/cover.jpg".to_owned(),
4156            hash: "h1".to_owned(),
4157        });
4158        manifest.insert("a", e);
4159        let plan = Plan {
4160            actions: vec![Action::DeleteArtifact {
4161                kind: ArtifactKind::CoverJpg,
4162                path: "a/cover.jpg".to_owned(),
4163                owner_id: "a".to_owned(),
4164            }],
4165        };
4166
4167        let outcome = run(
4168            &plan,
4169            &mut manifest,
4170            &[],
4171            &ScriptedHttp::new(),
4172            &fs,
4173            &StubFfmpeg::flac(),
4174            &RecordingClock::new(),
4175            &ExecOptions::default(),
4176        );
4177
4178        assert_eq!(outcome.artifacts_deleted, 1);
4179        assert!(!fs.exists("a/cover.jpg"));
4180        assert_eq!(manifest.get("a").unwrap().cover_jpg, None);
4181    }
4182
4183    #[test]
4184    fn delete_artifact_tolerates_already_absent_file() {
4185        // `remove` is idempotent, so co-deleting a sidecar that is already gone
4186        // is not a failure.
4187        let mut manifest = Manifest::new();
4188        let mut e = entry("a.mp3", AudioFormat::Mp3);
4189        e.cover_jpg = Some(ArtifactState {
4190            path: "a/cover.jpg".to_owned(),
4191            hash: "h1".to_owned(),
4192        });
4193        manifest.insert("a", e);
4194        let plan = Plan {
4195            actions: vec![Action::DeleteArtifact {
4196                kind: ArtifactKind::CoverJpg,
4197                path: "a/cover.jpg".to_owned(),
4198                owner_id: "a".to_owned(),
4199            }],
4200        };
4201
4202        let outcome = run(
4203            &plan,
4204            &mut manifest,
4205            &[],
4206            &ScriptedHttp::new(),
4207            &MemFs::new(),
4208            &StubFfmpeg::flac(),
4209            &RecordingClock::new(),
4210            &ExecOptions::default(),
4211        );
4212
4213        assert_eq!(outcome.artifacts_deleted, 1);
4214        assert_eq!(outcome.failed(), 0);
4215        assert_eq!(manifest.get("a").unwrap().cover_jpg, None);
4216    }
4217
4218    #[test]
4219    fn write_artifact_http_failure_is_a_per_clip_failure_not_a_run_abort() {
4220        // A permanent 404 on one sidecar fetch is recorded as a per-clip failure;
4221        // the run continues and the following WriteArtifact still succeeds.
4222        let mut manifest = Manifest::new();
4223        manifest.insert("a", entry("a.mp3", AudioFormat::Mp3));
4224        manifest.insert("b", entry("b.mp3", AudioFormat::Mp3));
4225        let plan = Plan {
4226            actions: vec![
4227                Action::WriteArtifact {
4228                    kind: ArtifactKind::CoverJpg,
4229                    path: "a/cover.jpg".to_owned(),
4230                    source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
4231                    hash: "h1".to_owned(),
4232                    owner_id: "a".to_owned(),
4233                    content: None,
4234                },
4235                Action::WriteArtifact {
4236                    kind: ArtifactKind::CoverJpg,
4237                    path: "b/cover.jpg".to_owned(),
4238                    source_url: "https://art.suno.ai/b/large.jpg".to_owned(),
4239                    hash: "h2".to_owned(),
4240                    owner_id: "b".to_owned(),
4241                    content: None,
4242                },
4243            ],
4244        };
4245        let http = ScriptedHttp::new()
4246            .route("a/large.jpg", Reply::status(404))
4247            .route("b/large.jpg", Reply::ok(b"jpg-b".to_vec()));
4248        let fs = MemFs::new();
4249
4250        let outcome = run(
4251            &plan,
4252            &mut manifest,
4253            &[],
4254            &http,
4255            &fs,
4256            &StubFfmpeg::flac(),
4257            &RecordingClock::new(),
4258            &ExecOptions::default(),
4259        );
4260
4261        assert_eq!(outcome.status, RunStatus::Completed);
4262        assert_eq!(outcome.failed(), 1);
4263        assert_eq!(outcome.failures[0].clip_id, "a");
4264        assert_eq!(outcome.artifacts_written, 1);
4265        // The failed sidecar left no file and no manifest record.
4266        assert!(!fs.exists("a/cover.jpg"));
4267        assert_eq!(manifest.get("a").unwrap().cover_jpg, None);
4268        // The following sidecar was written and recorded.
4269        assert_eq!(fs.read_file("b/cover.jpg").unwrap(), b"jpg-b");
4270        assert!(manifest.get("b").unwrap().cover_jpg.is_some());
4271    }
4272
4273    #[test]
4274    fn stranded_old_sidecar_removed_when_colliding_writer_fails() {
4275        // #142: clip A moves its cover shared -> a/cover.jpg (fetch succeeds);
4276        // clip B is planned to write the vacated `shared` path but its fetch
4277        // fails. The old-path cleanup is gated on COMMITTED writes, not planned
4278        // ones, so B's failed write no longer protects the stale file: A's old
4279        // `shared` copy is removed rather than left as an untracked orphan.
4280        let mut manifest = Manifest::new();
4281        let mut a = entry("a.mp3", AudioFormat::Mp3);
4282        a.cover_jpg = Some(ArtifactState {
4283            path: "shared/cover.jpg".to_owned(),
4284            hash: "ha".to_owned(),
4285        });
4286        manifest.insert("a", a);
4287        manifest.insert("b", entry("b.mp3", AudioFormat::Mp3));
4288        let fs = MemFs::new().with_file("shared/cover.jpg", b"old-shared".to_vec());
4289        let plan = Plan {
4290            actions: vec![
4291                Action::WriteArtifact {
4292                    kind: ArtifactKind::CoverJpg,
4293                    path: "a/cover.jpg".to_owned(),
4294                    source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
4295                    hash: "ha".to_owned(),
4296                    owner_id: "a".to_owned(),
4297                    content: None,
4298                },
4299                Action::WriteArtifact {
4300                    kind: ArtifactKind::CoverJpg,
4301                    path: "shared/cover.jpg".to_owned(),
4302                    source_url: "https://art.suno.ai/b/large.jpg".to_owned(),
4303                    hash: "hb".to_owned(),
4304                    owner_id: "b".to_owned(),
4305                    content: None,
4306                },
4307            ],
4308        };
4309        let http = ScriptedHttp::new()
4310            .route("a/large.jpg", Reply::ok(b"jpg-a".to_vec()))
4311            .route("b/large.jpg", Reply::status(404));
4312
4313        let outcome = run(
4314            &plan,
4315            &mut manifest,
4316            &[],
4317            &http,
4318            &fs,
4319            &StubFfmpeg::flac(),
4320            &RecordingClock::new(),
4321            &ExecOptions::default(),
4322        );
4323
4324        assert_eq!(outcome.failed(), 1);
4325        assert_eq!(outcome.failures[0].clip_id, "b");
4326        // A's move committed; the vacated file is gone, not an orphan.
4327        assert_eq!(fs.read_file("a/cover.jpg").unwrap(), b"jpg-a");
4328        assert!(
4329            !fs.exists("shared/cover.jpg"),
4330            "the vacated file must be removed once the colliding writer failed"
4331        );
4332        assert_eq!(
4333            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
4334            "a/cover.jpg"
4335        );
4336    }
4337
4338    #[test]
4339    fn committed_write_at_old_path_is_preserved() {
4340        // #142: clip B writes `shared` and commits BEFORE clip A vacates it
4341        // (A moves shared -> a/cover.jpg). A's cleanup sees `shared` in the
4342        // committed set and keeps B's freshly written file rather than deleting
4343        // it. This is the successful-collision case the guard must still protect.
4344        let mut manifest = Manifest::new();
4345        let mut a = entry("a.mp3", AudioFormat::Mp3);
4346        a.cover_jpg = Some(ArtifactState {
4347            path: "shared/cover.jpg".to_owned(),
4348            hash: "ha".to_owned(),
4349        });
4350        manifest.insert("a", a);
4351        manifest.insert("b", entry("b.mp3", AudioFormat::Mp3));
4352        let fs = MemFs::new().with_file("shared/cover.jpg", b"old-shared".to_vec());
4353        let plan = Plan {
4354            actions: vec![
4355                Action::WriteArtifact {
4356                    kind: ArtifactKind::CoverJpg,
4357                    path: "shared/cover.jpg".to_owned(),
4358                    source_url: "https://art.suno.ai/b/large.jpg".to_owned(),
4359                    hash: "hb".to_owned(),
4360                    owner_id: "b".to_owned(),
4361                    content: None,
4362                },
4363                Action::WriteArtifact {
4364                    kind: ArtifactKind::CoverJpg,
4365                    path: "a/cover.jpg".to_owned(),
4366                    source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
4367                    hash: "ha".to_owned(),
4368                    owner_id: "a".to_owned(),
4369                    content: None,
4370                },
4371            ],
4372        };
4373        let http = ScriptedHttp::new()
4374            .route("b/large.jpg", Reply::ok(b"jpg-b".to_vec()))
4375            .route("a/large.jpg", Reply::ok(b"jpg-a".to_vec()));
4376
4377        let outcome = run(
4378            &plan,
4379            &mut manifest,
4380            &[],
4381            &http,
4382            &fs,
4383            &StubFfmpeg::flac(),
4384            &RecordingClock::new(),
4385            &ExecOptions::default(),
4386        );
4387
4388        assert_eq!(outcome.failed(), 0);
4389        // B's committed write survives A's subsequent move; both files are present.
4390        assert_eq!(fs.read_file("shared/cover.jpg").unwrap(), b"jpg-b");
4391        assert_eq!(fs.read_file("a/cover.jpg").unwrap(), b"jpg-a");
4392        assert_eq!(
4393            manifest.get("b").unwrap().cover_jpg.as_ref().unwrap().path,
4394            "shared/cover.jpg"
4395        );
4396        assert_eq!(
4397            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
4398            "a/cover.jpg"
4399        );
4400    }
4401
4402    #[test]
4403    fn cover_move_renames_without_fetching() {
4404        // #141: a MoveArtifact relocates the cover with a local rename. The
4405        // ScriptedHttp has no route, so any fetch would fail the run; a clean
4406        // outcome proves the bytes were renamed, not re-downloaded.
4407        let mut manifest = Manifest::new();
4408        let mut e = entry("a.mp3", AudioFormat::Mp3);
4409        e.cover_jpg = Some(ArtifactState {
4410            path: "old/cover.jpg".to_owned(),
4411            hash: "h".to_owned(),
4412        });
4413        manifest.insert("a", e);
4414        let fs = MemFs::new().with_file("old/cover.jpg", b"JPGBYTES".to_vec());
4415        let plan = Plan {
4416            actions: vec![Action::MoveArtifact {
4417                kind: ArtifactKind::CoverJpg,
4418                from: "old/cover.jpg".to_owned(),
4419                to: "new/cover.jpg".to_owned(),
4420                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
4421                hash: "h".to_owned(),
4422                owner_id: "a".to_owned(),
4423            }],
4424        };
4425
4426        let outcome = run(
4427            &plan,
4428            &mut manifest,
4429            &[],
4430            &ScriptedHttp::new(),
4431            &fs,
4432            &StubFfmpeg::flac(),
4433            &RecordingClock::new(),
4434            &ExecOptions::default(),
4435        );
4436
4437        assert_eq!(outcome.failed(), 0);
4438        assert_eq!(outcome.renamed, 1, "counted as a rename, not a write");
4439        // Renamed in place: the new path carries the ORIGINAL bytes, old is gone.
4440        assert_eq!(fs.read_file("new/cover.jpg").unwrap(), b"JPGBYTES");
4441        assert!(!fs.exists("old/cover.jpg"));
4442        assert_eq!(
4443            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
4444            "new/cover.jpg"
4445        );
4446    }
4447
4448    #[test]
4449    fn cover_move_falls_back_to_fetch_when_old_file_missing() {
4450        // #141: the old file vanished before commit, so the rename fails and the
4451        // executor fetches fresh bytes at the new path rather than failing.
4452        let mut manifest = Manifest::new();
4453        let mut e = entry("a.mp3", AudioFormat::Mp3);
4454        e.cover_jpg = Some(ArtifactState {
4455            path: "old/cover.jpg".to_owned(),
4456            hash: "h".to_owned(),
4457        });
4458        manifest.insert("a", e);
4459        let fs = MemFs::new(); // old/cover.jpg is absent.
4460        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"FETCHED".to_vec()));
4461        let plan = Plan {
4462            actions: vec![Action::MoveArtifact {
4463                kind: ArtifactKind::CoverJpg,
4464                from: "old/cover.jpg".to_owned(),
4465                to: "new/cover.jpg".to_owned(),
4466                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
4467                hash: "h".to_owned(),
4468                owner_id: "a".to_owned(),
4469            }],
4470        };
4471
4472        let outcome = run(
4473            &plan,
4474            &mut manifest,
4475            &[],
4476            &http,
4477            &fs,
4478            &StubFfmpeg::flac(),
4479            &RecordingClock::new(),
4480            &ExecOptions::default(),
4481        );
4482
4483        assert_eq!(outcome.failed(), 0);
4484        assert_eq!(fs.read_file("new/cover.jpg").unwrap(), b"FETCHED");
4485        assert_eq!(
4486            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
4487            "new/cover.jpg"
4488        );
4489    }
4490
4491    #[test]
4492    fn cover_move_falls_back_when_source_co_referenced() {
4493        // Two clips' covers share old/cover.jpg after a prior failed swap. A move
4494        // for `a` must NOT rename the shared file away (that would strand `b`); it
4495        // falls back to a fetch, and `b`'s file survives.
4496        let mut manifest = Manifest::new();
4497        let mut a = entry("a.mp3", AudioFormat::Mp3);
4498        a.cover_jpg = Some(ArtifactState {
4499            path: "old/cover.jpg".to_owned(),
4500            hash: "h".to_owned(),
4501        });
4502        manifest.insert("a", a);
4503        let mut b = entry("b.mp3", AudioFormat::Mp3);
4504        b.cover_jpg = Some(ArtifactState {
4505            path: "old/cover.jpg".to_owned(),
4506            hash: "h".to_owned(),
4507        });
4508        manifest.insert("b", b);
4509        let fs = MemFs::new().with_file("old/cover.jpg", b"SHARED".to_vec());
4510        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"FETCHED-A".to_vec()));
4511        // Only `a` moves this run: old/cover.jpg -> a/cover.jpg.
4512        let plan = Plan {
4513            actions: vec![Action::MoveArtifact {
4514                kind: ArtifactKind::CoverJpg,
4515                from: "old/cover.jpg".to_owned(),
4516                to: "a/cover.jpg".to_owned(),
4517                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
4518                hash: "h".to_owned(),
4519                owner_id: "a".to_owned(),
4520            }],
4521        };
4522
4523        let outcome = run(
4524            &plan,
4525            &mut manifest,
4526            &[],
4527            &http,
4528            &fs,
4529            &StubFfmpeg::flac(),
4530            &RecordingClock::new(),
4531            &ExecOptions::default(),
4532        );
4533
4534        assert_eq!(outcome.failed(), 0);
4535        // `a` got a fresh fetched copy; `b`'s shared file is untouched.
4536        assert_eq!(fs.read_file("a/cover.jpg").unwrap(), b"FETCHED-A");
4537        assert_eq!(
4538            fs.read_file("old/cover.jpg").unwrap(),
4539            b"SHARED",
4540            "the co-referenced file must survive"
4541        );
4542    }
4543
4544    #[test]
4545    fn stem_move_renames_without_refetch() {
4546        // #141: a MoveStem relocates the raw stem with a rename; no route is set,
4547        // so a clean outcome proves it did not re-render or re-fetch.
4548        let mut manifest = Manifest::new();
4549        let mut e = entry("a.flac", AudioFormat::Flac);
4550        e.stems.insert(
4551            "voc".to_owned(),
4552            ArtifactState {
4553                path: "old.stems/voc.mp3".to_owned(),
4554                hash: "h1".to_owned(),
4555            },
4556        );
4557        manifest.insert("a", e);
4558        let fs = MemFs::new().with_file("old.stems/voc.mp3", b"STEMBYTES".to_vec());
4559        let plan = Plan {
4560            actions: vec![Action::MoveStem {
4561                clip_id: "a".to_owned(),
4562                key: "voc".to_owned(),
4563                stem_id: "voc".to_owned(),
4564                from: "old.stems/voc.mp3".to_owned(),
4565                to: "new.stems/voc.mp3".to_owned(),
4566                source_url: "https://cdn1.suno.ai/voc.mp3".to_owned(),
4567                format: StemFormat::Mp3,
4568                hash: "h1".to_owned(),
4569            }],
4570        };
4571
4572        let outcome = run(
4573            &plan,
4574            &mut manifest,
4575            &[],
4576            &ScriptedHttp::new(),
4577            &fs,
4578            &StubFfmpeg::flac(),
4579            &RecordingClock::new(),
4580            &ExecOptions::default(),
4581        );
4582
4583        assert_eq!(outcome.failed(), 0);
4584        assert_eq!(outcome.renamed, 1);
4585        assert_eq!(fs.read_file("new.stems/voc.mp3").unwrap(), b"STEMBYTES");
4586        assert!(!fs.exists("old.stems/voc.mp3"));
4587        assert_eq!(
4588            manifest.get("a").unwrap().stems.get("voc").unwrap().path,
4589            "new.stems/voc.mp3"
4590        );
4591    }
4592
4593    #[test]
4594    fn stem_move_falls_back_to_fetch_when_source_co_referenced() {
4595        // Two clips' stems share shared.stems/voc.mp3 after a partially-failed
4596        // swap (the file holds `a`'s bytes). When `b` moves it, move_stem must NOT
4597        // rename the shared file under `b`'s hash (that records `a`'s bytes as
4598        // `b`'s); it falls back to a fetch of `b`'s correct bytes.
4599        let mut manifest = Manifest::new();
4600        let mut a = entry("a.flac", AudioFormat::Flac);
4601        a.stems.insert(
4602            "voc".to_owned(),
4603            ArtifactState {
4604                path: "shared.stems/voc.mp3".to_owned(),
4605                hash: "h".to_owned(),
4606            },
4607        );
4608        manifest.insert("a", a);
4609        let mut b = entry("b.flac", AudioFormat::Flac);
4610        b.stems.insert(
4611            "voc".to_owned(),
4612            ArtifactState {
4613                path: "shared.stems/voc.mp3".to_owned(),
4614                hash: "h".to_owned(),
4615            },
4616        );
4617        manifest.insert("b", b);
4618        let fs = MemFs::new().with_file("shared.stems/voc.mp3", b"A-STEM".to_vec());
4619        let http = ScriptedHttp::new().route("bvoc.mp3", Reply::ok(b"B-STEM".to_vec()));
4620        let plan = Plan {
4621            actions: vec![Action::MoveStem {
4622                clip_id: "b".to_owned(),
4623                key: "voc".to_owned(),
4624                stem_id: "bvoc".to_owned(),
4625                from: "shared.stems/voc.mp3".to_owned(),
4626                to: "b.stems/voc.mp3".to_owned(),
4627                source_url: "https://cdn1.suno.ai/bvoc.mp3".to_owned(),
4628                format: StemFormat::Mp3,
4629                hash: "h".to_owned(),
4630            }],
4631        };
4632
4633        let outcome = run(
4634            &plan,
4635            &mut manifest,
4636            &[],
4637            &http,
4638            &fs,
4639            &StubFfmpeg::flac(),
4640            &RecordingClock::new(),
4641            &ExecOptions::default(),
4642        );
4643
4644        assert_eq!(outcome.failed(), 0);
4645        // b's new stem carries b's freshly fetched bytes, never a's renamed bytes.
4646        assert_eq!(fs.read_file("b.stems/voc.mp3").unwrap(), b"B-STEM");
4647        assert_eq!(
4648            fs.read_file("shared.stems/voc.mp3").unwrap(),
4649            b"A-STEM",
4650            "the co-referenced stem must survive"
4651        );
4652    }
4653
4654    #[test]
4655    fn write_stem_keeps_shared_stem_when_co_referenced() {
4656        // Two clips share shared.stems/voc.mp3 after a prior partially-failed swap.
4657        // When `b` writes to a new path, write_stem must NOT remove the shared file;
4658        // clip `a` still references it and its stem must survive.
4659        let mut manifest = Manifest::new();
4660        let mut a = entry("a.flac", AudioFormat::Flac);
4661        a.stems.insert(
4662            "voc".to_owned(),
4663            ArtifactState {
4664                path: "shared.stems/voc.mp3".to_owned(),
4665                hash: "h".to_owned(),
4666            },
4667        );
4668        manifest.insert("a", a);
4669        let mut b = entry("b.flac", AudioFormat::Flac);
4670        b.stems.insert(
4671            "voc".to_owned(),
4672            ArtifactState {
4673                path: "shared.stems/voc.mp3".to_owned(),
4674                hash: "h".to_owned(),
4675            },
4676        );
4677        manifest.insert("b", b);
4678        let fs = MemFs::new().with_file("shared.stems/voc.mp3", b"A-STEM".to_vec());
4679        let http = ScriptedHttp::new().route("bvoc.mp3", Reply::ok(b"B-STEM".to_vec()));
4680        let plan = Plan {
4681            actions: vec![Action::WriteStem {
4682                clip_id: "b".to_owned(),
4683                key: "voc".to_owned(),
4684                stem_id: "bvoc".to_owned(),
4685                path: "b.stems/voc.mp3".to_owned(),
4686                source_url: "https://cdn1.suno.ai/bvoc.mp3".to_owned(),
4687                format: StemFormat::Mp3,
4688                hash: "bh".to_owned(),
4689            }],
4690        };
4691
4692        let outcome = run(
4693            &plan,
4694            &mut manifest,
4695            &[],
4696            &http,
4697            &fs,
4698            &StubFfmpeg::flac(),
4699            &RecordingClock::new(),
4700            &ExecOptions::default(),
4701        );
4702
4703        assert_eq!(outcome.failed(), 0);
4704        assert_eq!(fs.read_file("b.stems/voc.mp3").unwrap(), b"B-STEM");
4705        assert_eq!(
4706            fs.read_file("shared.stems/voc.mp3").unwrap(),
4707            b"A-STEM",
4708            "the co-referenced stem must survive"
4709        );
4710    }
4711
4712    #[test]
4713    fn co_delete_executes_audio_delete_then_artifact_delete() {
4714        // The plan orders the audio Delete before its sidecar DeleteArtifact.
4715        // The audio delete removes the manifest entry; the sidecar delete then
4716        // removes the file and tolerates the now-absent entry.
4717        let fs = MemFs::new()
4718            .with_file("gone.mp3", b"DATA".to_vec())
4719            .with_file("gone/cover.jpg", b"jpg".to_vec());
4720        let mut manifest = Manifest::new();
4721        let mut e = entry("gone.mp3", AudioFormat::Mp3);
4722        e.cover_jpg = Some(ArtifactState {
4723            path: "gone/cover.jpg".to_owned(),
4724            hash: "h1".to_owned(),
4725        });
4726        manifest.insert("gone", e);
4727        let plan = Plan {
4728            actions: vec![
4729                Action::Delete {
4730                    path: "gone.mp3".to_owned(),
4731                    clip_id: "gone".to_owned(),
4732                },
4733                Action::DeleteArtifact {
4734                    kind: ArtifactKind::CoverJpg,
4735                    path: "gone/cover.jpg".to_owned(),
4736                    owner_id: "gone".to_owned(),
4737                },
4738            ],
4739        };
4740
4741        let outcome = run(
4742            &plan,
4743            &mut manifest,
4744            &[],
4745            &ScriptedHttp::new(),
4746            &fs,
4747            &StubFfmpeg::flac(),
4748            &RecordingClock::new(),
4749            &ExecOptions::default(),
4750        );
4751
4752        assert_eq!(outcome.deleted, 1);
4753        assert_eq!(outcome.artifacts_deleted, 1);
4754        assert_eq!(outcome.failed(), 0);
4755        assert!(!fs.exists("gone.mp3"));
4756        assert!(!fs.exists("gone/cover.jpg"));
4757        assert!(manifest.get("gone").is_none());
4758    }
4759
4760    #[test]
4761    fn write_stem_mp3_stores_raw_and_records_slot() {
4762        // An MP3 stem is downloaded straight from its CDN url and stored verbatim
4763        // (no transcode, no WAV render): the bytes land at the `.mp3` path and the
4764        // keyed slot records the path and hash.
4765        let mut manifest = Manifest::new();
4766        manifest.insert("a", entry("a.flac", AudioFormat::Flac));
4767        let plan = Plan {
4768            actions: vec![Action::WriteStem {
4769                clip_id: "a".to_owned(),
4770                key: "voc".to_owned(),
4771                stem_id: "voc".to_owned(),
4772                path: "a.stems/a - Vocals [voc].mp3".to_owned(),
4773                source_url: "https://cdn1.suno.ai/voc.mp3".to_owned(),
4774                format: StemFormat::Mp3,
4775                hash: "vh".to_owned(),
4776            }],
4777        };
4778        let http = ScriptedHttp::new().route("voc.mp3", Reply::ok(b"stem-bytes".to_vec()));
4779        let fs = MemFs::new();
4780
4781        let outcome = run(
4782            &plan,
4783            &mut manifest,
4784            &[],
4785            &http,
4786            &fs,
4787            &StubFfmpeg::flac(),
4788            &RecordingClock::new(),
4789            &ExecOptions::default(),
4790        );
4791
4792        assert_eq!(outcome.artifacts_written, 1);
4793        assert_eq!(outcome.failed(), 0);
4794        // Bytes are stored exactly as delivered (no transcode applied).
4795        assert_eq!(
4796            fs.read_file("a.stems/a - Vocals [voc].mp3").unwrap(),
4797            b"stem-bytes"
4798        );
4799        // An MP3 stem never renders WAV: no convert_wav, no generation.
4800        assert_eq!(http.count("convert_wav"), 0);
4801        assert_eq!(http.count("/api/gen/"), 0);
4802        assert_eq!(
4803            manifest.get("a").unwrap().stems.get("voc"),
4804            Some(&ArtifactState {
4805                path: "a.stems/a - Vocals [voc].mp3".to_owned(),
4806                hash: "vh".to_owned(),
4807            })
4808        );
4809    }
4810
4811    #[test]
4812    fn write_stem_wav_renders_via_convert_wav_and_stores_raw() {
4813        // A WAV stem (the default) renders the stem clip's lossless WAV through the
4814        // free convert_wav flow keyed on the stem id, then downloads and stores it
4815        // RAW as `.wav` — it is NEVER transcoded to FLAC, even for a FLAC song.
4816        let mut manifest = Manifest::new();
4817        manifest.insert("a", entry("a.flac", AudioFormat::Flac));
4818        let plan = Plan {
4819            actions: vec![Action::WriteStem {
4820                clip_id: "a".to_owned(),
4821                key: "voc".to_owned(),
4822                stem_id: "stemvoc".to_owned(),
4823                path: "a.stems/a - Vocals [stemvoc].wav".to_owned(),
4824                source_url: "https://cdn1.suno.ai/stemvoc.mp3".to_owned(),
4825                format: StemFormat::Wav,
4826                hash: "vh".to_owned(),
4827            }],
4828        };
4829        // wav_file is not ready on the first poll, so the flow POSTs convert_wav
4830        // (free) and polls again — exactly the main FLAC/WAV render path.
4831        let http = ScriptedHttp::new()
4832            .with_auth()
4833            .route_seq(
4834                "stemvoc/wav_file/",
4835                vec![
4836                    Reply::json("{}"),
4837                    Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/stemvoc.wav"}"#),
4838                ],
4839            )
4840            .route("stemvoc/convert_wav/", Reply::status(200))
4841            .route("stemvoc.wav", Reply::ok(b"RIFFwav-bytes".to_vec()));
4842        let fs = MemFs::new();
4843
4844        let outcome = run(
4845            &plan,
4846            &mut manifest,
4847            &[],
4848            &http,
4849            &fs,
4850            &StubFfmpeg::flac(),
4851            &RecordingClock::new(),
4852            &small_poll(),
4853        );
4854
4855        assert_eq!(outcome.artifacts_written, 1);
4856        assert_eq!(outcome.failed(), 0);
4857        // The rendered WAV is stored verbatim; ffmpeg (WAV->FLAC) is never invoked,
4858        // so the stored bytes are the raw WAV, not a FLAC transcode.
4859        assert_eq!(
4860            fs.read_file("a.stems/a - Vocals [stemvoc].wav").unwrap(),
4861            b"RIFFwav-bytes"
4862        );
4863        assert!(!fs.exists("a.stems/a - Vocals [stemvoc].flac"));
4864        // The free WAV render ran; no credit-spending generation endpoint did.
4865        assert_eq!(http.count("convert_wav"), 1);
4866        assert_eq!(http.count("stem_task"), 0);
4867        assert_eq!(http.count("separate"), 0);
4868        assert_eq!(
4869            manifest.get("a").unwrap().stems.get("voc").unwrap().path,
4870            "a.stems/a - Vocals [stemvoc].wav"
4871        );
4872    }
4873
4874    #[test]
4875    fn write_stem_is_skipped_when_owner_audio_is_absent() {
4876        // No owning manifest entry (audio failed or never existed) => skip with
4877        // no fetch and no write, so a stem is never stranded without its song.
4878        let mut manifest = Manifest::new();
4879        let plan = Plan {
4880            actions: vec![Action::WriteStem {
4881                clip_id: "ghost".to_owned(),
4882                key: "voc".to_owned(),
4883                stem_id: "voc".to_owned(),
4884                path: "ghost.stems/voc.mp3".to_owned(),
4885                source_url: "https://cdn1.suno.ai/voc.mp3".to_owned(),
4886                format: StemFormat::Mp3,
4887                hash: "vh".to_owned(),
4888            }],
4889        };
4890        // Empty HTTP script: any fetch would error, proving none happens.
4891        let http = ScriptedHttp::new();
4892        let fs = MemFs::new();
4893
4894        let outcome = run(
4895            &plan,
4896            &mut manifest,
4897            &[],
4898            &http,
4899            &fs,
4900            &StubFfmpeg::flac(),
4901            &RecordingClock::new(),
4902            &ExecOptions::default(),
4903        );
4904
4905        assert_eq!(outcome.skipped, 1);
4906        assert_eq!(outcome.artifacts_written, 0);
4907        assert_eq!(outcome.failed(), 0);
4908        assert!(!fs.exists("ghost.stems/voc.mp3"));
4909    }
4910
4911    #[test]
4912    fn write_stem_relocates_the_old_file_on_a_path_move() {
4913        // The song was renamed, so the stem moves: the new file is written and the
4914        // stale copy at the previously tracked path is removed (moved, not orphaned).
4915        let fs = MemFs::new().with_file("old.stems/voc.mp3", b"old".to_vec());
4916        let mut manifest = Manifest::new();
4917        let mut e = entry("new.flac", AudioFormat::Flac);
4918        e.stems.insert(
4919            "voc".to_owned(),
4920            ArtifactState {
4921                path: "old.stems/voc.mp3".to_owned(),
4922                hash: "vh".to_owned(),
4923            },
4924        );
4925        manifest.insert("a", e);
4926        let plan = Plan {
4927            actions: vec![Action::WriteStem {
4928                clip_id: "a".to_owned(),
4929                key: "voc".to_owned(),
4930                stem_id: "voc".to_owned(),
4931                path: "new.stems/voc.mp3".to_owned(),
4932                source_url: "https://cdn1.suno.ai/voc.mp3".to_owned(),
4933                format: StemFormat::Mp3,
4934                hash: "vh".to_owned(),
4935            }],
4936        };
4937        let http = ScriptedHttp::new().route("voc.mp3", Reply::ok(b"new".to_vec()));
4938
4939        let outcome = run(
4940            &plan,
4941            &mut manifest,
4942            &[],
4943            &http,
4944            &fs,
4945            &StubFfmpeg::flac(),
4946            &RecordingClock::new(),
4947            &ExecOptions::default(),
4948        );
4949
4950        assert_eq!(outcome.artifacts_written, 1);
4951        assert!(fs.exists("new.stems/voc.mp3"));
4952        assert!(
4953            !fs.exists("old.stems/voc.mp3"),
4954            "the old stem is moved, not left behind"
4955        );
4956        assert_eq!(
4957            manifest.get("a").unwrap().stems.get("voc").unwrap().path,
4958            "new.stems/voc.mp3"
4959        );
4960    }
4961
4962    #[test]
4963    fn delete_stem_removes_file_and_clears_slot() {
4964        let fs = MemFs::new().with_file("a.stems/voc.mp3", b"stem".to_vec());
4965        let mut manifest = Manifest::new();
4966        let mut e = entry("a.flac", AudioFormat::Flac);
4967        e.stems.insert(
4968            "voc".to_owned(),
4969            ArtifactState {
4970                path: "a.stems/voc.mp3".to_owned(),
4971                hash: "vh".to_owned(),
4972            },
4973        );
4974        manifest.insert("a", e);
4975        let plan = Plan {
4976            actions: vec![Action::DeleteStem {
4977                clip_id: "a".to_owned(),
4978                key: "voc".to_owned(),
4979                path: "a.stems/voc.mp3".to_owned(),
4980            }],
4981        };
4982
4983        let outcome = run(
4984            &plan,
4985            &mut manifest,
4986            &[],
4987            &ScriptedHttp::new(),
4988            &fs,
4989            &StubFfmpeg::flac(),
4990            &RecordingClock::new(),
4991            &ExecOptions::default(),
4992        );
4993
4994        assert_eq!(outcome.artifacts_deleted, 1);
4995        assert!(!fs.exists("a.stems/voc.mp3"));
4996        assert!(manifest.get("a").unwrap().stems.is_empty());
4997    }
4998
4999    #[test]
5000    fn co_deleting_the_last_stem_prunes_the_stems_folder() {
5001        // Deleting a song co-deletes its stems; the emptied `.stems` folder is
5002        // pruned by the end-of-run sweep, so it can never be orphaned.
5003        let fs = MemFs::new()
5004            .with_file("song.flac", b"DATA".to_vec())
5005            .with_file("song.stems/voc.mp3", b"stem".to_vec());
5006        assert!(fs.has_dir("song.stems"));
5007        let mut manifest = Manifest::new();
5008        let mut e = entry("song.flac", AudioFormat::Flac);
5009        e.stems.insert(
5010            "voc".to_owned(),
5011            ArtifactState {
5012                path: "song.stems/voc.mp3".to_owned(),
5013                hash: "vh".to_owned(),
5014            },
5015        );
5016        manifest.insert("a", e);
5017        let plan = Plan {
5018            actions: vec![
5019                Action::Delete {
5020                    path: "song.flac".to_owned(),
5021                    clip_id: "a".to_owned(),
5022                },
5023                Action::DeleteStem {
5024                    clip_id: "a".to_owned(),
5025                    key: "voc".to_owned(),
5026                    path: "song.stems/voc.mp3".to_owned(),
5027                },
5028            ],
5029        };
5030
5031        let outcome = run(
5032            &plan,
5033            &mut manifest,
5034            &[],
5035            &ScriptedHttp::new(),
5036            &fs,
5037            &StubFfmpeg::flac(),
5038            &RecordingClock::new(),
5039            &ExecOptions::default(),
5040        );
5041
5042        assert_eq!(outcome.deleted, 1);
5043        assert_eq!(outcome.artifacts_deleted, 1);
5044        assert!(!fs.exists("song.flac"));
5045        assert!(!fs.exists("song.stems/voc.mp3"));
5046        assert!(
5047            !fs.has_dir("song.stems"),
5048            "the emptied .stems folder is pruned"
5049        );
5050        assert!(manifest.get("a").is_none());
5051    }
5052
5053    #[test]
5054    fn full_stems_mirror_mp3_is_get_only_with_zero_gen_traffic() {
5055        // End-to-end #100 path with MP3 stems: list a clip's existing stems (free
5056        // GET over the live page-count + 0-indexed page shape), reconcile them into
5057        // WriteStem actions, and execute (download) them. With MP3 the whole flow
5058        // is GET-only and touches NO `/api/gen/` endpoint at all.
5059        let http = ScriptedHttp::new()
5060            .with_auth()
5061            .route("clip1/stems/pages", Reply::json(r#"{"pages": 1}"#))
5062            .route(
5063                "clip1/stems?page=0",
5064                Reply::json(
5065                    r#"{"stems":[
5066                        {"id":"s1","title":"Song (Vocals)","status":"complete","audio_url":"https://cdn1.suno.ai/s1.mp3"},
5067                        {"id":"s2","title":"Song (Drums)","status":"complete","audio_url":"https://cdn1.suno.ai/s2.mp3"}
5068                    ]}"#,
5069                ),
5070            )
5071            .route("s1.mp3", Reply::ok(b"vocals-bytes".to_vec()))
5072            .route("s2.mp3", Reply::ok(b"drums-bytes".to_vec()));
5073
5074        // List the existing stems through the client (GET-only, free).
5075        let auth = ClerkAuth::new("eyJtoken");
5076        pollster::block_on(auth.authenticate(&http)).unwrap();
5077        let client = SunoClient::new(auth, RecordingClock::new());
5078        let (stems, complete) = pollster::block_on(client.list_stems(&http, "clip1")).unwrap();
5079        assert!(complete);
5080        assert_eq!(stems.len(), 2);
5081        assert_eq!(stems[0].label, "Vocals");
5082
5083        // Reconcile the listed MP3 stems into a plan (audio already present -> Skip).
5084        let mut manifest = Manifest::new();
5085        manifest.insert("clip1", entry("clip1.flac", AudioFormat::Flac));
5086        let desired_stems: Vec<crate::reconcile::DesiredStem> = stems
5087            .iter()
5088            .map(|s| crate::reconcile::DesiredStem {
5089                key: s.id.clone(),
5090                stem_id: s.id.clone(),
5091                path: format!("clip1.stems/{}.mp3", s.id),
5092                source_url: s.url.clone(),
5093                format: StemFormat::Mp3,
5094                hash: crate::art_url_hash(&s.url),
5095            })
5096            .collect();
5097        let d = Desired {
5098            path: "clip1.flac".to_owned(),
5099            stems: Some(desired_stems),
5100            ..desired(clip("clip1"), AudioFormat::Flac)
5101        };
5102        let local: HashMap<String, crate::reconcile::LocalFile> = [(
5103            "clip1".to_owned(),
5104            crate::reconcile::LocalFile {
5105                exists: true,
5106                size: 100,
5107            },
5108        )]
5109        .into_iter()
5110        .collect();
5111        let sources = [crate::reconcile::SourceStatus {
5112            mode: SourceMode::Mirror,
5113            fully_enumerated: true,
5114        }];
5115        let plan =
5116            crate::reconcile::reconcile(&manifest, std::slice::from_ref(&d), &local, &sources);
5117        assert_eq!(plan.stem_writes(), 2);
5118
5119        let fs = MemFs::new();
5120        let outcome = run(
5121            &plan,
5122            &mut manifest,
5123            std::slice::from_ref(&d),
5124            &http,
5125            &fs,
5126            &StubFfmpeg::flac(),
5127            &RecordingClock::new(),
5128            &ExecOptions::default(),
5129        );
5130
5131        assert_eq!(outcome.artifacts_written, 2, "both stems downloaded");
5132        assert_eq!(fs.read_file("clip1.stems/s1.mp3").unwrap(), b"vocals-bytes");
5133        assert_eq!(fs.read_file("clip1.stems/s2.mp3").unwrap(), b"drums-bytes");
5134        // The MP3 mirror path never touches any /api/gen/ endpoint (no render, no
5135        // generation, no separation).
5136        assert_eq!(http.count("/api/gen/"), 0);
5137        assert_eq!(http.count("stem_task"), 0);
5138        assert_eq!(http.count("separate"), 0);
5139        assert_eq!(http.count("generate"), 0);
5140        // No stem is ever written as FLAC.
5141        assert!(!fs.exists("clip1.stems/s1.flac"));
5142    }
5143
5144    #[test]
5145    fn full_stems_mirror_wav_default_renders_free_wav_and_no_generation() {
5146        // End-to-end #100 path with WAV stems (the default): each stem's lossless
5147        // WAV is rendered through the FREE convert_wav flow and stored RAW as
5148        // `.wav`. The mirror makes NO credit-spending generation POST.
5149        let http = ScriptedHttp::new()
5150            .with_auth()
5151            .route("clip1/stems/pages", Reply::json(r#"{"pages": 1}"#))
5152            .route(
5153                "clip1/stems?page=0",
5154                Reply::json(
5155                    r#"{"stems":[
5156                        {"id":"s1","title":"Song (Vocals)","status":"complete","audio_url":"https://cdn1.suno.ai/s1.mp3"},
5157                        {"id":"s2","title":"Song (Drums)","status":"complete","audio_url":"https://cdn1.suno.ai/s2.mp3"}
5158                    ]}"#,
5159                ),
5160            )
5161            // Each stem's WAV is already rendered, so wav_file returns the url and
5162            // no convert_wav POST is even needed (still free either way).
5163            .route(
5164                "s1/wav_file/",
5165                Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/s1.wav"}"#),
5166            )
5167            .route(
5168                "s2/wav_file/",
5169                Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/s2.wav"}"#),
5170            )
5171            .route("s1.wav", Reply::ok(b"RIFFvocals".to_vec()))
5172            .route("s2.wav", Reply::ok(b"RIFFdrums".to_vec()));
5173
5174        let auth = ClerkAuth::new("eyJtoken");
5175        pollster::block_on(auth.authenticate(&http)).unwrap();
5176        let client = SunoClient::new(auth, RecordingClock::new());
5177        let (stems, _complete) = pollster::block_on(client.list_stems(&http, "clip1")).unwrap();
5178
5179        let mut manifest = Manifest::new();
5180        manifest.insert("clip1", entry("clip1.flac", AudioFormat::Flac));
5181        let desired_stems: Vec<crate::reconcile::DesiredStem> = stems
5182            .iter()
5183            .map(|s| crate::reconcile::DesiredStem {
5184                key: s.id.clone(),
5185                stem_id: s.id.clone(),
5186                path: format!("clip1.stems/{}.wav", s.id),
5187                source_url: s.url.clone(),
5188                format: StemFormat::Wav,
5189                hash: crate::art_url_hash(&s.url),
5190            })
5191            .collect();
5192        let d = Desired {
5193            path: "clip1.flac".to_owned(),
5194            stems: Some(desired_stems),
5195            ..desired(clip("clip1"), AudioFormat::Flac)
5196        };
5197        let local: HashMap<String, crate::reconcile::LocalFile> = [(
5198            "clip1".to_owned(),
5199            crate::reconcile::LocalFile {
5200                exists: true,
5201                size: 100,
5202            },
5203        )]
5204        .into_iter()
5205        .collect();
5206        let sources = [crate::reconcile::SourceStatus {
5207            mode: SourceMode::Mirror,
5208            fully_enumerated: true,
5209        }];
5210        let plan =
5211            crate::reconcile::reconcile(&manifest, std::slice::from_ref(&d), &local, &sources);
5212
5213        let fs = MemFs::new();
5214        let outcome = run(
5215            &plan,
5216            &mut manifest,
5217            std::slice::from_ref(&d),
5218            &http,
5219            &fs,
5220            &StubFfmpeg::flac(),
5221            &RecordingClock::new(),
5222            &small_poll(),
5223        );
5224
5225        assert_eq!(outcome.artifacts_written, 2);
5226        // Stems are stored RAW as WAV (no FLAC transcode, even for a FLAC song).
5227        assert_eq!(fs.read_file("clip1.stems/s1.wav").unwrap(), b"RIFFvocals");
5228        assert_eq!(fs.read_file("clip1.stems/s2.wav").unwrap(), b"RIFFdrums");
5229        assert!(!fs.exists("clip1.stems/s1.flac"));
5230        // No credit-spending generation/separation endpoint is ever hit.
5231        assert_eq!(http.count("stem_task"), 0);
5232        assert_eq!(http.count("separate"), 0);
5233        assert_eq!(http.count("generate"), 0);
5234    }
5235
5236    #[test]
5237    fn write_artifact_is_skipped_when_the_owner_audio_is_absent() {
5238        // A clip whose Download fails leaves no manifest entry, so its following
5239        // WriteArtifact must not strand an untracked sidecar: it is skipped with
5240        // no fetch and no write. A following healthy clip still succeeds.
5241        let ca = clip("a");
5242        let plan = Plan {
5243            actions: vec![
5244                Action::Download {
5245                    clip: ca.clone(),
5246                    lineage: LineageContext::own_root(&ca),
5247                    path: "a.mp3".to_owned(),
5248                    format: AudioFormat::Mp3,
5249                },
5250                Action::WriteArtifact {
5251                    kind: ArtifactKind::CoverJpg,
5252                    path: "a/cover.jpg".to_owned(),
5253                    source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
5254                    hash: "h1".to_owned(),
5255                    owner_id: "a".to_owned(),
5256                    content: None,
5257                },
5258                Action::WriteArtifact {
5259                    kind: ArtifactKind::CoverJpg,
5260                    path: "b/cover.jpg".to_owned(),
5261                    source_url: "https://art.suno.ai/b/large.jpg".to_owned(),
5262                    hash: "h2".to_owned(),
5263                    owner_id: "b".to_owned(),
5264                    content: None,
5265                },
5266            ],
5267        };
5268        // The Download's audio 404s (permanent), so no entry for "a" is created.
5269        let http = ScriptedHttp::new()
5270            .route("a.mp3", Reply::status(404))
5271            .route("a/large.jpg", Reply::ok(b"jpg-a".to_vec()))
5272            .route("b/large.jpg", Reply::ok(b"jpg-b".to_vec()));
5273        let fs = MemFs::new();
5274        let mut manifest = Manifest::new();
5275        // "b" already has audio (a prior-run clip), so its sidecar write proceeds.
5276        manifest.insert("b", entry("b.mp3", AudioFormat::Mp3));
5277
5278        let outcome = run(
5279            &plan,
5280            &mut manifest,
5281            &[],
5282            &http,
5283            &fs,
5284            &StubFfmpeg::flac(),
5285            &RecordingClock::new(),
5286            &ExecOptions::default(),
5287        );
5288
5289        assert_eq!(outcome.status, RunStatus::Completed);
5290        // The audio download is the only failure; the orphan artifact is skipped.
5291        assert_eq!(outcome.failed(), 1);
5292        assert_eq!(outcome.failures[0].clip_id, "a");
5293        assert_eq!(outcome.skipped, 1);
5294        // The orphan sidecar was neither fetched nor written, and left no record.
5295        assert_eq!(http.count("a/large.jpg"), 0);
5296        assert!(!fs.exists("a/cover.jpg"));
5297        assert!(manifest.get("a").is_none());
5298        // The healthy clip's sidecar still succeeded.
5299        assert_eq!(outcome.artifacts_written, 1);
5300        assert_eq!(fs.read_file("b/cover.jpg").unwrap(), b"jpg-b");
5301        assert!(manifest.get("b").unwrap().cover_jpg.is_some());
5302    }
5303
5304    #[test]
5305    fn download_embeds_animated_webp_cover_when_enabled() {
5306        // With animated covers on and a video preview present, the audio embeds
5307        // the transcoded WebP (image/webp) as its front cover, not the static JPEG.
5308        let c = Clip {
5309            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5310            ..art_clip("a")
5311        };
5312        let d = desired(c.clone(), AudioFormat::Mp3);
5313        let plan = Plan {
5314            actions: vec![Action::Download {
5315                clip: c.clone(),
5316                lineage: LineageContext::own_root(&c),
5317                path: d.path.clone(),
5318                format: AudioFormat::Mp3,
5319            }],
5320        };
5321        let http = ScriptedHttp::new()
5322            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5323            .route("a/video.mp4", Reply::ok(b"mp4-bytes".to_vec()))
5324            .route("a/large.jpg", Reply::ok(b"static-jpg".to_vec()));
5325        let fs = MemFs::new();
5326        let opts = ExecOptions {
5327            embed_animated_cover: true,
5328            ..ExecOptions::default()
5329        };
5330        let mut manifest = Manifest::new();
5331
5332        let outcome = run(
5333            &plan,
5334            &mut manifest,
5335            &[d],
5336            &http,
5337            &fs,
5338            &StubFfmpeg::flac(),
5339            &RecordingClock::new(),
5340            &opts,
5341        );
5342
5343        assert_eq!(outcome.downloaded, 1);
5344        assert_eq!(outcome.failed(), 0);
5345        let written = fs.read_file("a.mp3").unwrap();
5346        let tag = id3::Tag::read_from2(std::io::Cursor::new(&written)).unwrap();
5347        let pic = tag.pictures().next().expect("embedded cover");
5348        assert_eq!(pic.mime_type, "image/webp");
5349        assert!(
5350            pic.data.starts_with(b"RIFF"),
5351            "the embedded cover is the transcoded WebP"
5352        );
5353        // The MP4 preview was fetched and transcoded; the static JPEG was not needed.
5354        assert_eq!(http.count("a/video.mp4"), 1);
5355        assert_eq!(http.count("a/large.jpg"), 0);
5356    }
5357
5358    #[test]
5359    fn download_keeps_static_jpeg_cover_when_embed_disabled() {
5360        // With the feature off (default), even a clip with a video preview embeds
5361        // the static JPEG and never fetches or transcodes the MP4.
5362        let c = Clip {
5363            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5364            ..art_clip("a")
5365        };
5366        let d = desired(c.clone(), AudioFormat::Mp3);
5367        let plan = Plan {
5368            actions: vec![Action::Download {
5369                clip: c.clone(),
5370                lineage: LineageContext::own_root(&c),
5371                path: d.path.clone(),
5372                format: AudioFormat::Mp3,
5373            }],
5374        };
5375        let http = ScriptedHttp::new()
5376            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5377            .route("a/large.jpg", Reply::ok(b"static-jpg".to_vec()));
5378        let fs = MemFs::new();
5379        let mut manifest = Manifest::new();
5380
5381        let outcome = run(
5382            &plan,
5383            &mut manifest,
5384            &[d],
5385            &http,
5386            &fs,
5387            &StubFfmpeg::flac(),
5388            &RecordingClock::new(),
5389            &ExecOptions::default(),
5390        );
5391
5392        assert_eq!(outcome.failed(), 0);
5393        let written = fs.read_file("a.mp3").unwrap();
5394        let tag = id3::Tag::read_from2(std::io::Cursor::new(&written)).unwrap();
5395        let pic = tag.pictures().next().expect("embedded cover");
5396        assert_eq!(pic.mime_type, "image/jpeg");
5397        assert_eq!(pic.data, b"static-jpg");
5398        assert_eq!(http.count("a/video.mp4"), 0);
5399    }
5400
5401    #[test]
5402    fn oversized_animated_cover_falls_back_to_jpeg_embed() {
5403        // A transcoded WebP that would overflow the FLAC picture cap is not
5404        // embedded; the audio falls back to the static JPEG so the file stays
5405        // valid (and no re-tag loop, since the intent hash is unchanged).
5406        let c = Clip {
5407            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5408            ..art_clip("a")
5409        };
5410        let d = desired(c.clone(), AudioFormat::Mp3);
5411        let plan = Plan {
5412            actions: vec![Action::Download {
5413                clip: c.clone(),
5414                lineage: LineageContext::own_root(&c),
5415                path: d.path.clone(),
5416                format: AudioFormat::Mp3,
5417            }],
5418        };
5419        let http = ScriptedHttp::new()
5420            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5421            .route("a/video.mp4", Reply::ok(b"mp4-bytes".to_vec()))
5422            .route("a/large.jpg", Reply::ok(b"static-jpg".to_vec()));
5423        let fs = MemFs::new();
5424        let oversize = vec![b'R'; flac_picture_data_budget("image/webp") + 1];
5425        let ffmpeg = StubFfmpeg::flac().with_webp(oversize);
5426        let opts = ExecOptions {
5427            embed_animated_cover: true,
5428            ..ExecOptions::default()
5429        };
5430        let mut manifest = Manifest::new();
5431
5432        let outcome = run(
5433            &plan,
5434            &mut manifest,
5435            &[d],
5436            &http,
5437            &fs,
5438            &ffmpeg,
5439            &RecordingClock::new(),
5440            &opts,
5441        );
5442
5443        assert_eq!(outcome.failed(), 0);
5444        let written = fs.read_file("a.mp3").unwrap();
5445        let tag = id3::Tag::read_from2(std::io::Cursor::new(&written)).unwrap();
5446        let pic = tag.pictures().next().expect("embedded cover");
5447        assert_eq!(pic.mime_type, "image/jpeg");
5448        assert_eq!(pic.data, b"static-jpg");
5449    }
5450
5451    #[test]
5452    fn cover_transcode_failure_falls_back_to_jpeg_embed() {
5453        // A non-systemic MP4 fetch/transcode failure never fails the audio: the
5454        // embed falls back to the static JPEG, best-effort like a failed cover
5455        // fetch, and the run completes.
5456        let c = Clip {
5457            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5458            ..art_clip("a")
5459        };
5460        let d = desired(c.clone(), AudioFormat::Mp3);
5461        let plan = Plan {
5462            actions: vec![Action::Download {
5463                clip: c.clone(),
5464                lineage: LineageContext::own_root(&c),
5465                path: d.path.clone(),
5466                format: AudioFormat::Mp3,
5467            }],
5468        };
5469        let http = ScriptedHttp::new()
5470            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5471            .route("a/video.mp4", Reply::ok(b"mp4-bytes".to_vec()))
5472            .route("a/large.jpg", Reply::ok(b"static-jpg".to_vec()));
5473        let fs = MemFs::new();
5474        let opts = ExecOptions {
5475            embed_animated_cover: true,
5476            ..ExecOptions::default()
5477        };
5478        let mut manifest = Manifest::new();
5479
5480        let outcome = run(
5481            &plan,
5482            &mut manifest,
5483            &[d],
5484            &http,
5485            &fs,
5486            &StubFfmpeg::failing(),
5487            &RecordingClock::new(),
5488            &opts,
5489        );
5490
5491        assert_eq!(outcome.status, RunStatus::Completed);
5492        assert_eq!(outcome.failed(), 0);
5493        let written = fs.read_file("a.mp3").unwrap();
5494        let tag = id3::Tag::read_from2(std::io::Cursor::new(&written)).unwrap();
5495        assert_eq!(tag.pictures().next().unwrap().mime_type, "image/jpeg");
5496    }
5497
5498    #[test]
5499    fn disk_full_cover_transcode_aborts_the_run() {
5500        // A full scratch disk during the cover transcode is systemic: it aborts
5501        // the run (exit 9) rather than silently skipping the cover.
5502        let c = Clip {
5503            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5504            ..art_clip("a")
5505        };
5506        let d = desired(c.clone(), AudioFormat::Mp3);
5507        let plan = Plan {
5508            actions: vec![Action::Download {
5509                clip: c.clone(),
5510                lineage: LineageContext::own_root(&c),
5511                path: d.path.clone(),
5512                format: AudioFormat::Mp3,
5513            }],
5514        };
5515        let http = ScriptedHttp::new()
5516            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5517            .route("a/video.mp4", Reply::ok(b"mp4-bytes".to_vec()));
5518        let fs = MemFs::new();
5519        let opts = ExecOptions {
5520            embed_animated_cover: true,
5521            ..ExecOptions::default()
5522        };
5523        let mut manifest = Manifest::new();
5524
5525        let outcome = run(
5526            &plan,
5527            &mut manifest,
5528            &[d],
5529            &http,
5530            &fs,
5531            &StubFfmpeg::out_of_space(),
5532            &RecordingClock::new(),
5533            &opts,
5534        );
5535
5536        assert_eq!(outcome.status, RunStatus::DiskFull);
5537    }
5538
5539    #[test]
5540    fn video_only_clip_never_embeds_the_mp4_as_a_cover() {
5541        // A clip with a video preview but no static image must never embed the
5542        // MP4 bytes as a picture: when the WebP transcode fails and there is no
5543        // static image to fall back to, the audio is written with no cover.
5544        let c = Clip {
5545            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5546            ..clip("a")
5547        };
5548        let d = desired(c.clone(), AudioFormat::Mp3);
5549        let plan = Plan {
5550            actions: vec![Action::Download {
5551                clip: c.clone(),
5552                lineage: LineageContext::own_root(&c),
5553                path: d.path.clone(),
5554                format: AudioFormat::Mp3,
5555            }],
5556        };
5557        let http = ScriptedHttp::new()
5558            .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5559            .route("a/video.mp4", Reply::ok(b"mp4-bytes".to_vec()));
5560        let fs = MemFs::new();
5561        let opts = ExecOptions {
5562            embed_animated_cover: true,
5563            ..ExecOptions::default()
5564        };
5565        let mut manifest = Manifest::new();
5566
5567        let outcome = run(
5568            &plan,
5569            &mut manifest,
5570            &[d],
5571            &http,
5572            &fs,
5573            &StubFfmpeg::failing(),
5574            &RecordingClock::new(),
5575            &opts,
5576        );
5577
5578        assert_eq!(outcome.failed(), 0);
5579        let written = fs.read_file("a.mp3").unwrap();
5580        let tag = id3::Tag::read_from2(std::io::Cursor::new(&written)).unwrap();
5581        assert!(
5582            tag.pictures().next().is_none(),
5583            "no cover embedded, never the MP4"
5584        );
5585        assert!(
5586            !written
5587                .windows(b"mp4-bytes".len())
5588                .any(|w| w == b"mp4-bytes"),
5589            "the MP4 bytes must not be embedded as artwork"
5590        );
5591    }
5592
5593    #[test]
5594    fn embed_uses_configured_webp_settings() {
5595        use std::sync::{Arc, Mutex};
5596
5597        struct RecordingWebpFfmpeg {
5598            seen: Arc<Mutex<Vec<WebpEncodeSettings>>>,
5599        }
5600
5601        impl Ffmpeg for RecordingWebpFfmpeg {
5602            async fn wav_to_lossless(
5603                &self,
5604                _wav: &[u8],
5605                _format: AudioFormat,
5606            ) -> Result<Vec<u8>, crate::ffmpeg::FfmpegError> {
5607                Ok(Vec::new())
5608            }
5609
5610            async fn mp4_to_webp(
5611                &self,
5612                _mp4: &[u8],
5613                settings: WebpEncodeSettings,
5614            ) -> Result<Vec<u8>, crate::ffmpeg::FfmpegError> {
5615                let seen = Arc::clone(&self.seen);
5616                seen.lock().unwrap().push(settings);
5617                Ok(b"RIFF\x00\x00\x00\x00WEBP".to_vec())
5618            }
5619        }
5620
5621        let c = Clip {
5622            video_cover_url: "https://cdn.suno.ai/a/video.mp4".to_owned(),
5623            ..art_clip("a")
5624        };
5625        let d = desired(c.clone(), AudioFormat::Mp3);
5626        let plan = Plan {
5627            actions: vec![Action::Download {
5628                clip: c.clone(),
5629                lineage: LineageContext::own_root(&c),
5630                path: d.path.clone(),
5631                format: AudioFormat::Mp3,
5632            }],
5633        };
5634        let seen = Arc::new(Mutex::new(Vec::new()));
5635        let ffmpeg = RecordingWebpFfmpeg {
5636            seen: Arc::clone(&seen),
5637        };
5638        let opts = ExecOptions {
5639            embed_animated_cover: true,
5640            cover_webp: WebpEncodeSettings {
5641                quality: 88,
5642                max_fps: 12,
5643                max_width: Some(720),
5644                lossless: false,
5645                compression_level: 4,
5646            },
5647            ..ExecOptions::default()
5648        };
5649
5650        let _ = run(
5651            &plan,
5652            &mut Manifest::new(),
5653            &[d],
5654            &ScriptedHttp::new()
5655                .route("a.mp3", Reply::ok(b"mp3-body".to_vec()))
5656                .route("a/video.mp4", Reply::ok(b"mp4-bytes".to_vec())),
5657            &MemFs::new(),
5658            &ffmpeg,
5659            &RecordingClock::new(),
5660            &opts,
5661        );
5662
5663        assert_eq!(
5664            seen.lock().unwrap().as_slice(),
5665            &[WebpEncodeSettings {
5666                quality: 88,
5667                max_fps: 12,
5668                max_width: Some(720),
5669                lossless: false,
5670                compression_level: 4,
5671            }]
5672        );
5673    }
5674
5675    // ── Phase 8: folder art routes to the album store ───────────────
5676
5677    #[test]
5678    fn folder_jpg_write_records_album_state_and_skips_manifest() {
5679        // Folder art is owned by the album root id, not a manifest clip: it
5680        // writes even with an empty manifest and records on the album store.
5681        let mut manifest = Manifest::new();
5682        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5683        let plan = Plan {
5684            actions: vec![Action::WriteArtifact {
5685                kind: ArtifactKind::FolderJpg,
5686                path: "creator/album/folder.jpg".to_owned(),
5687                source_url: "https://art.suno.ai/root/large.jpg".to_owned(),
5688                hash: "jh".to_owned(),
5689                owner_id: "root".to_owned(),
5690                content: None,
5691            }],
5692        };
5693        let http = ScriptedHttp::new().route("root/large.jpg", Reply::ok(b"folder-jpg".to_vec()));
5694        let fs = MemFs::new();
5695
5696        let outcome = run_with_albums(
5697            &plan,
5698            &mut manifest,
5699            &mut albums,
5700            &[],
5701            &http,
5702            &fs,
5703            &StubFfmpeg::flac(),
5704            &RecordingClock::new(),
5705            &ExecOptions::default(),
5706        );
5707
5708        assert_eq!(outcome.artifacts_written, 1);
5709        assert_eq!(outcome.status, RunStatus::Completed);
5710        assert_eq!(
5711            fs.read_file("creator/album/folder.jpg").unwrap(),
5712            b"folder-jpg"
5713        );
5714        assert_eq!(
5715            albums.get("root").unwrap().folder_jpg,
5716            Some(ArtifactState {
5717                path: "creator/album/folder.jpg".to_owned(),
5718                hash: "jh".to_owned(),
5719            })
5720        );
5721        assert!(manifest.get("root").is_none());
5722    }
5723
5724    #[test]
5725    fn folder_webp_write_transcodes_and_records_album_state() {
5726        let mut manifest = Manifest::new();
5727        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5728        let plan = Plan {
5729            actions: vec![Action::WriteArtifact {
5730                kind: ArtifactKind::FolderWebp,
5731                path: "creator/album/cover.webp".to_owned(),
5732                source_url: "https://cdn.suno.ai/root/video.mp4".to_owned(),
5733                hash: "wh".to_owned(),
5734                owner_id: "root".to_owned(),
5735                content: None,
5736            }],
5737        };
5738        let http = ScriptedHttp::new().route("root/video.mp4", Reply::ok(b"mp4-bytes".to_vec()));
5739        let fs = MemFs::new();
5740
5741        let outcome = run_with_albums(
5742            &plan,
5743            &mut manifest,
5744            &mut albums,
5745            &[],
5746            &http,
5747            &fs,
5748            &StubFfmpeg::webp(),
5749            &RecordingClock::new(),
5750            &ExecOptions::default(),
5751        );
5752
5753        assert_eq!(outcome.artifacts_written, 1);
5754        assert_eq!(outcome.failed(), 0);
5755        // The MP4 was transcoded to WebP, not written verbatim.
5756        let written = fs.read_file("creator/album/cover.webp").unwrap();
5757        assert_ne!(written, b"mp4-bytes");
5758        assert!(written.starts_with(b"RIFF"));
5759        assert_eq!(
5760            albums.get("root").unwrap().folder_webp,
5761            Some(ArtifactState {
5762                path: "creator/album/cover.webp".to_owned(),
5763                hash: "wh".to_owned(),
5764            })
5765        );
5766    }
5767
5768    #[test]
5769    fn folder_mp4_write_keeps_the_source_verbatim() {
5770        let mut manifest = Manifest::new();
5771        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5772        let plan = Plan {
5773            actions: vec![Action::WriteArtifact {
5774                kind: ArtifactKind::FolderMp4,
5775                path: "creator/album/cover.mp4".to_owned(),
5776                source_url: "https://cdn.suno.ai/root/video.mp4".to_owned(),
5777                hash: "mh".to_owned(),
5778                owner_id: "root".to_owned(),
5779                content: None,
5780            }],
5781        };
5782        let http = ScriptedHttp::new().route("root/video.mp4", Reply::ok(b"mp4-bytes".to_vec()));
5783        let fs = MemFs::new();
5784
5785        let outcome = run_with_albums(
5786            &plan,
5787            &mut manifest,
5788            &mut albums,
5789            &[],
5790            &http,
5791            &fs,
5792            &StubFfmpeg::webp(),
5793            &RecordingClock::new(),
5794            &ExecOptions::default(),
5795        );
5796
5797        assert_eq!(outcome.artifacts_written, 1);
5798        assert_eq!(outcome.failed(), 0);
5799        // The raw MP4 is written byte-for-byte, never transcoded.
5800        assert_eq!(
5801            fs.read_file("creator/album/cover.mp4").unwrap(),
5802            b"mp4-bytes"
5803        );
5804        assert_eq!(
5805            albums.get("root").unwrap().folder_mp4,
5806            Some(ArtifactState {
5807                path: "creator/album/cover.mp4".to_owned(),
5808                hash: "mh".to_owned(),
5809            })
5810        );
5811    }
5812
5813    #[test]
5814    fn both_folder_covers_fetch_the_video_cover_once() {
5815        let mut manifest = Manifest::new();
5816        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5817        // `both` retention keeps cover.webp (transcoded) and cover.mp4 (raw) from
5818        // the one video_cover_url. FolderWebp sorts first and caches the fetched
5819        // source; FolderMp4 drains it, so the source is fetched exactly once.
5820        let plan = Plan {
5821            actions: vec![
5822                Action::WriteArtifact {
5823                    kind: ArtifactKind::FolderWebp,
5824                    path: "creator/album/cover.webp".to_owned(),
5825                    source_url: "https://cdn.suno.ai/root/video.mp4".to_owned(),
5826                    hash: "wh".to_owned(),
5827                    owner_id: "root".to_owned(),
5828                    content: None,
5829                },
5830                Action::WriteArtifact {
5831                    kind: ArtifactKind::FolderMp4,
5832                    path: "creator/album/cover.mp4".to_owned(),
5833                    source_url: "https://cdn.suno.ai/root/video.mp4".to_owned(),
5834                    hash: "mh".to_owned(),
5835                    owner_id: "root".to_owned(),
5836                    content: None,
5837                },
5838            ],
5839        };
5840        let http = ScriptedHttp::new().route("root/video.mp4", Reply::ok(b"mp4-bytes".to_vec()));
5841        let fs = MemFs::new();
5842
5843        let outcome = run_with_albums(
5844            &plan,
5845            &mut manifest,
5846            &mut albums,
5847            &[],
5848            &http,
5849            &fs,
5850            &StubFfmpeg::webp(),
5851            &RecordingClock::new(),
5852            &ExecOptions::default(),
5853        );
5854
5855        assert_eq!(outcome.artifacts_written, 2);
5856        assert_eq!(outcome.failed(), 0);
5857        // Fetched exactly once despite two artifacts consuming it (#90 / #89).
5858        assert_eq!(http.count("root/video.mp4"), 1);
5859        // The webp is transcoded; the mp4 is the raw source verbatim.
5860        assert!(
5861            fs.read_file("creator/album/cover.webp")
5862                .unwrap()
5863                .starts_with(b"RIFF")
5864        );
5865        assert_eq!(
5866            fs.read_file("creator/album/cover.mp4").unwrap(),
5867            b"mp4-bytes"
5868        );
5869    }
5870
5871    #[test]
5872    fn folder_art_delete_clears_album_state() {
5873        let fs = MemFs::new().with_file("creator/album/folder.jpg", b"jpg".to_vec());
5874        let mut manifest = Manifest::new();
5875        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5876        albums.insert(
5877            "root".to_owned(),
5878            AlbumArt {
5879                folder_jpg: Some(ArtifactState {
5880                    path: "creator/album/folder.jpg".to_owned(),
5881                    hash: "jh".to_owned(),
5882                }),
5883                folder_webp: None,
5884                folder_mp4: None,
5885            },
5886        );
5887        let plan = Plan {
5888            actions: vec![Action::DeleteArtifact {
5889                kind: ArtifactKind::FolderJpg,
5890                path: "creator/album/folder.jpg".to_owned(),
5891                owner_id: "root".to_owned(),
5892            }],
5893        };
5894
5895        let outcome = run_with_albums(
5896            &plan,
5897            &mut manifest,
5898            &mut albums,
5899            &[],
5900            &ScriptedHttp::new(),
5901            &fs,
5902            &StubFfmpeg::flac(),
5903            &RecordingClock::new(),
5904            &ExecOptions::default(),
5905        );
5906
5907        assert_eq!(outcome.artifacts_deleted, 1);
5908        assert!(!fs.exists("creator/album/folder.jpg"));
5909        // The album row had only the one kind, so it is pruned entirely.
5910        assert!(!albums.contains_key("root"));
5911    }
5912
5913    // ── Phase 9: playlist artifacts ─────────────────────────────────
5914
5915    #[test]
5916    fn playlist_write_uses_inline_content_and_records_state() {
5917        // A playlist body is generated, carried inline. With an empty manifest
5918        // and NO http routes, the write still succeeds — proving it skipped the
5919        // network — and records the playlist store keyed by the playlist id.
5920        let mut manifest = Manifest::new();
5921        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5922        let mut playlists: BTreeMap<String, PlaylistState> = BTreeMap::new();
5923        let body = "#EXTM3U\n#PLAYLIST:Road Trip\n#EXTINF:60,One\nA/One.flac\n";
5924        let plan = Plan {
5925            actions: vec![Action::WriteArtifact {
5926                kind: ArtifactKind::Playlist,
5927                path: "Road Trip.m3u8".to_owned(),
5928                source_url: String::new(),
5929                hash: "ph1".to_owned(),
5930                owner_id: "pl1".to_owned(),
5931                content: Some(body.to_owned()),
5932            }],
5933        };
5934        let fs = MemFs::new();
5935
5936        let outcome = run_full(
5937            &plan,
5938            &mut manifest,
5939            &mut albums,
5940            &mut playlists,
5941            &[],
5942            &ScriptedHttp::new(),
5943            &fs,
5944            &StubFfmpeg::flac(),
5945            &RecordingClock::new(),
5946            &ExecOptions::default(),
5947        );
5948
5949        assert_eq!(outcome.artifacts_written, 1);
5950        assert_eq!(outcome.failed(), 0);
5951        // The exact inline bytes were written, verbatim.
5952        assert_eq!(fs.read_file("Road Trip.m3u8").unwrap(), body.as_bytes());
5953        assert_eq!(
5954            playlists.get("pl1"),
5955            Some(&PlaylistState {
5956                name: "Road Trip".to_owned(),
5957                path: "Road Trip.m3u8".to_owned(),
5958                hash: "ph1".to_owned(),
5959            })
5960        );
5961    }
5962
5963    #[test]
5964    fn playlist_delete_removes_file_and_clears_state() {
5965        let fs = MemFs::new().with_file("Old.m3u8", b"#EXTM3U\n".to_vec());
5966        let mut manifest = Manifest::new();
5967        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
5968        let mut playlists: BTreeMap<String, PlaylistState> = BTreeMap::new();
5969        playlists.insert(
5970            "pl1".to_owned(),
5971            PlaylistState {
5972                name: "Old".to_owned(),
5973                path: "Old.m3u8".to_owned(),
5974                hash: "ph1".to_owned(),
5975            },
5976        );
5977        let plan = Plan {
5978            actions: vec![Action::DeleteArtifact {
5979                kind: ArtifactKind::Playlist,
5980                path: "Old.m3u8".to_owned(),
5981                owner_id: "pl1".to_owned(),
5982            }],
5983        };
5984
5985        let outcome = run_full(
5986            &plan,
5987            &mut manifest,
5988            &mut albums,
5989            &mut playlists,
5990            &[],
5991            &ScriptedHttp::new(),
5992            &fs,
5993            &StubFfmpeg::flac(),
5994            &RecordingClock::new(),
5995            &ExecOptions::default(),
5996        );
5997
5998        assert_eq!(outcome.artifacts_deleted, 1);
5999        assert!(!fs.exists("Old.m3u8"));
6000        assert!(
6001            !playlists.contains_key("pl1"),
6002            "the playlist row is cleared on delete"
6003        );
6004    }
6005
6006    // ── Phase 10: old-sidecar cleanup on move + empty-dir prune ──────
6007
6008    #[test]
6009    fn rename_move_relocates_cover_and_prunes_old_album() {
6010        // A title/album change moves the audio (Rename) and re-emits the cover
6011        // at the NEW path. The old cover must be removed and the now-empty old
6012        // album directory pruned, leaving no orphan sidecar and no ghost dir.
6013        let mut manifest = Manifest::new();
6014        let mut e = entry("Creator/AlbumA/song.flac", AudioFormat::Flac);
6015        e.cover_jpg = Some(ArtifactState {
6016            path: "Creator/AlbumA/cover.jpg".to_owned(),
6017            hash: "h1".to_owned(),
6018        });
6019        manifest.insert("a", e);
6020        let fs = MemFs::new()
6021            .with_file("Creator/AlbumA/song.flac", b"AUDIO".to_vec())
6022            .with_file("Creator/AlbumA/cover.jpg", b"old-jpg".to_vec());
6023        let plan = Plan {
6024            actions: vec![
6025                Action::Rename {
6026                    from: "Creator/AlbumA/song.flac".to_owned(),
6027                    to: "Creator/AlbumB/song.flac".to_owned(),
6028                },
6029                Action::WriteArtifact {
6030                    kind: ArtifactKind::CoverJpg,
6031                    path: "Creator/AlbumB/cover.jpg".to_owned(),
6032                    source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
6033                    hash: "h1".to_owned(),
6034                    owner_id: "a".to_owned(),
6035                    content: None,
6036                },
6037            ],
6038        };
6039        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"new-jpg".to_vec()));
6040
6041        let outcome = run(
6042            &plan,
6043            &mut manifest,
6044            &[],
6045            &http,
6046            &fs,
6047            &StubFfmpeg::flac(),
6048            &RecordingClock::new(),
6049            &ExecOptions::default(),
6050        );
6051
6052        assert_eq!(outcome.failed(), 0);
6053        // Audio moved, the new cover was written, the old cover removed.
6054        assert!(fs.exists("Creator/AlbumB/song.flac"));
6055        assert_eq!(
6056            fs.read_file("Creator/AlbumB/cover.jpg").unwrap(),
6057            b"new-jpg"
6058        );
6059        assert!(!fs.exists("Creator/AlbumA/cover.jpg"));
6060        assert!(!fs.exists("Creator/AlbumA/song.flac"));
6061        // The manifest cover slot now points at the new path.
6062        assert_eq!(
6063            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
6064            "Creator/AlbumB/cover.jpg"
6065        );
6066        // The emptied old album directory is pruned; the new one survives.
6067        assert!(!fs.has_dir("Creator/AlbumA"));
6068        assert!(fs.has_dir("Creator/AlbumB"));
6069    }
6070
6071    #[test]
6072    fn rename_move_relocates_folder_art_and_prunes_old_album() {
6073        // An album rename moves folder.jpg: the old file is removed, the album
6074        // store slot advanced to the new path, and the emptied dir pruned.
6075        let mut manifest = Manifest::new();
6076        let mut albums: BTreeMap<String, AlbumArt> = BTreeMap::new();
6077        albums.insert(
6078            "root".to_owned(),
6079            AlbumArt {
6080                folder_jpg: Some(ArtifactState {
6081                    path: "Creator/AlbumA/folder.jpg".to_owned(),
6082                    hash: "jh".to_owned(),
6083                }),
6084                folder_webp: None,
6085                folder_mp4: None,
6086            },
6087        );
6088        let fs = MemFs::new().with_file("Creator/AlbumA/folder.jpg", b"old-folder".to_vec());
6089        let plan = Plan {
6090            actions: vec![Action::WriteArtifact {
6091                kind: ArtifactKind::FolderJpg,
6092                path: "Creator/AlbumB/folder.jpg".to_owned(),
6093                source_url: "https://art.suno.ai/root/large.jpg".to_owned(),
6094                hash: "jh".to_owned(),
6095                owner_id: "root".to_owned(),
6096                content: None,
6097            }],
6098        };
6099        let http = ScriptedHttp::new().route("root/large.jpg", Reply::ok(b"new-folder".to_vec()));
6100
6101        let outcome = run_with_albums(
6102            &plan,
6103            &mut manifest,
6104            &mut albums,
6105            &[],
6106            &http,
6107            &fs,
6108            &StubFfmpeg::flac(),
6109            &RecordingClock::new(),
6110            &ExecOptions::default(),
6111        );
6112
6113        assert_eq!(outcome.failed(), 0);
6114        assert_eq!(
6115            fs.read_file("Creator/AlbumB/folder.jpg").unwrap(),
6116            b"new-folder"
6117        );
6118        assert!(!fs.exists("Creator/AlbumA/folder.jpg"));
6119        assert_eq!(
6120            albums
6121                .get("root")
6122                .unwrap()
6123                .folder_jpg
6124                .as_ref()
6125                .unwrap()
6126                .path,
6127            "Creator/AlbumB/folder.jpg"
6128        );
6129        assert!(!fs.has_dir("Creator/AlbumA"));
6130        assert!(fs.has_dir("Creator/AlbumB"));
6131    }
6132
6133    #[test]
6134    fn prune_empty_dirs_removes_only_empty_dirs() {
6135        // A direct exercise of the prune port's safety guarantees on a mixed
6136        // tree: nested empties go, anything holding a file (hidden ones too)
6137        // stays, and no file is touched.
6138        let fs = MemFs::new()
6139            .with_file("keep/full/song.flac", b"x".to_vec())
6140            .with_file("hidden/.suno-manifest.json", b"{}".to_vec())
6141            .with_dir("empty/leaf")
6142            .with_dir("nested/a/b/c");
6143
6144        fs.prune_empty_dirs("").unwrap();
6145
6146        // Every empty directory, however deeply nested, is pruned bottom-up.
6147        for gone in [
6148            "empty",
6149            "empty/leaf",
6150            "nested",
6151            "nested/a",
6152            "nested/a/b",
6153            "nested/a/b/c",
6154        ] {
6155            assert!(!fs.has_dir(gone), "empty dir {gone} should be pruned");
6156        }
6157        // A directory holding any file — including only a hidden dotfile — stays.
6158        assert!(fs.has_dir("keep"));
6159        assert!(fs.has_dir("keep/full"));
6160        assert!(fs.has_dir("hidden"));
6161        // No file was touched.
6162        assert!(fs.exists("keep/full/song.flac"));
6163        assert!(fs.exists("hidden/.suno-manifest.json"));
6164    }
6165
6166    #[test]
6167    fn prune_empty_dirs_never_removes_the_named_root() {
6168        // Pruning under a named root clears its empty children but keeps the
6169        // root itself, even when the root is now empty.
6170        let fs = MemFs::new().with_dir("empty/leaf");
6171        fs.prune_empty_dirs("empty").unwrap();
6172        assert!(fs.has_dir("empty"), "the named root is never removed");
6173        assert!(!fs.has_dir("empty/leaf"));
6174    }
6175
6176    #[test]
6177    fn old_sidecar_remove_failure_is_per_clip_and_converges_next_run() {
6178        // If removing the old sidecar fails, the write is a per-clip failure
6179        // that never aborts the run and does NOT advance the state slot, so the
6180        // next identical run re-attempts the cleanup and the tree converges.
6181        let mut manifest = Manifest::new();
6182        let mut e = entry("a.flac", AudioFormat::Flac);
6183        e.cover_jpg = Some(ArtifactState {
6184            path: "AlbumA/cover.jpg".to_owned(),
6185            hash: "h1".to_owned(),
6186        });
6187        manifest.insert("a", e);
6188        let fs = MemFs::new()
6189            .with_file("a.flac", b"AUDIO".to_vec())
6190            .with_file("AlbumA/cover.jpg", b"old".to_vec());
6191        let plan = Plan {
6192            actions: vec![Action::WriteArtifact {
6193                kind: ArtifactKind::CoverJpg,
6194                path: "AlbumB/cover.jpg".to_owned(),
6195                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
6196                hash: "h1".to_owned(),
6197                owner_id: "a".to_owned(),
6198                content: None,
6199            }],
6200        };
6201        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"new".to_vec()));
6202
6203        // Run 1: the old-cover remove is forced to fail.
6204        fs.arm_fail_remove("AlbumA/cover.jpg");
6205        let first = run(
6206            &plan,
6207            &mut manifest,
6208            &[],
6209            &http,
6210            &fs,
6211            &StubFfmpeg::flac(),
6212            &RecordingClock::new(),
6213            &ExecOptions::default(),
6214        );
6215        assert_eq!(
6216            first.status,
6217            RunStatus::Completed,
6218            "a remove failure never aborts the run"
6219        );
6220        assert_eq!(first.failed(), 1);
6221        // The new cover is written but the old one lingers and the slot is stale.
6222        assert!(fs.exists("AlbumB/cover.jpg"));
6223        assert!(fs.exists("AlbumA/cover.jpg"));
6224        assert_eq!(
6225            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
6226            "AlbumA/cover.jpg"
6227        );
6228        assert!(fs.has_dir("AlbumA"), "the orphan keeps its directory alive");
6229
6230        // Run 2: the same plan re-runs with the fault cleared and converges.
6231        fs.disarm_fail_remove("AlbumA/cover.jpg");
6232        let second = run(
6233            &plan,
6234            &mut manifest,
6235            &[],
6236            &http,
6237            &fs,
6238            &StubFfmpeg::flac(),
6239            &RecordingClock::new(),
6240            &ExecOptions::default(),
6241        );
6242        assert_eq!(second.failed(), 0);
6243        assert!(fs.exists("AlbumB/cover.jpg"));
6244        assert!(!fs.exists("AlbumA/cover.jpg"), "no orphan persists");
6245        assert_eq!(
6246            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().path,
6247            "AlbumB/cover.jpg"
6248        );
6249        assert!(!fs.has_dir("AlbumA"), "the emptied directory is pruned");
6250    }
6251
6252    #[test]
6253    fn same_path_artifact_rewrite_does_no_remove_and_prunes_nothing() {
6254        // The idempotent case: a content-only cover rewrite (hash drift, path
6255        // unchanged) attempts no remove and prunes no live directory. A remove
6256        // failure is armed on the cover path, so any spurious remove would
6257        // surface as a failure — none does.
6258        let mut manifest = Manifest::new();
6259        let mut e = entry("Album/a.mp3", AudioFormat::Mp3);
6260        e.cover_jpg = Some(ArtifactState {
6261            path: "Album/cover.jpg".to_owned(),
6262            hash: "h1".to_owned(),
6263        });
6264        manifest.insert("a", e);
6265        let fs = MemFs::new()
6266            .with_file("Album/a.mp3", b"AUDIO".to_vec())
6267            .with_file("Album/cover.jpg", b"old".to_vec());
6268        fs.arm_fail_remove("Album/cover.jpg");
6269        let plan = Plan {
6270            actions: vec![Action::WriteArtifact {
6271                kind: ArtifactKind::CoverJpg,
6272                path: "Album/cover.jpg".to_owned(),
6273                source_url: "https://art.suno.ai/a/large.jpg".to_owned(),
6274                hash: "h2".to_owned(),
6275                owner_id: "a".to_owned(),
6276                content: None,
6277            }],
6278        };
6279        let http = ScriptedHttp::new().route("a/large.jpg", Reply::ok(b"new".to_vec()));
6280
6281        let outcome = run(
6282            &plan,
6283            &mut manifest,
6284            &[],
6285            &http,
6286            &fs,
6287            &StubFfmpeg::flac(),
6288            &RecordingClock::new(),
6289            &ExecOptions::default(),
6290        );
6291
6292        assert_eq!(
6293            outcome.failed(),
6294            0,
6295            "no remove is attempted, so the armed failure never fires"
6296        );
6297        assert_eq!(outcome.artifacts_written, 1);
6298        assert_eq!(fs.read_file("Album/cover.jpg").unwrap(), b"new");
6299        assert_eq!(
6300            manifest.get("a").unwrap().cover_jpg.as_ref().unwrap().hash,
6301            "h2"
6302        );
6303        // The live directory is untouched by prune.
6304        assert!(fs.has_dir("Album"));
6305    }
6306
6307    // ── Concurrency (issue #22) ─────────────────────────────────────
6308
6309    mod concurrency {
6310        use super::*;
6311        use crate::ffmpeg::FfmpegError;
6312        use crate::fs::{FileStat, FsError};
6313        use crate::http::{HttpRequest, TransportError};
6314        use std::future::Future;
6315        use std::pin::Pin;
6316        use std::sync::Arc;
6317        use std::sync::atomic::{AtomicUsize, Ordering};
6318        use std::task::{Context, Poll};
6319
6320        /// A future that pends exactly once before resolving, waking itself so a
6321        /// single-threaded executor re-polls. It forces the [`Http`] port to
6322        /// yield, so [`buffer_unordered`](futures_util::stream::StreamExt) parks
6323        /// each in-flight request and the true overlap becomes observable.
6324        #[derive(Default)]
6325        struct YieldOnce {
6326            yielded: bool,
6327        }
6328
6329        impl Future for YieldOnce {
6330            type Output = ();
6331            fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
6332                if self.yielded {
6333                    Poll::Ready(())
6334                } else {
6335                    self.yielded = true;
6336                    cx.waker().wake_by_ref();
6337                    Poll::Pending
6338                }
6339            }
6340        }
6341
6342        /// An [`Http`] double that wraps [`ScriptedHttp`] and records the peak
6343        /// number of concurrently in-flight requests. Each `send` bumps a live
6344        /// counter, yields once (so peers can start), then delegates.
6345        struct GatedHttp {
6346            inner: ScriptedHttp,
6347            inflight: Arc<AtomicUsize>,
6348            peak: Arc<AtomicUsize>,
6349        }
6350
6351        impl GatedHttp {
6352            fn new(inner: ScriptedHttp) -> Self {
6353                Self {
6354                    inner,
6355                    inflight: Arc::new(AtomicUsize::new(0)),
6356                    peak: Arc::new(AtomicUsize::new(0)),
6357                }
6358            }
6359
6360            fn peak(&self) -> usize {
6361                self.peak.load(Ordering::SeqCst)
6362            }
6363
6364            fn count(&self, needle: &str) -> usize {
6365                self.inner.count(needle)
6366            }
6367        }
6368
6369        impl Http for GatedHttp {
6370            async fn send(&self, request: HttpRequest) -> Result<HttpResponse, TransportError> {
6371                let now = self.inflight.fetch_add(1, Ordering::SeqCst) + 1;
6372                self.peak.fetch_max(now, Ordering::SeqCst);
6373                YieldOnce::default().await;
6374                let out = self.inner.send(request).await;
6375                self.inflight.fetch_sub(1, Ordering::SeqCst);
6376                out
6377            }
6378        }
6379
6380        fn download(id: &str, format: AudioFormat) -> (Clip, Desired, Action) {
6381            let c = clip(id);
6382            let d = desired(c.clone(), format);
6383            let action = Action::Download {
6384                clip: c.clone(),
6385                lineage: LineageContext::own_root(&c),
6386                path: d.path.clone(),
6387                format,
6388            };
6389            (c, d, action)
6390        }
6391
6392        fn opts_with(concurrency: u32) -> ExecOptions {
6393            ExecOptions {
6394                concurrency,
6395                ..small_poll()
6396            }
6397        }
6398
6399        #[test]
6400        fn concurrency_never_exceeds_the_configured_bound() {
6401            let count = 6;
6402            let concurrency = 3;
6403            let mut scripted = ScriptedHttp::new().with_auth();
6404            let mut actions = Vec::new();
6405            let mut desireds = Vec::new();
6406            for i in 0..count {
6407                let id = format!("c{i}");
6408                scripted = scripted.route(&format!("{id}.mp3"), Reply::ok(b"mp3-body".to_vec()));
6409                let (_c, d, action) = download(&id, AudioFormat::Mp3);
6410                actions.push(action);
6411                desireds.push(d);
6412            }
6413            let http = GatedHttp::new(scripted);
6414            let fs = MemFs::new();
6415            let plan = Plan { actions };
6416            let mut manifest = Manifest::new();
6417
6418            let outcome = run_gated_fs(
6419                &plan,
6420                &mut manifest,
6421                &desireds,
6422                &http,
6423                &fs,
6424                &opts_with(concurrency),
6425            );
6426
6427            assert_eq!(outcome.downloaded, count);
6428            assert!(
6429                http.peak() <= concurrency as usize,
6430                "peak {} exceeded the bound {concurrency}",
6431                http.peak()
6432            );
6433            assert_eq!(
6434                http.peak(),
6435                concurrency as usize,
6436                "expected the run to saturate the bound"
6437            );
6438        }
6439
6440        /// Run a gated plan against a caller-supplied [`MemFs`], returning the
6441        /// outcome. The client is built here so the limiter can be inspected by
6442        /// the caller-facing variant below.
6443        fn run_gated_fs(
6444            plan: &Plan,
6445            manifest: &mut Manifest,
6446            desired: &[Desired],
6447            http: &GatedHttp,
6448            fs: &MemFs,
6449            opts: &ExecOptions,
6450        ) -> ExecOutcome {
6451            let ffmpeg = StubFfmpeg::flac();
6452            let clock = RecordingClock::new();
6453            let mut albums = BTreeMap::new();
6454            let mut playlists = BTreeMap::new();
6455            let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
6456            pollster::block_on(execute(
6457                plan,
6458                manifest,
6459                &mut albums,
6460                &mut playlists,
6461                desired,
6462                &HashMap::new(),
6463                Ports {
6464                    client: &client,
6465                    http,
6466                    fs,
6467                    ffmpeg: &ffmpeg,
6468                    clock: &clock,
6469                },
6470                opts,
6471            ))
6472        }
6473
6474        #[test]
6475        fn a_failing_clip_does_not_abort_the_others() {
6476            let mut scripted = ScriptedHttp::new().with_auth();
6477            scripted = scripted
6478                .route("ok1.mp3", Reply::ok(b"one".to_vec()))
6479                .route("bad.mp3", Reply::status(404))
6480                .route("ok2.mp3", Reply::ok(b"two".to_vec()));
6481            let (_a, d1, a1) = download("ok1", AudioFormat::Mp3);
6482            let (_b, d2, a2) = download("bad", AudioFormat::Mp3);
6483            let (_c, d3, a3) = download("ok2", AudioFormat::Mp3);
6484            let http = GatedHttp::new(scripted);
6485            let fs = MemFs::new();
6486            let plan = Plan {
6487                actions: vec![a1, a2, a3],
6488            };
6489            let mut manifest = Manifest::new();
6490
6491            let outcome = run_gated_fs(
6492                &plan,
6493                &mut manifest,
6494                &[d1, d2, d3],
6495                &http,
6496                &fs,
6497                &opts_with(3),
6498            );
6499
6500            assert_eq!(outcome.downloaded, 2);
6501            assert_eq!(outcome.failed(), 1);
6502            assert_eq!(outcome.status, RunStatus::Completed);
6503            assert_eq!(outcome.failures[0].clip_id, "bad");
6504            assert!(manifest.get("ok1").is_some());
6505            assert!(manifest.get("ok2").is_some());
6506            assert!(manifest.get("bad").is_none());
6507        }
6508
6509        #[test]
6510        fn outcome_is_identical_across_concurrency_levels() {
6511            // A plan mixing successful and failing downloads with serial phase-2
6512            // actions (a skip and a delete), so both phases contribute.
6513            fn build() -> (Plan, Vec<Desired>) {
6514                let mut actions = Vec::new();
6515                let mut desireds = Vec::new();
6516                for id in ["a", "b", "c", "d"] {
6517                    let (_c, d, action) = download(id, AudioFormat::Mp3);
6518                    actions.push(action);
6519                    desireds.push(d);
6520                }
6521                // A failing download in the middle of the audio set.
6522                let (_e, de, ae) = download("fail", AudioFormat::Mp3);
6523                actions.insert(2, ae);
6524                desireds.push(de);
6525                // Phase-2 actions.
6526                actions.push(Action::Skip {
6527                    clip_id: "gone".to_owned(),
6528                });
6529                actions.push(Action::Delete {
6530                    path: "old.mp3".to_owned(),
6531                    clip_id: "old".to_owned(),
6532                });
6533                (Plan { actions }, desireds)
6534            }
6535
6536            fn http() -> ScriptedHttp {
6537                ScriptedHttp::new()
6538                    .with_auth()
6539                    .route("a.mp3", Reply::ok(b"a".to_vec()))
6540                    .route("b.mp3", Reply::ok(b"b".to_vec()))
6541                    .route("c.mp3", Reply::ok(b"c".to_vec()))
6542                    .route("d.mp3", Reply::ok(b"d".to_vec()))
6543                    .route("fail.mp3", Reply::status(404))
6544            }
6545
6546            fn seed_manifest() -> Manifest {
6547                let mut m = Manifest::new();
6548                m.insert("old".to_owned(), entry("old.mp3", AudioFormat::Mp3));
6549                m
6550            }
6551
6552            let (plan, desireds) = build();
6553
6554            let mut m1 = seed_manifest();
6555            let fs1 = MemFs::new().with_file("old.mp3", b"x".to_vec());
6556            let out1 = run_gated_fs(
6557                &plan,
6558                &mut m1,
6559                &desireds,
6560                &GatedHttp::new(http()),
6561                &fs1,
6562                &opts_with(1),
6563            );
6564
6565            let mut m8 = seed_manifest();
6566            let fs8 = MemFs::new().with_file("old.mp3", b"x".to_vec());
6567            let out8 = run_gated_fs(
6568                &plan,
6569                &mut m8,
6570                &desireds,
6571                &GatedHttp::new(http()),
6572                &fs8,
6573                &opts_with(8),
6574            );
6575
6576            assert_eq!(out1, out8, "outcome must not depend on concurrency");
6577            assert_eq!(m1, m8, "final manifest must not depend on concurrency");
6578            assert_eq!(out8.downloaded, 4);
6579            assert_eq!(out8.deleted, 1);
6580            assert_eq!(out8.skipped, 1);
6581            assert_eq!(out8.failed(), 1);
6582        }
6583
6584        #[test]
6585        fn a_systemic_disk_full_aborts_promptly() {
6586            let count = 8;
6587            let concurrency = 2;
6588            let mut scripted = ScriptedHttp::new().with_auth();
6589            let mut actions = Vec::new();
6590            let mut desireds = Vec::new();
6591            for i in 0..count {
6592                let id = format!("d{i}");
6593                scripted = scripted.route(&format!("{id}.mp3"), Reply::ok(b"mp3-body".to_vec()));
6594                let (_c, d, action) = download(&id, AudioFormat::Mp3);
6595                actions.push(action);
6596                desireds.push(d);
6597            }
6598            // The very first clip's write hits ENOSPC, a systemic failure.
6599            let fs = MemFs::new().fail_write_out_of_space("d0.mp3");
6600            let http = GatedHttp::new(scripted);
6601            let plan = Plan { actions };
6602            let mut manifest = Manifest::new();
6603
6604            let outcome = run_gated_fs(
6605                &plan,
6606                &mut manifest,
6607                &desireds,
6608                &http,
6609                &fs,
6610                &opts_with(concurrency),
6611            );
6612
6613            assert_eq!(outcome.status, RunStatus::DiskFull);
6614            assert!(
6615                outcome.downloaded < count,
6616                "a systemic abort must stop remaining work, downloaded {}",
6617                outcome.downloaded
6618            );
6619        }
6620
6621        #[test]
6622        fn limiter_records_a_rate_limit_under_concurrent_calls() {
6623            // Three concurrent FLAC renders; exactly one clip is throttled once
6624            // on its wav_file read. The shared limiter must record that single
6625            // 429 (halving 2.0 -> 1.0) with no lost or duplicated update, proving
6626            // the mutex keeps the AIMD state correct under concurrency.
6627            let scripted = ScriptedHttp::new()
6628                .with_auth()
6629                .route_seq(
6630                    "/gen/x/wav_file/",
6631                    vec![
6632                        Reply::status(429),
6633                        Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/x.wav"}"#),
6634                    ],
6635                )
6636                .route(
6637                    "/gen/y/wav_file/",
6638                    Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/y.wav"}"#),
6639                )
6640                .route(
6641                    "/gen/z/wav_file/",
6642                    Reply::json(r#"{"wav_file_url": "https://cdn1.suno.ai/z.wav"}"#),
6643                )
6644                .route("x.wav", Reply::ok(b"wav-x".to_vec()))
6645                .route("y.wav", Reply::ok(b"wav-y".to_vec()))
6646                .route("z.wav", Reply::ok(b"wav-z".to_vec()));
6647
6648            let mut actions = Vec::new();
6649            let mut desireds = Vec::new();
6650            for id in ["x", "y", "z"] {
6651                let (_c, d, action) = download(id, AudioFormat::Flac);
6652                actions.push(action);
6653                desireds.push(d);
6654            }
6655            let plan = Plan { actions };
6656            let fs = MemFs::new();
6657            let ffmpeg = StubFfmpeg::flac();
6658            let clock = RecordingClock::new();
6659            let mut albums = BTreeMap::new();
6660            let mut playlists = BTreeMap::new();
6661            let mut manifest = Manifest::new();
6662            let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
6663
6664            let outcome = pollster::block_on(execute(
6665                &plan,
6666                &mut manifest,
6667                &mut albums,
6668                &mut playlists,
6669                &desireds,
6670                &HashMap::new(),
6671                Ports {
6672                    client: &client,
6673                    http: &scripted,
6674                    fs: &fs,
6675                    ffmpeg: &ffmpeg,
6676                    clock: &clock,
6677                },
6678                &opts_with(3),
6679            ));
6680
6681            assert_eq!(outcome.downloaded, 3);
6682            assert_eq!(outcome.failed(), 0);
6683            assert!(
6684                (client.limiter_rate() - 1.0).abs() < 1e-9,
6685                "one 429 must halve the rate to 1.0, got {}",
6686                client.limiter_rate()
6687            );
6688        }
6689
6690        #[test]
6691        fn a_download_is_committed_in_plan_order_around_a_rename() {
6692            // Plan order: rename "orig" away from shared.mp3 first, then download
6693            // a new clip into shared.mp3. A parallel executor that performed the
6694            // download's destination write off plan order would write shared.mp3
6695            // before the rename ran, letting the rename carry those fresh bytes
6696            // to moved.mp3 and stranding shared.mp3 - corrupting both clips.
6697            // Committing every destination effect serially in plan order keeps
6698            // moved.mp3 = the original and shared.mp3 = the new download.
6699            let c_new = clip("new");
6700            let mut d_new = desired(c_new.clone(), AudioFormat::Mp3);
6701            d_new.path = "shared.mp3".to_owned();
6702            let plan = Plan {
6703                actions: vec![
6704                    Action::Rename {
6705                        from: "shared.mp3".to_owned(),
6706                        to: "moved.mp3".to_owned(),
6707                    },
6708                    Action::Download {
6709                        clip: c_new.clone(),
6710                        lineage: LineageContext::own_root(&c_new),
6711                        path: "shared.mp3".to_owned(),
6712                        format: AudioFormat::Mp3,
6713                    },
6714                ],
6715            };
6716            let scripted = ScriptedHttp::new()
6717                .with_auth()
6718                .route("new.mp3", Reply::ok(b"NEW-BODY".to_vec()));
6719            let http = GatedHttp::new(scripted);
6720            let fs = MemFs::new().with_file("shared.mp3", b"ORIGINAL".to_vec());
6721            let mut manifest = Manifest::new();
6722            manifest.insert("orig", entry("shared.mp3", AudioFormat::Mp3));
6723
6724            let outcome = run_gated_fs(&plan, &mut manifest, &[d_new], &http, &fs, &opts_with(4));
6725
6726            assert_eq!(outcome.renamed, 1);
6727            assert_eq!(outcome.downloaded, 1);
6728            assert_eq!(
6729                fs.read_file("moved.mp3").as_deref(),
6730                Some(&b"ORIGINAL"[..]),
6731                "the rename must carry the original bytes, untouched by the download"
6732            );
6733            let landed = fs.read_file("shared.mp3").expect("new download must land");
6734            assert_ne!(
6735                landed, b"ORIGINAL",
6736                "the new download must replace the moved original, not corrupt it"
6737            );
6738            assert_eq!(manifest.get("orig").unwrap().path, "moved.mp3");
6739            assert_eq!(manifest.get("new").unwrap().path, "shared.mp3");
6740        }
6741
6742        #[test]
6743        fn an_aborted_reformat_leaves_the_old_file_and_manifest_consistent() {
6744            // A systemic disk-full abort strikes the download committed before the
6745            // reformat. Because the reformat's slow render is side-effect-free and
6746            // its destination write + old-file removal only happen in the serial
6747            // commit (which the abort skips), the old file survives and the
6748            // manifest still points at it: no removed-but-referenced file.
6749            let boom = clip("boom");
6750            let mut d_boom = desired(boom.clone(), AudioFormat::Mp3);
6751            d_boom.path = "boom.mp3".to_owned();
6752            let reformer = clip("r");
6753            let d_reformer = desired(reformer.clone(), AudioFormat::Mp3);
6754            let plan = Plan {
6755                actions: vec![
6756                    Action::Download {
6757                        clip: boom.clone(),
6758                        lineage: LineageContext::own_root(&boom),
6759                        path: "boom.mp3".to_owned(),
6760                        format: AudioFormat::Mp3,
6761                    },
6762                    Action::Reformat {
6763                        clip: reformer.clone(),
6764                        path: "r_new.mp3".to_owned(),
6765                        from_path: "r_old.flac".to_owned(),
6766                        from: AudioFormat::Flac,
6767                        to: AudioFormat::Mp3,
6768                    },
6769                ],
6770            };
6771            let scripted = ScriptedHttp::new()
6772                .with_auth()
6773                .route("boom.mp3", Reply::ok(b"boom-body".to_vec()))
6774                .route("r.mp3", Reply::ok(b"reformatted".to_vec()));
6775            let http = GatedHttp::new(scripted);
6776            // The download's write hits ENOSPC, a systemic abort.
6777            let fs = MemFs::new()
6778                .with_file("r_old.flac", b"OLD-FLAC".to_vec())
6779                .fail_write_out_of_space("boom.mp3");
6780            let mut manifest = Manifest::new();
6781            manifest.insert("r", entry("r_old.flac", AudioFormat::Flac));
6782
6783            let outcome = run_gated_fs(
6784                &plan,
6785                &mut manifest,
6786                &[d_boom, d_reformer],
6787                &http,
6788                &fs,
6789                &opts_with(4),
6790            );
6791
6792            assert_eq!(outcome.status, RunStatus::DiskFull);
6793            assert!(
6794                fs.exists("r_old.flac"),
6795                "the old file must survive the abort"
6796            );
6797            assert!(
6798                !fs.exists("r_new.mp3"),
6799                "no reformatted file may be written"
6800            );
6801            let still = manifest.get("r").expect("the manifest must still track r");
6802            assert_eq!(
6803                still.path, "r_old.flac",
6804                "the manifest must still point at the surviving old file"
6805            );
6806            assert_eq!(still.format, AudioFormat::Flac);
6807        }
6808
6809        #[test]
6810        fn a_systemic_abort_leaves_no_untracked_destination_files() {
6811            // Two clips commit, the third's write hits ENOSPC (a systemic abort),
6812            // and the rest never commit. Every file remaining on disk must be one
6813            // the manifest tracks: producers write nothing, so an abort cannot
6814            // strand an untracked file from an in-flight or buffered render.
6815            let mut scripted = ScriptedHttp::new().with_auth();
6816            let mut actions = Vec::new();
6817            let mut desireds = Vec::new();
6818            for id in ["a0", "a1", "boom", "a3", "a4"] {
6819                scripted = scripted.route(&format!("{id}.mp3"), Reply::ok(b"body".to_vec()));
6820                let (_c, d, action) = download(id, AudioFormat::Mp3);
6821                actions.push(action);
6822                desireds.push(d);
6823            }
6824            let http = GatedHttp::new(scripted);
6825            let fs = MemFs::new().fail_write_out_of_space("boom.mp3");
6826            let plan = Plan { actions };
6827            let mut manifest = Manifest::new();
6828
6829            let outcome = run_gated_fs(&plan, &mut manifest, &desireds, &http, &fs, &opts_with(2));
6830
6831            assert_eq!(outcome.status, RunStatus::DiskFull);
6832            let tracked: std::collections::BTreeSet<String> = manifest
6833                .entries
6834                .values()
6835                .map(|entry| entry.path.clone())
6836                .collect();
6837            for path in fs.paths() {
6838                assert!(
6839                    tracked.contains(&path),
6840                    "found an untracked destination file: {path}"
6841                );
6842            }
6843            assert!(
6844                !fs.exists("a3.mp3"),
6845                "uncommitted renders must not be on disk"
6846            );
6847            assert!(
6848                !fs.exists("a4.mp3"),
6849                "uncommitted renders must not be on disk"
6850            );
6851        }
6852
6853        /// An [`Ffmpeg`] double that counts how many rendered FLAC payloads are
6854        /// live: it bumps a shared counter (tracking the peak) when a transcode
6855        /// yields bytes, and [`CountingFs`] drops it back on the committing write.
6856        /// The [transcode, write] window is a superset of the true in-memory hold,
6857        /// so the observed peak upper-bounds the real one.
6858        struct CountingFfmpeg {
6859            inner: StubFfmpeg,
6860            held: Arc<AtomicUsize>,
6861            peak: Arc<AtomicUsize>,
6862        }
6863
6864        impl Ffmpeg for CountingFfmpeg {
6865            fn wav_to_lossless(
6866                &self,
6867                wav: &[u8],
6868                format: AudioFormat,
6869            ) -> impl Future<Output = Result<Vec<u8>, FfmpegError>> + Send {
6870                let fut = self.inner.wav_to_lossless(wav, format);
6871                let held = self.held.clone();
6872                let peak = self.peak.clone();
6873                async move {
6874                    let out = fut.await;
6875                    if out.is_ok() {
6876                        let now = held.fetch_add(1, Ordering::SeqCst) + 1;
6877                        peak.fetch_max(now, Ordering::SeqCst);
6878                    }
6879                    out
6880                }
6881            }
6882
6883            fn mp4_to_webp(
6884                &self,
6885                mp4: &[u8],
6886                settings: WebpEncodeSettings,
6887            ) -> impl Future<Output = Result<Vec<u8>, FfmpegError>> + Send {
6888                self.inner.mp4_to_webp(mp4, settings)
6889            }
6890        }
6891
6892        /// A [`Filesystem`] double wrapping [`MemFs`] that decrements the live
6893        /// payload counter on each committing write, closing the window opened by
6894        /// [`CountingFfmpeg`].
6895        struct CountingFs {
6896            inner: MemFs,
6897            held: Arc<AtomicUsize>,
6898        }
6899
6900        impl Filesystem for CountingFs {
6901            fn write_atomic(&self, path: &str, bytes: &[u8]) -> Result<(), FsError> {
6902                let out = self.inner.write_atomic(path, bytes);
6903                self.held.fetch_sub(1, Ordering::SeqCst);
6904                out
6905            }
6906
6907            fn rename(&self, from: &str, to: &str) -> Result<(), FsError> {
6908                self.inner.rename(from, to)
6909            }
6910
6911            fn remove(&self, path: &str) -> Result<(), FsError> {
6912                self.inner.remove(path)
6913            }
6914
6915            fn prune_empty_dirs(&self, root: &str) -> Result<(), FsError> {
6916                self.inner.prune_empty_dirs(root)
6917            }
6918
6919            fn read(&self, path: &str) -> Result<Vec<u8>, FsError> {
6920                self.inner.read(path)
6921            }
6922
6923            fn metadata(&self, path: &str) -> Option<FileStat> {
6924                self.inner.metadata(path)
6925            }
6926        }
6927
6928        #[test]
6929        fn rendered_payloads_in_memory_stay_bounded_by_concurrency() {
6930            // Far more FLAC clips than the concurrency bound. The ordered buffered
6931            // render keeps at most about `concurrency` transcoded payloads live at
6932            // once (never the whole library), so peak held <= concurrency + 1.
6933            let count = 12;
6934            let concurrency = 3;
6935            let mut scripted = ScriptedHttp::new().with_auth();
6936            let mut actions = Vec::new();
6937            let mut desireds = Vec::new();
6938            for i in 0..count {
6939                let id = format!("f{i}");
6940                scripted = scripted
6941                    .route(
6942                        &format!("/gen/{id}/wav_file/"),
6943                        Reply::json(&format!(
6944                            r#"{{"wav_file_url": "https://cdn1.suno.ai/{id}.wav"}}"#
6945                        )),
6946                    )
6947                    .route(&format!("{id}.wav"), Reply::ok(b"wav-body".to_vec()));
6948                let (_c, d, action) = download(&id, AudioFormat::Flac);
6949                actions.push(action);
6950                desireds.push(d);
6951            }
6952            let http = GatedHttp::new(scripted);
6953            let held = Arc::new(AtomicUsize::new(0));
6954            let peak = Arc::new(AtomicUsize::new(0));
6955            let ffmpeg = CountingFfmpeg {
6956                inner: StubFfmpeg::flac(),
6957                held: held.clone(),
6958                peak: peak.clone(),
6959            };
6960            let fs = CountingFs {
6961                inner: MemFs::new(),
6962                held: held.clone(),
6963            };
6964            let clock = RecordingClock::new();
6965            let mut albums = BTreeMap::new();
6966            let mut playlists = BTreeMap::new();
6967            let mut manifest = Manifest::new();
6968            let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
6969            let plan = Plan { actions };
6970
6971            let outcome = pollster::block_on(execute(
6972                &plan,
6973                &mut manifest,
6974                &mut albums,
6975                &mut playlists,
6976                &desireds,
6977                &HashMap::new(),
6978                Ports {
6979                    client: &client,
6980                    http: &http,
6981                    fs: &fs,
6982                    ffmpeg: &ffmpeg,
6983                    clock: &clock,
6984                },
6985                &opts_with(concurrency),
6986            ));
6987
6988            assert_eq!(outcome.downloaded, count as usize);
6989            assert_eq!(
6990                held.load(Ordering::SeqCst),
6991                0,
6992                "every payload must be committed"
6993            );
6994            assert!(
6995                peak.load(Ordering::SeqCst) <= concurrency as usize + 1,
6996                "peak live payloads {} exceeded the bound {}",
6997                peak.load(Ordering::SeqCst),
6998                concurrency + 1
6999            );
7000            assert!(
7001                peak.load(Ordering::SeqCst) >= 2,
7002                "the render should genuinely overlap, peak was {}",
7003                peak.load(Ordering::SeqCst)
7004            );
7005        }
7006
7007        #[test]
7008        fn artifact_fetches_run_concurrently() {
7009            // Four CoverJpg sidecars whose owning clips are already in the manifest.
7010            // With concurrency=2 the two HTTP fetches should overlap, so the peak
7011            // in-flight count must reach at least 2.
7012            let count = 4usize;
7013            let concurrency = 2u32;
7014            let mut scripted = ScriptedHttp::new().with_auth();
7015            let mut actions = Vec::new();
7016            let mut manifest = Manifest::new();
7017            for i in 0..count {
7018                let id = format!("a{i}");
7019                scripted = scripted.route(&format!("{id}.jpg"), Reply::ok(b"jpg-bytes".to_vec()));
7020                manifest.insert(&id, entry(&format!("{id}.mp3"), AudioFormat::Mp3));
7021                actions.push(Action::WriteArtifact {
7022                    kind: ArtifactKind::CoverJpg,
7023                    path: format!("{id}/cover.jpg"),
7024                    source_url: format!("https://art.suno.ai/{id}.jpg"),
7025                    hash: format!("h{i}"),
7026                    owner_id: id,
7027                    content: None,
7028                });
7029            }
7030            let http = GatedHttp::new(scripted);
7031            let fs = MemFs::new();
7032            let plan = Plan { actions };
7033
7034            let outcome = run_gated_fs(
7035                &plan,
7036                &mut manifest,
7037                &[],
7038                &http,
7039                &fs,
7040                &opts_with(concurrency),
7041            );
7042
7043            assert_eq!(outcome.artifacts_written, count);
7044            assert_eq!(outcome.failed(), 0);
7045            assert!(
7046                http.peak() >= concurrency as usize,
7047                "artifact fetches must overlap: peak {} < concurrency {}",
7048                http.peak(),
7049                concurrency,
7050            );
7051        }
7052
7053        #[test]
7054        fn stem_fetches_run_concurrently() {
7055            // Four Mp3 stem fetches whose owning clips are in the manifest.
7056            // With concurrency=2 the peak in-flight HTTP count must reach at least 2.
7057            let count = 4usize;
7058            let concurrency = 2u32;
7059            let mut scripted = ScriptedHttp::new().with_auth();
7060            let mut actions = Vec::new();
7061            let mut manifest = Manifest::new();
7062            for i in 0..count {
7063                let id = format!("s{i}");
7064                scripted =
7065                    scripted.route(&format!("{id}voc.mp3"), Reply::ok(b"stem-bytes".to_vec()));
7066                manifest.insert(&id, entry(&format!("{id}.mp3"), AudioFormat::Mp3));
7067                actions.push(Action::WriteStem {
7068                    clip_id: id.clone(),
7069                    key: "voc".to_owned(),
7070                    stem_id: format!("{id}voc"),
7071                    path: format!("{id}.stems/voc.mp3"),
7072                    source_url: format!("https://cdn1.suno.ai/{id}voc.mp3"),
7073                    format: StemFormat::Mp3,
7074                    hash: format!("h{i}"),
7075                });
7076            }
7077            let http = GatedHttp::new(scripted);
7078            let fs = MemFs::new();
7079            let plan = Plan { actions };
7080
7081            let outcome = run_gated_fs(
7082                &plan,
7083                &mut manifest,
7084                &[],
7085                &http,
7086                &fs,
7087                &opts_with(concurrency),
7088            );
7089
7090            assert_eq!(outcome.artifacts_written, count);
7091            assert_eq!(outcome.failed(), 0);
7092            assert!(
7093                http.peak() >= concurrency as usize,
7094                "stem fetches must overlap: peak {} < concurrency {}",
7095                http.peak(),
7096                concurrency,
7097            );
7098        }
7099
7100        #[test]
7101        fn prepareable_outcome_is_identical_across_concurrency_levels_with_artifacts_and_stems() {
7102            // A plan mixing downloads, artifact writes, and stem writes. Both a
7103            // failing clip and a serial-only action (delete) are included so all
7104            // code paths contribute. Outcome and final manifest must be the same
7105            // whether concurrency is 1 or 8, proving commits remain serial and
7106            // deterministic while preparation runs in parallel.
7107            fn build() -> (Plan, Vec<Desired>) {
7108                let mut actions = Vec::new();
7109                let mut desireds = Vec::new();
7110                for id in ["x", "y", "z"] {
7111                    let (_c, d, action) = download(id, AudioFormat::Mp3);
7112                    desireds.push(d);
7113                    actions.push(action);
7114                    // A CoverJpg sidecar for each clip.
7115                    actions.push(Action::WriteArtifact {
7116                        kind: ArtifactKind::CoverJpg,
7117                        path: format!("{id}/cover.jpg"),
7118                        source_url: format!("https://art.suno.ai/{id}.jpg"),
7119                        hash: format!("art-{id}"),
7120                        owner_id: id.to_owned(),
7121                        content: None,
7122                    });
7123                    // An Mp3 stem for each clip.
7124                    actions.push(Action::WriteStem {
7125                        clip_id: id.to_owned(),
7126                        key: "voc".to_owned(),
7127                        stem_id: format!("{id}voc"),
7128                        path: format!("{id}.stems/voc.mp3"),
7129                        source_url: format!("https://cdn1.suno.ai/{id}voc.mp3"),
7130                        format: StemFormat::Mp3,
7131                        hash: format!("stem-{id}"),
7132                    });
7133                }
7134                // A failing download in the middle.
7135                let (_f, df, af) = download("fail", AudioFormat::Mp3);
7136                desireds.push(df);
7137                actions.insert(3, af);
7138                // A serial-only delete.
7139                actions.push(Action::Delete {
7140                    path: "old.mp3".to_owned(),
7141                    clip_id: "old".to_owned(),
7142                });
7143                (Plan { actions }, desireds)
7144            }
7145
7146            fn http() -> ScriptedHttp {
7147                ScriptedHttp::new()
7148                    .with_auth()
7149                    .route("x.mp3", Reply::ok(b"x-audio".to_vec()))
7150                    .route("y.mp3", Reply::ok(b"y-audio".to_vec()))
7151                    .route("z.mp3", Reply::ok(b"z-audio".to_vec()))
7152                    .route("fail.mp3", Reply::status(404))
7153                    .route("x.jpg", Reply::ok(b"x-jpg".to_vec()))
7154                    .route("y.jpg", Reply::ok(b"y-jpg".to_vec()))
7155                    .route("z.jpg", Reply::ok(b"z-jpg".to_vec()))
7156                    .route("xvoc.mp3", Reply::ok(b"x-voc".to_vec()))
7157                    .route("yvoc.mp3", Reply::ok(b"y-voc".to_vec()))
7158                    .route("zvoc.mp3", Reply::ok(b"z-voc".to_vec()))
7159            }
7160
7161            fn seed_manifest() -> Manifest {
7162                let mut m = Manifest::new();
7163                m.insert("old".to_owned(), entry("old.mp3", AudioFormat::Mp3));
7164                m
7165            }
7166
7167            let (plan, desireds) = build();
7168
7169            let mut m1 = seed_manifest();
7170            let fs1 = MemFs::new().with_file("old.mp3", b"x".to_vec());
7171            let out1 = run_gated_fs(
7172                &plan,
7173                &mut m1,
7174                &desireds,
7175                &GatedHttp::new(http()),
7176                &fs1,
7177                &opts_with(1),
7178            );
7179
7180            let mut m8 = seed_manifest();
7181            let fs8 = MemFs::new().with_file("old.mp3", b"x".to_vec());
7182            let out8 = run_gated_fs(
7183                &plan,
7184                &mut m8,
7185                &desireds,
7186                &GatedHttp::new(http()),
7187                &fs8,
7188                &opts_with(8),
7189            );
7190
7191            assert_eq!(out1, out8, "outcome must not depend on concurrency");
7192            assert_eq!(m1, m8, "final manifest must not depend on concurrency");
7193            assert_eq!(out8.downloaded, 3);
7194            assert_eq!(out8.deleted, 1);
7195            assert_eq!(out8.failed(), 1);
7196            // Covers and stems for the 3 successful clips.
7197            assert_eq!(out8.artifacts_written, 6);
7198        }
7199
7200        #[test]
7201        fn both_folder_covers_fetch_video_cover_once_under_concurrency() {
7202            // FolderWebp and FolderMp4 share a source_url (the `both` retention).
7203            // Even with other downloads running concurrently, they must stay serial
7204            // so the first fetch inserts into cover_cache and the second drains it
7205            // (#90), fetching the video_cover_url exactly once.
7206            let scripted = ScriptedHttp::new()
7207                .with_auth()
7208                .route("root/video.mp4", Reply::ok(b"mp4-bytes".to_vec()))
7209                .route("d0.mp3", Reply::ok(b"audio".to_vec()))
7210                .route("d1.mp3", Reply::ok(b"audio".to_vec()));
7211            let mut actions = vec![
7212                Action::WriteArtifact {
7213                    kind: ArtifactKind::FolderWebp,
7214                    path: "album/cover.webp".to_owned(),
7215                    source_url: "https://cdn.suno.ai/root/video.mp4".to_owned(),
7216                    hash: "wh".to_owned(),
7217                    owner_id: "root".to_owned(),
7218                    content: None,
7219                },
7220                Action::WriteArtifact {
7221                    kind: ArtifactKind::FolderMp4,
7222                    path: "album/cover.mp4".to_owned(),
7223                    source_url: "https://cdn.suno.ai/root/video.mp4".to_owned(),
7224                    hash: "mh".to_owned(),
7225                    owner_id: "root".to_owned(),
7226                    content: None,
7227                },
7228            ];
7229            let mut desireds = vec![];
7230            for id in ["d0", "d1"] {
7231                let (_c, d, a) = download(id, AudioFormat::Mp3);
7232                actions.push(a);
7233                desireds.push(d);
7234            }
7235            let plan = Plan { actions };
7236            let http = GatedHttp::new(scripted);
7237            let ffmpeg = StubFfmpeg::webp();
7238            let clock = RecordingClock::new();
7239            let mut manifest = Manifest::new();
7240            let mut albums = BTreeMap::new();
7241            let mut playlists = BTreeMap::new();
7242            let client = SunoClient::new(ClerkAuth::new("eyJtoken"), RecordingClock::new());
7243            pollster::block_on(execute(
7244                &plan,
7245                &mut manifest,
7246                &mut albums,
7247                &mut playlists,
7248                &desireds,
7249                &HashMap::new(),
7250                Ports {
7251                    client: &client,
7252                    http: &http,
7253                    fs: &MemFs::new(),
7254                    ffmpeg: &ffmpeg,
7255                    clock: &clock,
7256                },
7257                &opts_with(4),
7258            ));
7259
7260            assert_eq!(
7261                http.count("root/video.mp4"),
7262                1,
7263                "video_cover_url must be fetched exactly once even under concurrency"
7264            );
7265        }
7266
7267        #[test]
7268        fn existing_clip_audio_and_cover_sidecar_share_cover_fetch() {
7269            // Clip "e" is already in the manifest; this run reformats its audio
7270            // AND updates its CoverJpg sidecar. The audio producer caches the
7271            // cover; the sidecar drains it. Even under concurrency the cover must
7272            // be fetched exactly once and cover_cache must not accumulate a
7273            // leaked entry.
7274            let c = art_clip("e");
7275            let cover_url = c.image_large_url.clone();
7276            let d = desired(c.clone(), AudioFormat::Mp3);
7277            let scripted = ScriptedHttp::new()
7278                .with_auth()
7279                .route("e.mp3", Reply::ok(b"audio".to_vec()))
7280                .route("e/large.jpg", Reply::ok(b"cover-jpg".to_vec()));
7281            let plan = Plan {
7282                actions: vec![
7283                    Action::Reformat {
7284                        clip: c,
7285                        path: "e.mp3".to_owned(),
7286                        from_path: "e-old.mp3".to_owned(),
7287                        from: AudioFormat::Mp3,
7288                        to: AudioFormat::Mp3,
7289                    },
7290                    Action::WriteArtifact {
7291                        kind: ArtifactKind::CoverJpg,
7292                        path: "e/cover.jpg".to_owned(),
7293                        source_url: cover_url,
7294                        hash: "new-art".to_owned(),
7295                        owner_id: "e".to_owned(),
7296                        content: None,
7297                    },
7298                ],
7299            };
7300            let mut manifest = Manifest::new();
7301            manifest.insert("e".to_owned(), entry("e-old.mp3", AudioFormat::Mp3));
7302            let fs = MemFs::new().with_file("e-old.mp3", b"old-audio".to_vec());
7303            let http = GatedHttp::new(scripted);
7304            let outcome = run_gated_fs(&plan, &mut manifest, &[d], &http, &fs, &opts_with(4));
7305
7306            assert_eq!(outcome.reformatted, 1);
7307            assert_eq!(outcome.failed(), 0);
7308            assert_eq!(
7309                http.count("e/large.jpg"),
7310                1,
7311                "cover must be fetched exactly once, not once per concurrent action"
7312            );
7313        }
7314    }
7315}