Isolate speech transport failures and keep experimental media usable

crates/searchi-service/src/speech.rs

@@ -24,6 +24,10 @@ pub trait SpeechProvider: Send {
     fn admission_us(&self) -> u64 {
         0
     }
+    /// Starts a new source without retrying a failed window from the previous source.
+    fn begin_source(&mut self) -> Result<()> {
+        Ok(())
+    }
     fn transcribe(&mut self, pcm: &[f32], sample_rate: u32) -> Result<SpeechTranscription>;
 }
 
@@ -31,6 +35,9 @@ impl SpeechProvider for searchi_speech::SpeechWorker {
     fn fingerprint(&self) -> &str {
         searchi_speech::SpeechWorker::fingerprint(self)
     }
+    fn begin_source(&mut self) -> Result<()> {
+        searchi_speech::SpeechWorker::begin_source(self)
+    }
     fn transcribe(&mut self, pcm: &[f32], sample_rate: u32) -> Result<SpeechTranscription> {
         searchi_speech::SpeechWorker::transcribe(self, pcm, sample_rate)
     }
@@ -58,6 +65,13 @@ impl SpeechProvider for BudgetedSpeechProvider {
     fn admission_us(&self) -> u64 {
         self.admission_us
     }
+    fn begin_source(&mut self) -> Result<()> {
+        let start = std::time::Instant::now();
+        let permit = self.budget.acquire(InferenceRole::Background, 1);
+        self.admission_us = start.elapsed().as_micros().min(u64::MAX as u128) as u64;
+        let _permit = permit?;
+        self.inner.begin_source()
+    }
     fn transcribe(&mut self, pcm: &[f32], sample_rate: u32) -> Result<SpeechTranscription> {
         let start = std::time::Instant::now();
         let permit = self.budget.acquire(InferenceRole::Background, 1);
@@ -275,6 +289,10 @@ impl HybridSearch {
         let filename = path.file_name().unwrap_or_default().to_string_lossy();
         let mut ordinal = 0usize;
         if metadata.has_audio {
+            self.speech
+                .as_mut()
+                .ok_or_else(|| anyhow::anyhow!("Speech provider unavailable"))?
+                .begin_source()?;
             for window in speech_windows(metadata.audio_start_ms, metadata.audio_end_ms)? {
                 let pcm = decode_audio_aligned(tools, path, window, limits)?;
                 let provider = self
@@ -372,6 +390,49 @@ impl HybridSearch {
 mod tests {
     use super::*;
     #[test]
+    fn new_source_recovers_provider_without_retrying_failed_window() -> Result<()> {
+        struct SourceProvider {
+            sources: usize,
+            calls: usize,
+            failed: bool,
+        }
+        impl SpeechProvider for SourceProvider {
+            fn fingerprint(&self) -> &str {
+                "lifecycle-test"
+            }
+            fn begin_source(&mut self) -> Result<()> {
+                self.sources += 1;
+                self.failed = false;
+                Ok(())
+            }
+            fn transcribe(&mut self, _: &[f32], _: u32) -> Result<SpeechTranscription> {
+                self.calls += 1;
+                if self.sources == 1 || self.failed {
+                    self.failed = true;
+                    anyhow::bail!("first source transport failure");
+                }
+                Ok(SpeechTranscription {
+                    passages: Vec::new(),
+                    metrics: SpeechMetrics::default(),
+                })
+            }
+        }
+        let mut provider = SourceProvider {
+            sources: 0,
+            calls: 0,
+            failed: false,
+        };
+        provider.begin_source()?;
+        let first_failure = provider.transcribe(&[0.], 16000).unwrap_err().to_string();
+        assert_eq!(provider.calls, 1, "failed PCM is never retried");
+        provider.begin_source()?;
+        provider.transcribe(&[0.], 16000)?;
+        assert_eq!(provider.calls, 2);
+        assert_eq!(provider.sources, 2);
+        assert_eq!(first_failure, "first source transport failure");
+        Ok(())
+    }
+    #[test]
     fn point_spans_are_legal_and_invalid_native_evidence_is_rejected() -> Result<()> {
         let mut result = SpeechTranscription {
             passages: vec![SpeechPassage {

crates/searchi-speech/src/lib.rs

@@ -52,6 +52,7 @@ pub struct SpeechWorker {
     base_path: PathBuf,
     vad_path: PathBuf,
     device: Device,
+    reset_on_next_source: bool,
 }
 
 fn verified_file(path: &Path, bytes: u64, expected: &str) -> Result<String> {
@@ -114,11 +115,31 @@ impl SpeechWorker {
             base_path: base.to_owned(),
             vad_path: vad.to_owned(),
             device,
+            reset_on_next_source: false,
         })
     }
     pub fn fingerprint(&self) -> &str {
         &self.fingerprint
     }
+    /// Recover a failed transport only at admission of a different source.
+    /// Never retries the previous source or malformed native output.
+    pub fn begin_source(&mut self) -> Result<()> {
+        if !std::mem::take(&mut self.reset_on_next_source) {
+            return Ok(());
+        }
+        let replacement = Self::load(
+            &self.worker_path,
+            &self.base_path,
+            &self.vad_path,
+            self.device,
+        )?;
+        ensure!(
+            replacement.fingerprint == self.fingerprint,
+            "Speech assets changed before source recovery"
+        );
+        *self = replacement;
+        Ok(())
+    }
     pub fn transcribe(&mut self, pcm: &[f32], sample_rate: u32) -> Result<SpeechTranscription> {
         ensure!(
             sample_rate == SAMPLE_RATE && !pcm.is_empty() && pcm.len() <= MAX_SAMPLES,
@@ -126,7 +147,7 @@ impl SpeechWorker {
         );
         ensure!(pcm.iter().all(|v| v.is_finite()), "Nonfinite speech input");
         let fallback_started = Instant::now();
-        match self.transcribe_once(pcm) {
+        let result = (|| match self.transcribe_once(pcm) {
             Ok(value) => Ok(value),
             Err(error) if self.device == Device::Auto && process::recoverable(&error) => {
                 // Retry only initialization/compute/transport failures. Reverify
@@ -148,7 +169,9 @@ impl SpeechWorker {
                 Ok(result)
             }
             Err(error) => Err(error),
-        }
+        })();
+        self.reset_on_next_source = result.as_ref().err().is_some_and(process::failed_transport);
+        result
     }
     fn transcribe_once(&mut self, pcm: &[f32]) -> Result<SpeechTranscription> {
         let started = Instant::now();

crates/searchi-speech/src/process.rs

@@ -60,6 +60,9 @@ pub(crate) fn recoverable(error: &anyhow::Error) -> bool {
             .downcast_ref::<NativeFailure>()
             .is_some_and(NativeFailure::recoverable)
 }
+pub(crate) fn failed_transport(error: &anyhow::Error) -> bool {
+    error.is::<TransportFailure>()
+}
 
 fn read_response<R: Read>(reader: &mut R) -> Result<Value> {
     let mut prefix = [0; 4];
@@ -259,6 +262,7 @@ mod tests {
             .exchange(vec![1], Duration::from_millis(50))
             .unwrap_err();
         assert!(recoverable(&error));
+        assert!(failed_transport(&error));
         assert!(process.child.try_wait().unwrap().is_some());
         assert!(process.failed);
         let mut command = Command::new(&python);
@@ -268,6 +272,7 @@ mod tests {
             .exchange(vec![1], Duration::from_secs(1))
             .unwrap_err();
         assert!(!recoverable(&error));
+        assert!(!failed_transport(&error));
         assert!(process.child.try_wait().unwrap().is_some());
         assert!(process.failed);
     }
@@ -279,9 +284,12 @@ mod tests {
             anyhow::Error::new(ProtocolFailure),
         ] {
             assert!(!recoverable(&error));
+            assert!(!failed_transport(&error));
         }
         assert!(recoverable(&NativeFailure::Initialization.into()));
         assert!(recoverable(&NativeFailure::Compute.into()));
+        assert!(!failed_transport(&NativeFailure::Initialization.into()));
+        assert!(!failed_transport(&NativeFailure::Compute.into()));
         assert!(!recoverable(
             &read_response(&mut &[2, 0, 0, 0, 0xff, 0xff][..]).unwrap_err()
         ));