Skip to content

fix: post transcript quote-reply before in-process fan-out #33

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We鈥檒l occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
daopunk merged 1 commit into from
Aug 14, 2026
Merged

fix: post transcript quote-reply before in-process fan-out #33

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
56 changes: 55 additions & 1 deletion crates/signal-bot-transcription/src/fanout.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<SharedTranscriptFanout>, original: &BotMessage, spoken: &str) {
let Some(fanout) = fanout else {
return;
Expand All @@ -30,6 +31,59 @@ pub fn spawn_fanout(fanout: Option<SharedTranscriptFanout>, original: &BotMessag
/// Shared handle for voice / `!transcribe` handlers.
pub type SharedTranscriptFanout = std::sync::Arc<dyn TranscriptFanout>;

#[cfg(test)]
pub(crate) struct RecordingFanout {
pub events: std::sync::Mutex<Vec<&'static str>>,
pub spoken: std::sync::Mutex<Vec<String>>,
pub sources: std::sync::Mutex<Vec<String>>,
}

#[cfg(test)]
impl RecordingFanout {
pub(crate) fn new() -> std::sync::Arc<Self> {
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<RecordingFanout>);

#[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::*;
Expand Down
190 changes: 143 additions & 47 deletions crates/signal-bot-transcription/src/manual_transcribe.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -217,34 +217,39 @@ 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())
}
}

#[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 {
Expand Down Expand Up @@ -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;
Expand All @@ -336,14 +363,17 @@ 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()),
"馃摑 Transcript:",
5_000_000,
VoiceAttachmentCache::new(10),
empty_store(),
);
)
.with_fanout(Some(fanout));

let msg = BotMessage {
source: "+15550002222".into(),
Expand All @@ -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]
Expand Down Expand Up @@ -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(&quoted_transcribe_msg()).await.unwrap();
assert!(out.is_empty());
}

Expand All @@ -441,35 +453,119 @@ 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()),
"馃摑 Transcript:",
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(&quoted_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(&quoted_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(&quoted_transcribe_msg()).await.is_err());
assert!(rec.spoken.lock().unwrap().is_empty());
assert!(rec.events.lock().unwrap().is_empty());
}
}
Loading
Loading