Added merge

furumi-agent/src/ingest/mod.rs

@@ -25,6 +25,18 @@ pub async fn run(state: Arc<AppState>) {
             Ok(count) => tracing::info!(count, "re-processed pending tracks"),
             Err(e) => tracing::error!(?e, "pending re-processing failed"),
         }
+        // Process pending merge proposals
+        match db::get_pending_merges_for_processing(&state.pool).await {
+            Ok(merge_ids) => {
+                for merge_id in merge_ids {
+                    if let Err(e) = crate::merge::propose_merge(&state, merge_id).await {
+                        tracing::error!(id = %merge_id, ?e, "Merge proposal failed");
+                        let _ = db::update_merge_status(&state.pool, merge_id, "error", Some(&e.to_string())).await;
+                    }
+                }
+            }
+            Err(e) => tracing::error!(?e, "Failed to load pending merges"),
+        }
         tokio::time::sleep(interval).await;
     }
 }
@@ -161,13 +173,14 @@ async fn reprocess_pending(state: &Arc<AppState>) -> anyhow::Result<usize> {
                         .join(sanitize_filename(album))
                         .join(&dest_filename);
 
-                    let storage_path = if dest.exists() && !source.exists() {
-                        dest.to_string_lossy().to_string()
+                    let (storage_path, was_merged) = if dest.exists() && !source.exists() {
+                        (dest.to_string_lossy().to_string(), false)
                     } else if source.exists() {
                         match mover::move_to_storage(
                             &state.config.storage_dir, artist, album, &dest_filename, source,
                         ).await {
-                            Ok(p) => p.to_string_lossy().to_string(),
+                            Ok(mover::MoveOutcome::Moved(p)) => (p.to_string_lossy().to_string(), false),
+                            Ok(mover::MoveOutcome::Merged(p)) => (p.to_string_lossy().to_string(), true),
                             Err(e) => {
                                 tracing::error!(id = %pt.id, ?e, "Failed to move file");
                                 db::update_pending_status(&state.pool, pt.id, "error", Some(&e.to_string())).await?;
@@ -181,7 +194,12 @@ async fn reprocess_pending(state: &Arc<AppState>) -> anyhow::Result<usize> {
                     };
 
                     match db::approve_and_finalize(&state.pool, pt.id, &storage_path).await {
-                        Ok(track_id) => tracing::info!(id = %pt.id, track_id, "Track finalized"),
+                        Ok(track_id) => {
+                            if was_merged {
+                                let _ = db::update_pending_status(&state.pool, pt.id, "merged", None).await;
+                            }
+                            tracing::info!(id = %pt.id, track_id, "Track finalized");
+                        }
                         Err(e) => tracing::error!(id = %pt.id, ?e, "Failed to finalize"),
                     }
                 }
@@ -472,10 +490,17 @@ async fn process_file(state: &Arc<AppState>, file_path: &std::path::Path) -> any
                 )
                 .await
                 {
-                    Ok(storage_path) => {
+                    Ok(outcome) => {
+                        let (storage_path, was_merged) = match outcome {
+                            mover::MoveOutcome::Moved(p) => (p, false),
+                            mover::MoveOutcome::Merged(p) => (p, true),
+                        };
                         let rel_path = storage_path.to_string_lossy().to_string();
                         match db::approve_and_finalize(&state.pool, pending_id, &rel_path).await {
                             Ok(track_id) => {
+                                if was_merged {
+                                    let _ = db::update_pending_status(&state.pool, pending_id, "merged", None).await;
+                                }
                                 tracing::info!(file = filename, track_id, storage = %rel_path, "Track finalized in database");
                             }
                             Err(e) => {

furumi-agent/src/ingest/mover.rs

@@ -1,18 +1,27 @@
 use std::path::{Path, PathBuf};
 
+pub enum MoveOutcome {
+    /// File was moved/renamed to destination.
+    Moved(PathBuf),
+    /// Destination already existed; inbox duplicate was removed.
+    Merged(PathBuf),
+}
+
 /// Move a file from inbox to the permanent storage directory.
 ///
 /// Creates the directory structure: `storage_dir/artist/album/filename`
 /// Returns the full path of the moved file.
 ///
 /// If `rename` fails (cross-device), falls back to copy + remove.
+/// If the destination already exists the inbox copy is removed and
+/// `MoveOutcome::Merged` is returned instead of an error.
 pub async fn move_to_storage(
     storage_dir: &Path,
     artist: &str,
     album: &str,
     filename: &str,
     source: &Path,
-) -> anyhow::Result<PathBuf> {
+) -> anyhow::Result<MoveOutcome> {
     let artist_dir = sanitize_dir_name(artist);
     let album_dir = sanitize_dir_name(album);
 
@@ -21,9 +30,13 @@ pub async fn move_to_storage(
 
     let dest = dest_dir.join(filename);
 
-    // Avoid overwriting existing files
+    // File already at destination — remove the inbox duplicate
     if dest.exists() {
-        anyhow::bail!("Destination already exists: {:?}", dest);
+        if source.exists() {
+            tokio::fs::remove_file(source).await?;
+            tracing::info!(from = ?source, to = ?dest, "merged duplicate into existing storage file");
+        }
+        return Ok(MoveOutcome::Merged(dest));
     }
 
     // Try atomic rename first (same filesystem)
@@ -37,7 +50,7 @@ pub async fn move_to_storage(
     }
 
     tracing::info!(from = ?source, to = ?dest, "moved file to storage");
-    Ok(dest)
+    Ok(MoveOutcome::Moved(dest))
 }
 
 /// Remove characters that are unsafe for directory names.

furumi-agent/src/ingest/normalize.rs

@@ -121,7 +121,7 @@ struct OllamaResponseMessage {
     content: String,
 }
 
-async fn call_ollama(
+pub async fn call_ollama(
     base_url: &str,
     model: &str,
     system_prompt: &str,