[gobby-cli-#444] fix: apply remaining CodeRabbit triage

Author: joshwilhelmi

crates/gcore/Cargo.toml

@@ -19,6 +19,7 @@ qdrant = ["dep:reqwest", "dep:urlencoding"]
 indexing = ["dep:ignore", "dep:sha2"]
 search = []
 ai = ["dep:reqwest", "dep:base64", "dep:bytes", "dep:httpdate", "dep:rand", "reqwest/multipart"]
+test-helpers = []
 full = ["postgres", "falkor", "qdrant", "indexing", "search", "ai"]
 
 [dependencies]

crates/gcore/src/lib.rs

@@ -50,5 +50,5 @@ pub mod indexing;
 #[cfg(feature = "search")]
 pub mod search;
 
-#[cfg(test)]
-pub(crate) mod test_http;
+#[cfg(any(test, feature = "test-helpers"))]
+pub mod test_http;

crates/gcore/src/test_http.rs

@@ -5,13 +5,13 @@ use std::net::TcpListener;
 use std::thread;
 use std::time::Duration;
 
-pub(crate) type RequestHandle = thread::JoinHandle<io::Result<String>>;
+pub type RequestHandle = thread::JoinHandle<io::Result<String>>;
 
-pub(crate) fn spawn_json_response(body: impl Into<String>) -> io::Result<(String, RequestHandle)> {
+pub fn spawn_json_response(body: impl Into<String>) -> io::Result<(String, RequestHandle)> {
     spawn_response(200, "OK", "application/json", body.into())
 }
 
-pub(crate) fn spawn_json_response_with_status(
+pub fn spawn_json_response_with_status(
     status: u16,
     body: impl Into<String>,
 ) -> io::Result<(String, RequestHandle)> {

crates/gwiki/Cargo.toml

@@ -57,3 +57,6 @@ pdfium-render = { version = "0.8.37", default-features = false, features = ["pdf
 png = { version = "0.17", optional = true }
 quick-xml = { version = "0.38", optional = true }
 zip = { version = "4", default-features = false, features = ["deflate"], optional = true }
+
+[dev-dependencies]
+gobby-core = { path = "../gcore", version = "0.3.0", features = ["postgres", "falkor", "qdrant", "indexing", "search", "ai", "test-helpers"] }

crates/gwiki/src/ai/chunk.rs

@@ -129,8 +129,9 @@ fn transcribe_chunks(
     _original_request: &TranscriptionRequest<'_>,
     client: &dyn TranscriptionClient,
     mode: ChunkTranscriptionMode<'_>,
-    chunks: Vec<AudioChunk>,
+    mut chunks: Vec<AudioChunk>,
 ) -> Result<ChunkedTranscription, WikiError> {
+    chunks.sort_by_key(|chunk| chunk.start_ms);
     let mut aggregate = empty_output();
     let mut completed_ranges = Vec::new();
     let mut missing_ranges = Vec::new();
@@ -354,7 +355,7 @@ mod tests {
             )),
             Ok(output("en", &[(0, 500, "overlap"), (500, 1_500, "beta")])),
         ]);
-        let chunker = FakeChunker::new(vec![chunk(0, 10_000), chunk(9_000, 19_000)]);
+        let chunker = FakeChunker::new(vec![chunk(9_000, 19_000), chunk(0, 10_000)]);
 
         let stitched = transcribe_audio_request_with_chunker(
             &request,

crates/gwiki/src/collect.rs

@@ -311,19 +311,21 @@ fn accept_item(
     };
 
     if let Err(error) = fs::remove_file(&path) {
-        let _ = fs::remove_file(vault_root.join(&raw_path));
+        let mut cleanup_errors = Vec::new();
+        cleanup_collect_file(vault_root, &raw_path, &mut cleanup_errors);
         if let Some(asset_path) = &asset_path {
-            let _ = fs::remove_file(vault_root.join(asset_path));
+            cleanup_collect_file(vault_root, asset_path, &mut cleanup_errors);
         }
         let original_error = io_error("remove accepted inbox item", &path, error);
         if let Err(rollback_error) = previous_manifest.write(vault_root) {
             return Err(WikiError::Config {
                 detail: format!(
-                    "failed to roll back source manifest after inbox removal failed: {rollback_error}; original error: {original_error}"
+                    "failed to roll back source manifest after inbox removal failed: {rollback_error}; original error: {original_error}{}",
+                    collect_cleanup_detail(&cleanup_errors)
                 ),
             });
         }
-        return Err(original_error);
+        return collect_error_with_cleanup(original_error, cleanup_errors);
     }
     report.accepted.push(CollectAction {
         inbox_path: relative,
@@ -342,20 +344,54 @@ fn rollback_registered_collect_source<T>(
     asset_path: Option<&PathBuf>,
     original_error: WikiError,
 ) -> Result<T, WikiError> {
+    let mut cleanup_errors = Vec::new();
     if let Some(raw_path) = raw_path {
-        let _ = fs::remove_file(vault_root.join(raw_path));
+        cleanup_collect_file(vault_root, raw_path, &mut cleanup_errors);
     }
     if let Some(asset_path) = asset_path {
-        let _ = fs::remove_file(vault_root.join(asset_path));
+        cleanup_collect_file(vault_root, asset_path, &mut cleanup_errors);
     }
     if let Err(rollback_error) = previous_manifest.write(vault_root) {
         return Err(WikiError::Config {
             detail: format!(
-                "failed to roll back source manifest after collect write failure: {rollback_error}; original error: {original_error}"
+                "failed to roll back source manifest after collect write failure: {rollback_error}; original error: {original_error}{}",
+                collect_cleanup_detail(&cleanup_errors)
             ),
         });
     }
-    Err(original_error)
+    collect_error_with_cleanup(original_error, cleanup_errors)
+}
+
+fn cleanup_collect_file(vault_root: &Path, relative_path: &Path, cleanup_errors: &mut Vec<String>) {
+    let path = vault_root.join(relative_path);
+    match fs::remove_file(&path) {
+        Ok(()) => {}
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
+        Err(error) => cleanup_errors.push(format!("{}: {error}", path.display())),
+    }
+}
+
+fn collect_error_with_cleanup<T>(
+    original_error: WikiError,
+    cleanup_errors: Vec<String>,
+) -> Result<T, WikiError> {
+    if cleanup_errors.is_empty() {
+        return Err(original_error);
+    }
+    Err(WikiError::Config {
+        detail: format!(
+            "{original_error}; cleanup failures: {}",
+            cleanup_errors.join("; ")
+        ),
+    })
+}
+
+fn collect_cleanup_detail(cleanup_errors: &[String]) -> String {
+    if cleanup_errors.is_empty() {
+        String::new()
+    } else {
+        format!("; cleanup failures: {}", cleanup_errors.join("; "))
+    }
 }
 
 fn skip_item(

crates/gwiki/src/commands/refresh.rs

@@ -356,11 +356,6 @@ fn refresh_changed_url_source(
     previous_paths.extend(source_asset_paths_for_id(vault_root, &previous.id)?);
 
     let result = ingest::url::ingest_snapshot_without_index(vault_root, snapshot)?;
-    SourceManifest::update(vault_root, |manifest| {
-        let before = manifest.entries.len();
-        manifest.entries.retain(|entry| entry.id != previous.id);
-        Ok(manifest.entries.len() != before)
-    })?;
 
     let mut removed_paths = Vec::new();
     for path in previous_paths {
@@ -377,6 +372,12 @@ fn refresh_changed_url_source(
         }
     }
 
+    SourceManifest::update(vault_root, |manifest| {
+        let before = manifest.entries.len();
+        manifest.entries.retain(|entry| entry.id != previous.id);
+        Ok(manifest.entries.len() != before)
+    })?;
+
     Ok(ChangedRefresh {
         result,
         previous_raw_path,
@@ -402,13 +403,6 @@ fn refresh_changed_local_source(
         vault_root, scope, ai_context, options, path, fetched_at,
     )?;
     let result = local_result.result;
-    if previous.id != result.record.id {
-        SourceManifest::update(vault_root, |manifest| {
-            let before = manifest.entries.len();
-            manifest.entries.retain(|entry| entry.id != previous.id);
-            Ok(manifest.entries.len() != before)
-        })?;
-    }
 
     let mut removed_paths = Vec::new();
     for path in previous_paths {
@@ -425,6 +419,14 @@ fn refresh_changed_local_source(
         }
     }
 
+    if previous.id != result.record.id {
+        SourceManifest::update(vault_root, |manifest| {
+            let before = manifest.entries.len();
+            manifest.entries.retain(|entry| entry.id != previous.id);
+            Ok(manifest.entries.len() != before)
+        })?;
+    }
+
     Ok(ChangedRefresh {
         result,
         previous_raw_path,

crates/gwiki/src/commands/sources.rs

@@ -56,15 +56,15 @@ pub(crate) fn execute_remove(
         scope.root(),
         &raw_path,
         source.raw_exists,
-        dry_run,
+        true,
         &mut path_changes,
     )?;
 
     if let Some(asset_path) = asset_path {
         stage_source_asset(
             scope.root(),
             &asset_path,
-            dry_run,
+            true,
             keep_asset,
             &mut path_changes,
             &mut degradations,
@@ -75,9 +75,20 @@ pub(crate) fn execute_remove(
         IndexStatus::not_run()
     } else {
         match SourceManifest::remove(scope.root(), &record.id) {
-            Ok(Some(_)) => {}
-            Ok(None) => degradations.push("manifest_entry_missing_after_plan".to_string()),
-            Err(error) => degradations.push(format!("manifest_remove_failed:{error}")),
+            Ok(Some(removed)) => {
+                if let Err(error) = remove_staged_source_files(scope.root(), &mut path_changes) {
+                    rollback_removed_source(scope.root(), removed, &error)?;
+                    return Err(error);
+                }
+            }
+            Ok(None) => {
+                degradations.push("manifest_entry_missing_after_plan".to_string());
+                path_changes.removed_paths.clear();
+            }
+            Err(error) => {
+                degradations.push(format!("manifest_remove_failed:{error}"));
+                path_changes.removed_paths.clear();
+            }
         }
         match index::index_resolved_scope(&scope) {
             Ok(counts) => IndexStatus::indexed(counts),
@@ -326,6 +337,52 @@ fn stage_source_asset(
     Ok(())
 }
 
+fn remove_staged_source_files(
+    vault_root: &Path,
+    path_changes: &mut PathChanges,
+) -> Result<(), WikiError> {
+    let planned = std::mem::take(&mut path_changes.removed_paths);
+    let mut removed = Vec::with_capacity(planned.len());
+    for relative in planned {
+        let full_path = vault_root.join(&relative);
+        match fs::remove_file(&full_path) {
+            Ok(()) => removed.push(relative),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+                path_changes.missing_paths.push(relative);
+            }
+            Err(source) => {
+                path_changes.removed_paths = removed;
+                return Err(WikiError::Io {
+                    action: "remove source file",
+                    path: Some(full_path),
+                    source,
+                });
+            }
+        }
+    }
+    path_changes.removed_paths = removed;
+    Ok(())
+}
+
+fn rollback_removed_source(
+    vault_root: &Path,
+    record: SourceRecord,
+    original_error: &WikiError,
+) -> Result<(), WikiError> {
+    SourceManifest::update(vault_root, |manifest| {
+        if manifest.entries.iter().any(|entry| entry.id == record.id) {
+            return Ok(false);
+        }
+        manifest.entries.push(record);
+        Ok(true)
+    })
+    .map_err(|rollback_error| WikiError::Config {
+        detail: format!(
+            "failed to roll back source manifest after source file removal failed: {rollback_error}; original error: {original_error}"
+        ),
+    })
+}
+
 fn raw_source_path(id: &str) -> Result<PathBuf, WikiError> {
     if id.trim().is_empty()
         || id.contains('/')

crates/gwiki/src/health.rs

@@ -329,6 +329,7 @@ fn source_reference_patterns(needle: &str) -> Vec<String> {
     ]
 }
 
+#[cfg(test)]
 fn source_reference_is_present(markdown: &str, needle: &str) -> bool {
     let needle = needle.trim();
     if needle.is_empty() {

crates/gwiki/src/ingest/audio.rs

@@ -523,8 +523,8 @@ mod tests {
     #[cfg(feature = "ai")]
     fn spawn_transcription_server(
         response: &'static str,
-    ) -> (String, std::thread::JoinHandle<String>) {
-        crate::test_http::spawn_json_response(response)
+    ) -> (String, gobby_core::test_http::RequestHandle) {
+        gobby_core::test_http::spawn_json_response(response).expect("spawn test server")
     }
 
     #[cfg(feature = "ai")]
@@ -595,7 +595,7 @@ mod tests {
         )
         .expect("ingest audio with production transcript");
 
-        let request = request.join().expect("request");
+        let request = request.join().expect("request").expect("request ok");
         assert!(request.starts_with("POST /v1/audio/transcriptions HTTP/1.1"));
         assert!(result.transcription_degradation.is_none());
 
@@ -628,7 +628,7 @@ mod tests {
         )
         .expect("ingest translated audio");
 
-        let request = request.join().expect("request");
+        let request = request.join().expect("request").expect("request ok");
         assert!(request.starts_with("POST /v1/audio/translations HTTP/1.1"));
         assert!(result.transcription_degradation.is_none());