fix: clean up realtime asr finish failures

cai

2026-07-10 47b001f9a864dc829f7d0c911ef3ac7fd15ad2a1

fix: clean up realtime asr finish failures

 src/asr_realtime.rs

@@ -15,8 +15,7 @@

const SAMPLE_RATE_16K: u32 = 16_000;
const CHANNELS_MONO: u32 = 1;
const MIN_CHUNK_DURATION_MS: u64 = 20;
const MAX_CHUNK_DURATION_MS: u64 = 1_000;
const REQUIRED_CHUNK_DURATION_MS: u64 = 200;
const UPLOAD_QUEUE_CAPACITY: usize = 64;
const MAX_NDJSON_LINE_BYTES: usize = 24 * 1024;
const FINISH_RESPONSE_TIMEOUT: Duration = Duration::from_secs(15);
@@ -37,9 +36,7 @@
            && non_blank(&self.url)
            && non_blank(&self.runtime_token)
            && non_blank(&self.runtime_session_nonce)
            && self.chunk_duration_ms >= MIN_CHUNK_DURATION_MS
            && self.chunk_duration_ms <= MAX_CHUNK_DURATION_MS
            && self.chunk_duration_ms % MIN_CHUNK_DURATION_MS == 0
            && self.chunk_duration_ms == REQUIRED_CHUNK_DURATION_MS
    }
}

@@ -115,21 +112,27 @@
        speech_duration_ms: u64,
        end_reason: &str,
    ) -> Result<RealtimeAsrOutcome> {
        let enqueue_result = (|| -> Result<()> {
        if let Some(chunk) = self.chunker.flush() {
            let line = audio_chunk_line(chunk.seq, &chunk.samples)?;
            self.try_send_line(line)?;
        }
        self.try_send_line(vad_speech_end_line(speech_duration_ms, end_reason)?)?;
        self.try_send_line(finish_line()?)?;
            Ok(())
        })();
        if let Err(error) = enqueue_result {
            self.abort_and_wait().await;
            return Err(error);
        }
        self.sender.take();

        let chunk_count = self.chunker.chunk_count;
        let audio_bytes = self.chunker.audio_bytes;
        let mut task = self.task;
        let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut task).await {
        let mut outcome = match timeout(FINISH_RESPONSE_TIMEOUT, &mut self.task).await {
            Ok(joined) => joined.context("realtime asr upload task failed")??,
            Err(_) => {
                task.abort();
                self.abort_and_wait().await;
                return Err(anyhow!("realtime asr finish response timeout"));
            }
        };
@@ -147,8 +150,14 @@
            .await
            .is_err()
        {
            self.task.abort();
            self.abort_and_wait().await;
        }
    }

    async fn abort_and_wait(&mut self) {
        self.sender.take();
        self.task.abort();
        let _ = timeout(CANCEL_RESPONSE_TIMEOUT, &mut self.task).await;
    }

    fn try_send_line(&self, line: Vec<u8>) -> Result<()> {
@@ -416,6 +425,48 @@
        assert!(line.len() <= MAX_NDJSON_LINE_BYTES);
    }

    #[test]
    fn realtime_config_rejects_chunk_duration_other_than_frozen_200ms() {
        let mut config = fixture_config("http://127.0.0.1/realtime".to_string());
        assert!(config.is_ready());

        config.chunk_duration_ms = 500;
        assert!(!config.is_ready());
        config.chunk_duration_ms = 1_000;
        assert!(!config.is_ready());
    }

    #[tokio::test]
    async fn finish_when_tail_chunk_enqueue_fails_then_stops_upload_task() {
        let (mut upload, task_active) = blocked_upload(true);
        upload.chunker.push_48k(&vec![1; 480]);

        let result = upload.finish(10, "silence").await;

        assert!(result.is_err());
        assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
    }

    #[tokio::test]
    async fn finish_when_speech_end_enqueue_fails_then_stops_upload_task() {
        let (upload, task_active) = blocked_upload(true);

        let result = upload.finish(200, "silence").await;

        assert!(result.is_err());
        assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
    }

    #[tokio::test]
    async fn finish_when_finish_enqueue_fails_then_stops_upload_task() {
        let (upload, task_active) = blocked_upload(false);

        let result = upload.finish(200, "silence").await;

        assert!(result.is_err());
        assert!(!task_active.load(std::sync::atomic::Ordering::Acquire));
    }

    #[tokio::test]
    async fn upload_streams_canonical_ndjson_and_reads_final_response() {
        let response = json!({
@@ -500,6 +551,42 @@
        }
    }

    fn blocked_upload(
        prefill_queue: bool,
    ) -> (
        RealtimeAsrUpload,
        std::sync::Arc<std::sync::atomic::AtomicBool>,
    ) {
        let (sender, receiver) = mpsc::channel(1);
        if prefill_queue {
            sender.try_send(vec![b'x']).expect("prefill queue");
        }
        let task_active = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(true));
        let task_state = task_active.clone();
        let active_guard = TestUploadTaskGuard(task_state);
        let task = tokio::spawn(async move {
            let _active_guard = active_guard;
            let _receiver = receiver;
            std::future::pending::<Result<RealtimeAsrOutcome>>().await
        });
        (
            RealtimeAsrUpload {
                sender: Some(sender),
                task,
                chunker: Pcm16kChunker::new(200),
            },
            task_active,
        )
    }

    struct TestUploadTaskGuard(std::sync::Arc<std::sync::atomic::AtomicBool>);

    impl Drop for TestUploadTaskGuard {
        fn drop(&mut self) {
            self.0.store(false, std::sync::atomic::Ordering::Release);
        }
    }

    struct CapturedRequest {
        headers: String,
        body: String,