diff --git a/crates/signal-bot-transcription/src/fanout.rs b/crates/signal-bot-transcription/src/fanout.rs index eb8ca75..d4e7faa 100644 --- a/crates/signal-bot-transcription/src/fanout.rs +++ b/crates/signal-bot-transcription/src/fanout.rs @@ -12,7 +12,8 @@ pub trait TranscriptFanout: Send + Sync { async fn fan_out_transcript(&self, original: &BotMessage, spoken_text: &str); } -/// Fire-and-forget so the transcript quote-reply is not delayed by NEAR chat. +/// Fire-and-forget so NEAR chat does not block the handler after the transcript +/// quote-reply is sent. Callers must spawn only after that send succeeds. pub fn spawn_fanout(fanout: Option, original: &BotMessage, spoken: &str) { let Some(fanout) = fanout else { return; @@ -30,6 +31,59 @@ pub fn spawn_fanout(fanout: Option, original: &BotMessag /// Shared handle for voice / `!transcribe` handlers. pub type SharedTranscriptFanout = std::sync::Arc; +#[cfg(test)] +pub(crate) struct RecordingFanout { + pub events: std::sync::Mutex>, + pub spoken: std::sync::Mutex>, + pub sources: std::sync::Mutex>, +} + +#[cfg(test)] +impl RecordingFanout { + pub(crate) fn new() -> std::sync::Arc { + std::sync::Arc::new(Self { + events: std::sync::Mutex::new(Vec::new()), + spoken: std::sync::Mutex::new(Vec::new()), + sources: std::sync::Mutex::new(Vec::new()), + }) + } +} + +#[cfg(test)] +#[async_trait] +impl TranscriptFanout for RecordingFanout { + async fn fan_out_transcript(&self, original: &BotMessage, spoken_text: &str) { + self.events.lock().unwrap().push("fanout"); + self.spoken.lock().unwrap().push(spoken_text.to_string()); + self.sources.lock().unwrap().push(original.source.clone()); + } +} + +#[cfg(test)] +pub(crate) struct RecordSend(pub std::sync::Arc); + +#[cfg(test)] +impl wiremock::Respond for RecordSend { + fn respond(&self, _request: &wiremock::Request) -> wiremock::ResponseTemplate { + self.0.events.lock().unwrap().push("send"); + wiremock::ResponseTemplate::new(200).set_body_json(serde_json::json!({})) + } +} + +#[cfg(test)] +pub(crate) async fn wait_for_fanout(rec: &RecordingFanout) { + tokio::time::timeout(std::time::Duration::from_secs(2), async { + loop { + if rec.events.lock().unwrap().contains(&"fanout") { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("fan-out should run"); +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/signal-bot-transcription/src/manual_transcribe.rs b/crates/signal-bot-transcription/src/manual_transcribe.rs index 3c0c69a..5e0b1c4 100644 --- a/crates/signal-bot-transcription/src/manual_transcribe.rs +++ b/crates/signal-bot-transcription/src/manual_transcribe.rs @@ -217,27 +217,31 @@ impl CommandHandler for ManualTranscribeHandler { return Ok(String::new()); } - let body = match self.transcribe_audio(&audio, &bytes).await { + match self.transcribe_audio(&audio, &bytes).await { Ok((spoken, transcript)) => { info!( source = %message.source, chars = transcript.len(), "!transcribe completed" ); + self.send_reply(message, Some(quote), &transcript).await?; spawn_fanout( self.fanout.clone(), &speaker_msg_for_fanout(message, quote), &spoken, ); - transcript } Err(e) => { warn!("Whisper transcription failed: {}", e); - Self::user_message_for_whisper_error(&e).to_string() + self.send_reply( + message, + Some(quote), + Self::user_message_for_whisper_error(&e), + ) + .await?; } - }; + } - self.send_reply(message, Some(quote), &body).await?; Ok(String::new()) } } @@ -245,6 +249,7 @@ impl CommandHandler for ManualTranscribeHandler { #[cfg(test)] mod tests { use super::*; + use crate::fanout::{wait_for_fanout, RecordSend, RecordingFanout, SharedTranscriptFanout}; use signal_client::{BotMessage, QuotedMessage}; fn sample_audio() -> signal_client::Attachment { @@ -322,6 +327,28 @@ mod tests { assert!(!handler.matches(&msg)); } + fn quoted_transcribe_msg() -> BotMessage { + BotMessage { + source: "+15550002222".into(), + source_number: Some("+15550002222".into()), + source_name: None, + text: "!transcribe".into(), + timestamp: 2, + message_timestamp: 2, + is_group: false, + group_id: None, + group_name: None, + receiving_account: "+15550001111".into(), + attachments: vec![], + quote: Some(QuotedMessage { + id: 100, + author_number: Some("+15550003333".into()), + text: None, + audio_attachment: Some(sample_audio()), + }), + } + } + #[tokio::test] async fn execute_without_quote_sends_usage_hint() { use serde_json::json; @@ -336,6 +363,8 @@ mod tests { .mount(&signal_mock) .await; + let rec = RecordingFanout::new(); + let fanout: SharedTranscriptFanout = rec.clone(); let handler = ManualTranscribeHandler::new( test_whisper("http://127.0.0.1:9"), Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), @@ -343,7 +372,8 @@ mod tests { 5_000_000, VoiceAttachmentCache::new(10), empty_store(), - ); + ) + .with_fanout(Some(fanout)); let msg = BotMessage { source: "+15550002222".into(), @@ -361,6 +391,7 @@ mod tests { }; let out = handler.execute(&msg).await.unwrap(); assert!(out.is_empty()); + assert!(rec.spoken.lock().unwrap().is_empty()); } #[tokio::test] @@ -401,26 +432,7 @@ mod tests { empty_store(), ); - let msg = BotMessage { - source: "+15550002222".into(), - source_number: Some("+15550002222".into()), - source_name: None, - text: "!transcribe".into(), - timestamp: 2, - message_timestamp: 2, - is_group: false, - group_id: None, - group_name: None, - receiving_account: "+15550001111".into(), - attachments: vec![], - quote: Some(QuotedMessage { - id: 100, - author_number: Some("+15550003333".into()), - text: None, - audio_attachment: Some(sample_audio()), - }), - }; - let out = handler.execute(&msg).await.unwrap(); + let out = handler.execute("ed_transcribe_msg()).await.unwrap(); assert!(out.is_empty()); } @@ -441,6 +453,8 @@ mod tests { let store = empty_store(); store.set_enabled("+15550002222", true, false); + let rec = RecordingFanout::new(); + let fanout: SharedTranscriptFanout = rec.clone(); let handler = ManualTranscribeHandler::new( test_whisper("http://127.0.0.1:9"), Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), @@ -448,28 +462,110 @@ mod tests { 5_000_000, VoiceAttachmentCache::new(10), store, - ); + ) + .with_fanout(Some(fanout)); - let msg = BotMessage { - source: "+15550002222".into(), - source_number: Some("+15550002222".into()), - source_name: None, - text: "!transcribe".into(), - timestamp: 2, - message_timestamp: 2, - is_group: false, - group_id: None, - group_name: None, - receiving_account: "+15550001111".into(), - attachments: vec![], - quote: Some(QuotedMessage { - id: 100, - author_number: Some("+15550003333".into()), - text: None, - audio_attachment: Some(sample_audio()), - }), - }; - let out = handler.execute(&msg).await.unwrap(); + let out = handler.execute("ed_transcribe_msg()).await.unwrap(); assert!(out.is_empty()); + assert!(rec.spoken.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn execute_sends_transcript_before_fanout() { + use serde_json::json; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let signal_mock = MockServer::start().await; + let whisper_mock = MockServer::start().await; + let rec = RecordingFanout::new(); + + Mock::given(method("GET")) + .and(path("/v1/attachments/cached-voice-id")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"audio")) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/v2/send")) + .respond_with(RecordSend(rec.clone())) + .expect(1) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/audio/transcriptions")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "text": "quoted transcript", + "language": "english" + }))) + .mount(&whisper_mock) + .await; + + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = ManualTranscribeHandler::new( + test_whisper(&whisper_mock.uri()), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + "📝 Transcript:", + 5_000_000, + VoiceAttachmentCache::new(10), + empty_store(), + ) + .with_fanout(Some(fanout)); + + handler.execute("ed_transcribe_msg()).await.unwrap(); + wait_for_fanout(&rec).await; + assert_eq!(*rec.events.lock().unwrap(), vec!["send", "fanout"]); + assert_eq!( + *rec.spoken.lock().unwrap(), + vec!["quoted transcript".to_string()] + ); + assert_eq!( + *rec.sources.lock().unwrap(), + vec!["+15550003333".to_string()] + ); + } + + #[tokio::test] + async fn execute_skips_fanout_when_send_fails() { + use serde_json::json; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + let signal_mock = MockServer::start().await; + let whisper_mock = MockServer::start().await; + let rec = RecordingFanout::new(); + + Mock::given(method("GET")) + .and(path("/v1/attachments/cached-voice-id")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"audio")) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/v2/send")) + .respond_with(ResponseTemplate::new(500).set_body_string("fail")) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/audio/transcriptions")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "text": "quoted transcript", + "language": "english" + }))) + .mount(&whisper_mock) + .await; + + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = ManualTranscribeHandler::new( + test_whisper(&whisper_mock.uri()), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + "📝 Transcript:", + 5_000_000, + VoiceAttachmentCache::new(10), + empty_store(), + ) + .with_fanout(Some(fanout)); + + assert!(handler.execute("ed_transcribe_msg()).await.is_err()); + assert!(rec.spoken.lock().unwrap().is_empty()); + assert!(rec.events.lock().unwrap().is_empty()); } } diff --git a/crates/signal-bot-transcription/src/voice.rs b/crates/signal-bot-transcription/src/voice.rs index ccd999c..4b600b3 100644 --- a/crates/signal-bot-transcription/src/voice.rs +++ b/crates/signal-bot-transcription/src/voice.rs @@ -60,6 +60,11 @@ impl VoiceHandler { crate::fanout::spawn_fanout(self.fanout.clone(), original, spoken); } + async fn send_quote_reply(&self, message: &BotMessage, body: &str) -> AppResult<()> { + self.signal.reply_quoted(message, body, None).await?; + Ok(()) + } + pub fn format_transcript(text: &str, prefix: &str) -> String { format!("{prefix}\n{text}") } @@ -102,6 +107,10 @@ impl CommandHandler for VoiceHandler { true } + fn handles_own_reply(&self) -> bool { + true + } + fn label(&self) -> &'static str { "voice" } @@ -110,7 +119,11 @@ impl CommandHandler for VoiceHandler { async fn execute(&self, message: &BotMessage) -> AppResult { let audio = match message.primary_audio_attachment() { Some(a) => a, - None => return Ok("Could not read voice attachment.".into()), + None => { + self.send_quote_reply(message, "Could not read voice attachment.") + .await?; + return Ok(String::new()); + } }; if let Some(cache) = &self.voice_cache { @@ -124,7 +137,12 @@ impl CommandHandler for VoiceHandler { max = self.max_attachment_bytes, "Voice attachment exceeds size limit" ); - return Ok("Voice note too long (max 5 min). Send a shorter clip.".into()); + self.send_quote_reply( + message, + "Voice note too long (max 5 min). Send a shorter clip.", + ) + .await?; + return Ok(String::new()); } } @@ -132,7 +150,9 @@ impl CommandHandler for VoiceHandler { Ok(bytes) => bytes, Err(e) => { warn!("Failed to download voice attachment {}: {}", audio.id, e); - return Ok("Could not download voice note. Try again later.".into()); + self.send_quote_reply(message, "Could not download voice note. Try again later.") + .await?; + return Ok(String::new()); } }; @@ -142,7 +162,12 @@ impl CommandHandler for VoiceHandler { max = self.max_attachment_bytes, "Downloaded voice attachment exceeds size limit" ); - return Ok("Voice note too long (max 5 min). Send a shorter clip.".into()); + self.send_quote_reply( + message, + "Voice note too long (max 5 min). Send a shorter clip.", + ) + .await?; + return Ok(String::new()); } let filename = Self::attachment_filename(audio); @@ -159,12 +184,16 @@ impl CommandHandler for VoiceHandler { "Voice note transcribed" ); let spoken = transcription.trimmed_text().to_string(); + let body = Self::format_transcript(&spoken, &self.reply_prefix); + self.send_quote_reply(message, &body).await?; self.spawn_fanout(message, &spoken); - Ok(Self::format_transcript(&spoken, &self.reply_prefix)) + Ok(String::new()) } Err(e) => { warn!("Whisper transcription failed: {}", e); - Ok(Self::user_message_for_whisper_error(&e).into()) + self.send_quote_reply(message, Self::user_message_for_whisper_error(&e)) + .await?; + Ok(String::new()) } } } @@ -173,7 +202,11 @@ impl CommandHandler for VoiceHandler { #[cfg(test)] mod tests { use super::*; + use crate::fanout::{wait_for_fanout, RecordSend, RecordingFanout, SharedTranscriptFanout}; use crate::transcribe_store::TranscribeStore; + use serde_json::json; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; #[test] fn format_transcript_includes_prefix() { @@ -216,47 +249,67 @@ mod tests { } } - #[tokio::test] - async fn execute_without_audio_attachment() { - let whisper = Arc::new( + fn test_whisper(url: &str) -> Arc { + Arc::new( WhisperClient::new( - "http://127.0.0.1:9", - std::time::Duration::from_secs(2), + url, + std::time::Duration::from_secs(5), "test-key", "openai/whisper-large-v3", ) .unwrap(), - ); - let signal = Arc::new(SignalClient::new("http://127.0.0.1:9").unwrap()); - let handler = VoiceHandler::new(whisper, signal, DEFAULT_REPLY_PREFIX, 1024); + ) + } + + async fn mount_send_ok(signal_mock: &MockServer) { + Mock::given(method("POST")) + .and(path("/v2/send")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .expect(1) + .mount(signal_mock) + .await; + } + + #[tokio::test] + async fn execute_without_audio_attachment() { + let signal_mock = MockServer::start().await; + mount_send_ok(&signal_mock).await; + let rec = RecordingFanout::new(); + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = VoiceHandler::new( + test_whisper("http://127.0.0.1:9"), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 1024, + ) + .with_fanout(Some(fanout)); let mut msg = dm_voice(None); msg.attachments.clear(); let out = handler.execute(&msg).await.unwrap(); - assert!(out.contains("Could not read voice attachment")); + assert!(out.is_empty()); + assert!(rec.spoken.lock().unwrap().is_empty()); } #[tokio::test] async fn execute_rejects_oversized_declared_size() { - let whisper = Arc::new( - WhisperClient::new( - "http://127.0.0.1:9", - std::time::Duration::from_secs(2), - "test-key", - "openai/whisper-large-v3", - ) - .unwrap(), - ); - let signal = Arc::new(SignalClient::new("http://127.0.0.1:9").unwrap()); - let handler = VoiceHandler::new(whisper, signal, DEFAULT_REPLY_PREFIX, 100); + let signal_mock = MockServer::start().await; + mount_send_ok(&signal_mock).await; + let rec = RecordingFanout::new(); + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = VoiceHandler::new( + test_whisper("http://127.0.0.1:9"), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 100, + ) + .with_fanout(Some(fanout)); let out = handler.execute(&dm_voice(Some(500))).await.unwrap(); - assert!(out.contains("too long")); + assert!(out.is_empty()); + assert!(rec.spoken.lock().unwrap().is_empty()); } #[tokio::test] async fn execute_transcribes_via_whisper() { - use wiremock::matchers::{method, path}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - let signal_mock = MockServer::start().await; let whisper_mock = MockServer::start().await; @@ -265,63 +318,171 @@ mod tests { .respond_with(ResponseTemplate::new(200).set_body_bytes(b"fake-audio-bytes")) .mount(&signal_mock) .await; + mount_send_ok(&signal_mock).await; Mock::given(method("POST")) .and(path("/audio/transcriptions")) - .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ "text": " Hola mundo\n", "language": "spanish" }))) .mount(&whisper_mock) .await; - let whisper = Arc::new( - WhisperClient::new( - whisper_mock.uri(), - std::time::Duration::from_secs(5), - "test-key", - "openai/whisper-large-v3", - ) - .unwrap(), - ); - let signal = Arc::new(SignalClient::new(signal_mock.uri()).unwrap()); let cache = VoiceAttachmentCache::with_default_capacity(); let store = Arc::new(TranscribeStore::new(None)); let msg = dm_voice(Some(16)); store.set_enabled(msg.reply_target(), true, false); - let handler = VoiceHandler::new(whisper, signal, DEFAULT_REPLY_PREFIX, 10_000) - .with_voice_cache(cache.clone()) - .with_transcribe_store(store); + let handler = VoiceHandler::new( + test_whisper(&whisper_mock.uri()), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 10_000, + ) + .with_voice_cache(cache.clone()) + .with_transcribe_store(store); assert!(handler.matches(&msg)); + assert!(handler.handles_own_reply()); let out = handler.execute(&msg).await.unwrap(); - assert_eq!(out, "📝 Transcript:\nHola mundo"); + assert!(out.is_empty()); assert!(cache.lookup(msg.reply_target(), msg.timestamp).is_some()); } #[tokio::test] async fn execute_handles_download_failure() { - use wiremock::matchers::{method, path}; - use wiremock::{Mock, MockServer, ResponseTemplate}; - let signal_mock = MockServer::start().await; Mock::given(method("GET")) .and(path("/v1/attachments/att-1")) .respond_with(ResponseTemplate::new(500).set_body_string("fail")) .mount(&signal_mock) .await; + mount_send_ok(&signal_mock).await; + + let rec = RecordingFanout::new(); + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = VoiceHandler::new( + test_whisper("http://127.0.0.1:9"), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 10_000, + ) + .with_fanout(Some(fanout)); + let out = handler.execute(&dm_voice(Some(10))).await.unwrap(); + assert!(out.is_empty()); + assert!(rec.spoken.lock().unwrap().is_empty()); + } - let whisper = Arc::new( - WhisperClient::new( - "http://127.0.0.1:9", - std::time::Duration::from_secs(2), - "test-key", - "openai/whisper-large-v3", - ) - .unwrap(), + #[tokio::test] + async fn execute_sends_transcript_before_fanout() { + let signal_mock = MockServer::start().await; + let whisper_mock = MockServer::start().await; + let rec = RecordingFanout::new(); + + Mock::given(method("GET")) + .and(path("/v1/attachments/att-1")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"fake-audio-bytes")) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/v2/send")) + .respond_with(RecordSend(rec.clone())) + .expect(1) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/audio/transcriptions")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "text": "hola", + "language": "spanish" + }))) + .mount(&whisper_mock) + .await; + + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = VoiceHandler::new( + test_whisper(&whisper_mock.uri()), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 10_000, + ) + .with_fanout(Some(fanout)); + let msg = dm_voice(Some(16)); + handler.execute(&msg).await.unwrap(); + wait_for_fanout(&rec).await; + assert_eq!(*rec.events.lock().unwrap(), vec!["send", "fanout"]); + assert_eq!(*rec.spoken.lock().unwrap(), vec!["hola".to_string()]); + assert_eq!( + *rec.sources.lock().unwrap(), + vec!["+15550002222".to_string()] ); - let signal = Arc::new(SignalClient::new(signal_mock.uri()).unwrap()); - let handler = VoiceHandler::new(whisper, signal, DEFAULT_REPLY_PREFIX, 10_000); - let out = handler.execute(&dm_voice(Some(10))).await.unwrap(); - assert!(out.contains("Could not download")); + } + + #[tokio::test] + async fn execute_skips_fanout_when_send_fails() { + let signal_mock = MockServer::start().await; + let whisper_mock = MockServer::start().await; + let rec = RecordingFanout::new(); + + Mock::given(method("GET")) + .and(path("/v1/attachments/att-1")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"fake-audio-bytes")) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/v2/send")) + .respond_with(ResponseTemplate::new(500).set_body_string("fail")) + .mount(&signal_mock) + .await; + Mock::given(method("POST")) + .and(path("/audio/transcriptions")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "text": "hola", + "language": "spanish" + }))) + .mount(&whisper_mock) + .await; + + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = VoiceHandler::new( + test_whisper(&whisper_mock.uri()), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 10_000, + ) + .with_fanout(Some(fanout)); + assert!(handler.execute(&dm_voice(Some(16))).await.is_err()); + assert!(rec.spoken.lock().unwrap().is_empty()); + assert!(rec.events.lock().unwrap().is_empty()); + } + + #[tokio::test] + async fn execute_skips_fanout_on_whisper_error() { + let signal_mock = MockServer::start().await; + let whisper_mock = MockServer::start().await; + let rec = RecordingFanout::new(); + + Mock::given(method("GET")) + .and(path("/v1/attachments/att-1")) + .respond_with(ResponseTemplate::new(200).set_body_bytes(b"fake-audio-bytes")) + .mount(&signal_mock) + .await; + mount_send_ok(&signal_mock).await; + Mock::given(method("POST")) + .and(path("/audio/transcriptions")) + .respond_with(ResponseTemplate::new(500).set_body_string("fail")) + .mount(&whisper_mock) + .await; + + let fanout: SharedTranscriptFanout = rec.clone(); + let handler = VoiceHandler::new( + test_whisper(&whisper_mock.uri()), + Arc::new(SignalClient::new(signal_mock.uri()).unwrap()), + DEFAULT_REPLY_PREFIX, + 10_000, + ) + .with_fanout(Some(fanout)); + let out = handler.execute(&dm_voice(Some(16))).await.unwrap(); + assert!(out.is_empty()); + assert!(rec.spoken.lock().unwrap().is_empty()); } }