From a22ad6b54441f3c651a4daefae3f427df6aa135a Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Mon, 22 Jun 2026 21:13:45 +0200 Subject: [PATCH 1/8] First version of Deepgram transcribe --- CONTEXT.md | 18 ++ Cargo.toml | 3 + README.md | 7 + examples/transcribe.rs | 115 ++++++++++--- justfile | 3 + services/deepgram/Cargo.toml | 17 ++ services/deepgram/src/lib.rs | 4 + services/deepgram/src/transcribe.rs | 244 ++++++++++++++++++++++++++++ src/context_switch.rs | 1 + src/lib.rs | 1 + 10 files changed, 387 insertions(+), 26 deletions(-) create mode 100644 services/deepgram/Cargo.toml create mode 100644 services/deepgram/src/lib.rs create mode 100644 services/deepgram/src/transcribe.rs diff --git a/CONTEXT.md b/CONTEXT.md index fa94750..db7d34b 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -23,3 +23,21 @@ These three Microsoft offerings are distinct and must not all be called "Azure". Note: Voice Live and Azure OpenAI Realtime are expected to converge over time; the `microsoft-voice-live` boundary is kept deliberately thin so the two can be merged later without a rewrite. + +- **Deepgram Flux** — Deepgram's turn-based conversational speech-to-text API + (`listen` v2). Unlike classic streaming recognition, it emits per-turn events + rather than a continuous interim/final stream. Backs the `deepgram-service` + crate (`DeepgramTranscribe`, registered as `deepgram-transcribe`). + +- **Turn** — a single contiguous span of one speaker talking, as detected by + Flux. A turn accumulates transcript across `Update` events and is closed by an + `EndOfTurn`. + +- **End of Turn (EOT)** — Flux's decision that the speaker has finished the + current turn. Governed by `eotThreshold` (confidence to close a turn) and + `eotTimeoutMs` (silence after which a turn is force-closed). + +- **Eager End of Turn** — an early, lower-confidence EOT signal (enabled by + `eagerEotThreshold`) that lets a downstream agent begin preparing a reply + before the turn is confirmed. May be retracted by a `TurnResumed` event if the + speaker continues. diff --git a/Cargo.toml b/Cargo.toml index 25ecdf5..54d048f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,6 +12,7 @@ members = [ "filter-test", "services/aristech", "services/azure", + "services/deepgram", "services/elevenlabs", "services/google-dialog", "services/google-transcribe", @@ -37,6 +38,7 @@ google-dialog = { workspace = true } azure = { workspace = true } azure-speech = { workspace = true } aristech = { workspace = true } +deepgram-service = { workspace = true } elevenlabs = { workspace = true } google-transcribe = { workspace = true } microsoft-voice-live = { workspace = true } @@ -103,6 +105,7 @@ context-switch-core = { path = "core" } azure = { path = "services/azure" } playback = { path = "services/playback" } aristech = { path = "services/aristech" } +deepgram-service = { path = "services/deepgram" } elevenlabs = { path = "services/elevenlabs" } google-transcribe = { path = "services/google-transcribe" } google-dialog = { path = "services/google-dialog" } diff --git a/README.md b/README.md index d2c9871..2a3b031 100644 --- a/README.md +++ b/README.md @@ -63,6 +63,9 @@ cargo run --example openai-dialog # Run generic transcribe example with Azure provider cargo run --example transcribe -- azure +# Run generic transcribe example with Deepgram provider +cargo run --example transcribe -- deepgram + # Run generic transcribe example with ElevenLabs provider cargo run --example transcribe -- elevenlabs @@ -108,6 +111,10 @@ AZURE_REGION=your_azure_region # ElevenLabs Configuration ELEVENLABS_API_KEY=your_elevenlabs_key +# Deepgram Configuration +DEEPGRAM_API_KEY=your_deepgram_key +DEEPGRAM_ENDPOINT=wss://api.deepgram.com + # Microsoft Voice Live Configuration MICROSOFT_VOICE_LIVE_API_KEY=your_voice_live_key MICROSOFT_VOICE_LIVE_ENDPOINT=wss://your-resource.services.ai.azure.com/voice-live/realtime diff --git a/examples/transcribe.rs b/examples/transcribe.rs index 94aaa6a..1bf66f0 100644 --- a/examples/transcribe.rs +++ b/examples/transcribe.rs @@ -4,18 +4,19 @@ use std::time::Duration; use anyhow::{Context, Result, bail}; use clap::{Parser, ValueEnum}; +use tokio::select; +use tokio::sync::mpsc::{channel, unbounded_channel}; + use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; use openai_api_rs::realtime::types::{ AzureSemanticVadConfig, EndOfUtteranceDetectionConfig, EndOfUtteranceDetectionModel, EndOfUtteranceThresholdLevel, TurnDetection, }; use rodio::DeviceSinkBuilder; -use tokio::select; -use tokio::sync::mpsc::{channel, unbounded_channel}; use context_switch::services::{ - AristechTranscribe, AzureTranscribe, ElevenLabsTranscribe, GoogleTranscribe, - MicrosoftVoiceLiveTranscribe, + AristechTranscribe, AzureTranscribe, DeepgramTranscribe, ElevenLabsTranscribe, + GoogleTranscribe, MicrosoftVoiceLiveTranscribe, }; use context_switch::{AudioConsumer, InputModality, OutputModality}; use context_switch_core::language::Languages; @@ -51,6 +52,8 @@ enum Provider { Aristech, #[value(name = "voice-live")] VoiceLive, + #[value(name = "deepgram")] + Deepgram, } #[tokio::main] @@ -240,11 +243,10 @@ async fn start_conversation( diarization: bool, conversation: Conversation, ) -> Result<()> { + validate_provider_args(provider, model, region, diarization)?; + match provider { Provider::Azure => { - if region.is_some() { - bail!("--region is only supported for the google provider"); - } let params = azure::transcribe::Params { endpoint: env::var("AZURE_ENDPOINT") .ok() @@ -259,12 +261,6 @@ async fn start_conversation( AzureTranscribe.conversation(params, conversation).await } Provider::Elevenlabs => { - if diarization { - bail!("--diarization is only supported for the azure provider"); - } - if region.is_some() { - bail!("--region is only supported for the google provider"); - } let language = Some( languages .single() @@ -317,12 +313,6 @@ async fn start_conversation( GoogleTranscribe.conversation(params, conversation).await } Provider::Aristech => { - if diarization { - bail!("--diarization is only supported for the azure provider"); - } - if region.is_some() { - bail!("--region is only supported for the google provider"); - } let language = languages .single() .context("Aristech provider supports exactly one --language value")? @@ -351,13 +341,6 @@ async fn start_conversation( AristechTranscribe.conversation(params, conversation).await } Provider::VoiceLive => { - if diarization { - bail!("--diarization is only supported for the azure provider"); - } - if region.is_some() { - bail!("--region is only supported for the google provider"); - } - let language = Some( languages .single() @@ -396,5 +379,85 @@ async fn start_conversation( .conversation(params, conversation) .await } + Provider::Deepgram => { + let params = deepgram_service::transcribe::Params { + api_key: env::var("DEEPGRAM_API_KEY").expect("DEEPGRAM_API_KEY undefined"), + endpoint: env::var("DEEPGRAM_ENDPOINT").expect("DEEPGRAM_ENDPOINT undefined"), + language: languages.join_csv(), + profanity_filter: false, + keyterm: vec![], + turn_detection: deepgram_service::transcribe::TurnDetection::default(), + }; + + DeepgramTranscribe.conversation(params, conversation).await + } } } + +#[derive(Debug, Clone, Copy, Default)] +struct ProviderCapabilities { + region: bool, + diarization: bool, + model: bool, +} + +impl Provider { + fn capabilities(self) -> ProviderCapabilities { + let mut capabilities = ProviderCapabilities::default(); + + match self { + Provider::Azure => { + capabilities.diarization = true; + capabilities.model = true; + } + Provider::Deepgram => {} + Provider::Elevenlabs => { + capabilities.model = true; + } + Provider::Google => { + capabilities.region = true; + capabilities.diarization = true; + capabilities.model = true; + } + Provider::Aristech => { + capabilities.model = true; + } + Provider::VoiceLive => { + capabilities.model = true; + } + } + + capabilities + } +} + +fn validate_capability( + option_name: &str, + is_used: bool, + capability: bool, + provider: Provider, +) -> Result<()> { + if !is_used || capability { + return Ok(()); + } + + bail!("{option_name} is unsupported for provider '{provider:?}'") +} + +fn validate_provider_args( + provider: Provider, + model: Option<&str>, + region: Option<&str>, + diarization: bool, +) -> Result<()> { + let capabilities = provider.capabilities(); + + validate_capability("--model", model.is_some(), capabilities.model, provider)?; + validate_capability("--region", region.is_some(), capabilities.region, provider)?; + validate_capability( + "--diarization", + diarization, + capabilities.diarization, + provider, + ) +} diff --git a/justfile b/justfile index 6a026aa..e811348 100644 --- a/justfile +++ b/justfile @@ -28,6 +28,9 @@ transcribe-google-latest-long: transcribe-voice-live-de: cargo run --example transcribe -- voice-live --language de-DE +transcribe-deepgram-de: + cargo run --example transcribe -- deepgram --language de-DE + transcribe-google-diarization: cargo run --example transcribe -- google --diarization --language de-DE --model chirp_3 --region eu diff --git a/services/deepgram/Cargo.toml b/services/deepgram/Cargo.toml new file mode 100644 index 0000000..cc10e8c --- /dev/null +++ b/services/deepgram/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "deepgram-service" +version = "0.1.0" +edition.workspace = true + +[dependencies] +context-switch-core = { workspace = true } + +deepgram = "0.10.0" + +anyhow = { workspace = true } +async-trait = { workspace = true } +bytes = { workspace = true } +futures = { workspace = true } +serde = { workspace = true, features = ["derive"] } +tokio = { workspace = true } +tracing = { workspace = true } diff --git a/services/deepgram/src/lib.rs b/services/deepgram/src/lib.rs new file mode 100644 index 0000000..401cf86 --- /dev/null +++ b/services/deepgram/src/lib.rs @@ -0,0 +1,4 @@ +//! A Deepgram Flux speech-to-text service. + +pub mod transcribe; +pub use transcribe::DeepgramTranscribe; diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs new file mode 100644 index 0000000..d2253f3 --- /dev/null +++ b/services/deepgram/src/transcribe.rs @@ -0,0 +1,244 @@ +use std::io; + +use anyhow::{Context, Result, bail}; +use async_trait::async_trait; +use bytes::Bytes; +use futures::channel::mpsc; +use futures::{SinkExt, StreamExt}; +use serde::{Deserialize, Serialize}; +use tokio::task; +use tracing::{debug, warn}; + +use deepgram::Deepgram; +use deepgram::common::flux_response::{FluxResponse, TurnEvent}; +use deepgram::common::options::{Encoding, Model, Options}; + +use context_switch_core::language::{Languages, bcp47_to_iso639_3}; +use context_switch_core::{ + BillingRecord, BillingSchedule, Conversation, Input, OutputPath, Service, +}; + +#[derive(Debug, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct Params { + pub api_key: String, + #[serde(alias = "host")] + pub endpoint: String, + pub language: String, + #[serde(default)] + pub profanity_filter: bool, + #[serde(default)] + pub keyterm: Vec, + #[serde(flatten)] + pub turn_detection: TurnDetection, +} + +#[derive(Debug, Default, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct TurnDetection { + pub threshold: Option, + pub timeout_ms: Option, + pub eager_threshold: Option, +} + +#[derive(Debug)] +pub struct DeepgramTranscribe; + +#[derive(Debug, Serialize)] +#[serde(rename_all = "camelCase")] +enum ServiceOutputEvent { + StartOfTurn, + EagerEndOfTurn, + TurnResumed, +} + +#[async_trait] +impl Service for DeepgramTranscribe { + type Params = Params; + + async fn conversation(&self, params: Params, conversation: Conversation) -> Result<()> { + let input_format = conversation.require_audio_input()?; + conversation.require_text_output(true)?; + + let languages = Languages::from_csv(¶ms.language) + .context("language must contain at least one locale code")?; + + let (model, language_hints) = select_model_and_language_hints(&languages)?; + let mut options_builder = Options::builder().model(model); + + if let Some(eot_threshold) = params.turn_detection.threshold { + options_builder = options_builder.eot_threshold(eot_threshold); + } + if let Some(eot_timeout_ms) = params.turn_detection.timeout_ms { + options_builder = options_builder.eot_timeout_ms(eot_timeout_ms); + } + if let Some(eager_eot_threshold) = params.turn_detection.eager_threshold { + options_builder = options_builder.eager_eot_threshold(eager_eot_threshold); + } + if params.profanity_filter { + options_builder = options_builder.profanity_filter(true); + } + + let options_builder = if let Some(language_hints) = language_hints { + options_builder.language_hint(language_hints) + } else { + options_builder + }; + + let options_builder = if params.keyterm.is_empty() { + options_builder + } else { + options_builder.keyterms(params.keyterm.iter().map(String::as_str)) + }; + + let options = options_builder.build(); + + // ADR: endpoint is required for GDPR-safe explicit routing. + let deepgram = + Deepgram::with_base_url_and_api_key(params.endpoint.as_str(), params.api_key)?; + + let (mut input, output) = conversation.start()?; + let (mut audio_tx, audio_rx) = mpsc::channel::>(8); + + let billing_output = output.clone(); + let forward_audio_task = task::spawn(async move { + while let Some(Input::Audio { frame }) = input.recv().await { + let duration = frame.duration(); + if let Err(error) = billing_output.billing_records( + None, + None, + [BillingRecord::duration("input:audio", duration)], + BillingSchedule::Now, + ) { + return Err(io::Error::other(format!( + "Failed to output billing records: {error}" + ))); + } + + if let Err(error) = audio_tx.send(Ok(Bytes::from(frame.to_le_bytes()))).await { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + format!("Deepgram audio stream channel closed: {error}"), + )); + } + } + + Ok::<(), io::Error>(()) + }); + + let mut stream = deepgram + .transcription() + .flux_request_with_options(options) + .encoding(Encoding::Linear16) + .sample_rate(input_format.sample_rate) + .stream(audio_rx) + .await?; + + while let Some(message) = stream.next().await { + let response = message?; + match response { + FluxResponse::Connected { .. } => {} + FluxResponse::ConfigureSuccess { .. } => {} + FluxResponse::ConfigureFailure { .. } => { + bail!("Deepgram rejected a Flux reconfiguration update"); + } + FluxResponse::FatalError { + code, description, .. + } => { + bail!("Deepgram stream error ({code}): {description}"); + } + FluxResponse::TurnInfo { + event, + transcript, + languages, + .. + } => { + let language = languages.first().cloned(); + + match event { + TurnEvent::Update => { + if !transcript.is_empty() { + output.text(false, transcript, language, None)?; + } + } + TurnEvent::EndOfTurn => { + if !transcript.is_empty() { + output.text(true, transcript, language, None)?; + } + } + TurnEvent::StartOfTurn => { + output.service_event( + OutputPath::Media, + ServiceOutputEvent::StartOfTurn, + )?; + } + TurnEvent::EagerEndOfTurn => { + output.service_event( + OutputPath::Media, + ServiceOutputEvent::EagerEndOfTurn, + )?; + } + TurnEvent::TurnResumed => { + output.service_event( + OutputPath::Media, + ServiceOutputEvent::TurnResumed, + )?; + } + TurnEvent::Unknown => { + warn!( + transcript = %transcript, + languages = ?languages, + "Deepgram returned unknown turn event" + ); + } + _ => { + warn!( + event = ?event, + transcript = %transcript, + languages = ?languages, + "Deepgram returned unhandled turn event variant" + ); + } + } + } + FluxResponse::Unknown(value) => { + warn!( + payload = %value, + "Deepgram returned unknown Flux response payload" + ); + } + other => { + debug!(?other, "Deepgram returned unhandled Flux response variant"); + } + } + } + + let audio_forward_result = forward_audio_task + .await + .context("Joining Deepgram audio forward task")?; + audio_forward_result.context("Deepgram audio forward task failed")?; + + Ok(()) + } +} + +fn select_model_and_language_hints(languages: &Languages) -> Result<(Model, Option>)> { + if languages.len() > 1 { + return Ok(( + Model::FluxGeneralMulti, + Some(languages.iter().cloned().collect()), + )); + } + + let is_english = bcp47_to_iso639_3(languages.first()) + .context("language must be a valid BCP 47 tag")? + == "eng"; + if is_english { + Ok((Model::FluxGeneralEn, None)) + } else { + Ok(( + Model::FluxGeneralMulti, + Some(vec![languages.first().to_owned()]), + )) + } +} diff --git a/src/context_switch.rs b/src/context_switch.rs index 776d192..6357525 100644 --- a/src/context_switch.rs +++ b/src/context_switch.rs @@ -40,6 +40,7 @@ pub fn registry() -> Registry { .add_service("azure-transcribe", azure::AzureTranscribe) .add_service("azure-synthesize", azure::AzureSynthesize) .add_service("azure-translate", azure::AzureTranslate) + .add_service("deepgram-transcribe", deepgram_service::DeepgramTranscribe) .add_service("elevenlabs-transcribe", elevenlabs::ElevenLabsTranscribe) .add_service("google-transcribe", google_transcribe::GoogleTranscribe) .add_service( diff --git a/src/lib.rs b/src/lib.rs index a749929..72f441d 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -14,6 +14,7 @@ pub use speech_gate::make_speech_gate_processor; pub mod services { pub use aristech::AristechTranscribe; pub use azure::AzureTranscribe; + pub use deepgram_service::DeepgramTranscribe; pub use elevenlabs::ElevenLabsTranscribe; pub use google_dialog::GoogleDialog; pub use google_transcribe::GoogleTranscribe; From 1c6529bdaeabafd2121b1641169e34aabef84c96 Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Mon, 22 Jun 2026 21:20:00 +0200 Subject: [PATCH 2/8] deepgram-transcribe: Support a complete endpoint --- README.md | 2 +- services/deepgram/src/transcribe.rs | 30 ++++++++++++++++++++++++++--- 2 files changed, 28 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 2a3b031..e6119a2 100644 --- a/README.md +++ b/README.md @@ -113,7 +113,7 @@ ELEVENLABS_API_KEY=your_elevenlabs_key # Deepgram Configuration DEEPGRAM_API_KEY=your_deepgram_key -DEEPGRAM_ENDPOINT=wss://api.deepgram.com +DEEPGRAM_ENDPOINT=wss://api.deepgram.com/v2/listen # Microsoft Voice Live Configuration MICROSOFT_VOICE_LIVE_API_KEY=your_voice_live_key diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs index d2253f3..7c8fb79 100644 --- a/services/deepgram/src/transcribe.rs +++ b/services/deepgram/src/transcribe.rs @@ -7,7 +7,7 @@ use futures::channel::mpsc; use futures::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; use tokio::task; -use tracing::{debug, warn}; +use tracing::{debug, info, warn}; use deepgram::Deepgram; use deepgram::common::flux_response::{FluxResponse, TurnEvent}; @@ -92,10 +92,14 @@ impl Service for DeepgramTranscribe { }; let options = options_builder.build(); + info!(endpoint = %params.endpoint, "Using Deepgram endpoint"); + let endpoint = normalize_endpoint(¶ms.endpoint)?; + if endpoint != params.endpoint { + info!(endpoint = %endpoint, "Normalized Deepgram endpoint to base URL"); + } // ADR: endpoint is required for GDPR-safe explicit routing. - let deepgram = - Deepgram::with_base_url_and_api_key(params.endpoint.as_str(), params.api_key)?; + let deepgram = Deepgram::with_base_url_and_api_key(endpoint.as_str(), params.api_key)?; let (mut input, output) = conversation.start()?; let (mut audio_tx, audio_rx) = mpsc::channel::>(8); @@ -242,3 +246,23 @@ fn select_model_and_language_hints(languages: &Languages) -> Result<(Model, Opti )) } } + +fn normalize_endpoint(endpoint: &str) -> Result { + let trimmed = endpoint.trim_end_matches('/'); + + for listen_path in ["/v1/listen", "/v2/listen"] { + if let Some(prefix) = trimmed.strip_suffix(listen_path) { + return Ok(format!("{prefix}/")); + } + } + + if trimmed.contains("/listen") { + bail!( + "Invalid DEEPGRAM_ENDPOINT: expected terminal /v1/listen or /v2/listen path ({endpoint})" + ); + } + + bail!( + "Invalid DEEPGRAM_ENDPOINT: expected full listen path (for example wss://api.deepgram.com/v2/listen), got base URL ({endpoint})" + ) +} From 0e81bebbfdd40d5d924928d5fe1641d64de4505f Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Tue, 23 Jun 2026 19:01:21 +0200 Subject: [PATCH 3/8] deepgram-transcribe: The set EndOfUtteranceThresholdLevel to default --- examples/transcribe.rs | 2 +- services/microsoft-voice-live/src/transcribe.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/examples/transcribe.rs b/examples/transcribe.rs index 1bf66f0..009d3e0 100644 --- a/examples/transcribe.rs +++ b/examples/transcribe.rs @@ -366,7 +366,7 @@ async fn start_conversation( AzureSemanticVadConfig { end_of_utterance_detection: Some(EndOfUtteranceDetectionConfig { model: EndOfUtteranceDetectionModel::SmartEndOfTurnDetection, - threshold_level: Some(EndOfUtteranceThresholdLevel::Low), + threshold_level: Some(EndOfUtteranceThresholdLevel::Default), timeout_ms: Some(5000), }), // remove_filler_words: Some(true), diff --git a/services/microsoft-voice-live/src/transcribe.rs b/services/microsoft-voice-live/src/transcribe.rs index b7e4abe..827e887 100644 --- a/services/microsoft-voice-live/src/transcribe.rs +++ b/services/microsoft-voice-live/src/transcribe.rs @@ -26,7 +26,7 @@ pub struct Params { /// Transcription model set in `audio.input.transcription.model`. #[serde(default = "default_transcription_model")] pub transcription_model: String, - /// Input audio language hint in ISO-639-1 form (e.g. `en`). + /// Input audio language hint in `ISO-639-1` form (e.g. `en`). pub language: Option, /// Input-audio noise reduction (Azure deep noise suppression, near/far field). pub noise_reduction: Option, From 2677c58346a562a7540cbf01752daeeee43df8b5 Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Tue, 23 Jun 2026 20:32:08 +0200 Subject: [PATCH 4/8] deepgram-transcribe: Don't end on an unexpected input event --- services/deepgram/src/transcribe.rs | 43 +++++++++++++++++------------ 1 file changed, 26 insertions(+), 17 deletions(-) diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs index 7c8fb79..fcd7cc4 100644 --- a/services/deepgram/src/transcribe.rs +++ b/services/deepgram/src/transcribe.rs @@ -106,24 +106,33 @@ impl Service for DeepgramTranscribe { let billing_output = output.clone(); let forward_audio_task = task::spawn(async move { - while let Some(Input::Audio { frame }) = input.recv().await { - let duration = frame.duration(); - if let Err(error) = billing_output.billing_records( - None, - None, - [BillingRecord::duration("input:audio", duration)], - BillingSchedule::Now, - ) { - return Err(io::Error::other(format!( - "Failed to output billing records: {error}" - ))); - } + while let Some(input_event) = input.recv().await { + match input_event { + Input::Audio { frame } => { + let duration = frame.duration(); + if let Err(error) = billing_output.billing_records( + None, + None, + [BillingRecord::duration("input:audio", duration)], + BillingSchedule::Now, + ) { + return Err(io::Error::other(format!( + "Failed to output billing records: {error}" + ))); + } - if let Err(error) = audio_tx.send(Ok(Bytes::from(frame.to_le_bytes()))).await { - return Err(io::Error::new( - io::ErrorKind::BrokenPipe, - format!("Deepgram audio stream channel closed: {error}"), - )); + if let Err(error) = + audio_tx.send(Ok(Bytes::from(frame.to_le_bytes()))).await + { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + format!("Deepgram audio stream channel closed: {error}"), + )); + } + } + _ => { + warn!("Ignoring non-audio input in Deepgram transcribe audio forwarder"); + } } } From 591a4355fa8a00d84c3f5c9289c5fd2b13799e5a Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Tue, 23 Jun 2026 20:44:38 +0200 Subject: [PATCH 5/8] deepgram-transcribe: Make termination more robust by driving audio fowarding and message processing within the same select loop --- .github/copilot-instructions.md | 3 ++ services/deepgram/src/transcribe.rs | 84 ++++++++++++++--------------- 2 files changed, 44 insertions(+), 43 deletions(-) diff --git a/.github/copilot-instructions.md b/.github/copilot-instructions.md index d35f0ff..38570e8 100644 --- a/.github/copilot-instructions.md +++ b/.github/copilot-instructions.md @@ -24,6 +24,9 @@ ## Control Flow Style - Prefer exhaustive `match` statements for enum-based control flow instead of `if matches!(...)` shortcuts. +## Streaming Services +- Streaming transcribe/translate services drive the provider response stream and audio forwarding (plus billing) in a single `select!` loop — never a detached forwarder task — so termination and billing stay deterministic. + ## Memory Promotion - When a durable repository-specific preference is learned during a session, write it into this file as a concise bullet if it can help future sessions. - Keep additions short, actionable, and scoped to coding behavior in this repository. diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs index fcd7cc4..4448e19 100644 --- a/services/deepgram/src/transcribe.rs +++ b/services/deepgram/src/transcribe.rs @@ -6,7 +6,7 @@ use bytes::Bytes; use futures::channel::mpsc; use futures::{SinkExt, StreamExt}; use serde::{Deserialize, Serialize}; -use tokio::task; +use tokio::select; use tracing::{debug, info, warn}; use deepgram::Deepgram; @@ -104,41 +104,6 @@ impl Service for DeepgramTranscribe { let (mut input, output) = conversation.start()?; let (mut audio_tx, audio_rx) = mpsc::channel::>(8); - let billing_output = output.clone(); - let forward_audio_task = task::spawn(async move { - while let Some(input_event) = input.recv().await { - match input_event { - Input::Audio { frame } => { - let duration = frame.duration(); - if let Err(error) = billing_output.billing_records( - None, - None, - [BillingRecord::duration("input:audio", duration)], - BillingSchedule::Now, - ) { - return Err(io::Error::other(format!( - "Failed to output billing records: {error}" - ))); - } - - if let Err(error) = - audio_tx.send(Ok(Bytes::from(frame.to_le_bytes()))).await - { - return Err(io::Error::new( - io::ErrorKind::BrokenPipe, - format!("Deepgram audio stream channel closed: {error}"), - )); - } - } - _ => { - warn!("Ignoring non-audio input in Deepgram transcribe audio forwarder"); - } - } - } - - Ok::<(), io::Error>(()) - }); - let mut stream = deepgram .transcription() .flux_request_with_options(options) @@ -147,8 +112,46 @@ impl Service for DeepgramTranscribe { .stream(audio_rx) .await?; - while let Some(message) = stream.next().await { - let response = message?; + // Drive audio forwarding (with billing) and Deepgram response processing in a single loop so + // termination and billing stay deterministic: any error or end-of-input breaks immediately, + // and closing `audio_tx` triggers the SDK finalize/close handshake that drains final turns. + let mut audio_input_open = true; + loop { + let response = select! { + input_event = input.recv(), if audio_input_open => { + match input_event { + Some(Input::Audio { frame }) => { + let duration = frame.duration(); + output + .billing_records( + None, + None, + [BillingRecord::duration("input:audio", duration)], + BillingSchedule::Now, + ) + .context("Failed to output billing records")?; + audio_tx + .send(Ok(Bytes::from(frame.to_le_bytes()))) + .await + .context("Deepgram audio stream channel closed")?; + } + // Any non-audio input or a closed input channel ends audio forwarding. + // Closing `audio_tx` lets the SDK finalize and deliver the remaining turns. + Some(_) | None => { + audio_input_open = false; + audio_tx.close_channel(); + } + } + continue; + } + message = stream.next() => { + match message { + Some(message) => message?, + None => break, + } + } + }; + match response { FluxResponse::Connected { .. } => {} FluxResponse::ConfigureSuccess { .. } => {} @@ -226,11 +229,6 @@ impl Service for DeepgramTranscribe { } } - let audio_forward_result = forward_audio_task - .await - .context("Joining Deepgram audio forward task")?; - audio_forward_result.context("Deepgram audio forward task failed")?; - Ok(()) } } From 4c3163d339ce6da9b1edb1648038a145ab0fba90 Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Tue, 23 Jun 2026 20:48:03 +0200 Subject: [PATCH 6/8] deepgram-transcribe: Output billing records after the audio was sent --- services/deepgram/src/transcribe.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs index 4448e19..6f4a3cd 100644 --- a/services/deepgram/src/transcribe.rs +++ b/services/deepgram/src/transcribe.rs @@ -122,6 +122,10 @@ impl Service for DeepgramTranscribe { match input_event { Some(Input::Audio { frame }) => { let duration = frame.duration(); + audio_tx + .send(Ok(Bytes::from(frame.to_le_bytes()))) + .await + .context("Deepgram audio stream channel closed")?; output .billing_records( None, @@ -130,10 +134,6 @@ impl Service for DeepgramTranscribe { BillingSchedule::Now, ) .context("Failed to output billing records")?; - audio_tx - .send(Ok(Bytes::from(frame.to_le_bytes()))) - .await - .context("Deepgram audio stream channel closed")?; } // Any non-audio input or a closed input channel ends audio forwarding. // Closing `audio_tx` lets the SDK finalize and deliver the remaining turns. From a3848df91e1890c5d2a1f88d2dce0d2e20565896 Mon Sep 17 00:00:00 2001 From: Armin Date: Tue, 23 Jun 2026 20:51:00 +0200 Subject: [PATCH 7/8] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- services/deepgram/src/transcribe.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs index 6f4a3cd..7dfbd87 100644 --- a/services/deepgram/src/transcribe.rs +++ b/services/deepgram/src/transcribe.rs @@ -265,7 +265,7 @@ fn normalize_endpoint(endpoint: &str) -> Result { if trimmed.contains("/listen") { bail!( - "Invalid DEEPGRAM_ENDPOINT: expected terminal /v1/listen or /v2/listen path ({endpoint})" + "Invalid Deepgram endpoint: expected terminal /v1/listen or /v2/listen path ({endpoint})" ); } From 7fc631953fd0c5d291ff3add942c62a262237c3b Mon Sep 17 00:00:00 2001 From: Armin Sander Date: Tue, 23 Jun 2026 20:56:58 +0200 Subject: [PATCH 8/8] Incorporate review feedback --- examples/transcribe.rs | 8 +++++++- services/deepgram/src/transcribe.rs | 6 ++---- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/examples/transcribe.rs b/examples/transcribe.rs index 009d3e0..130c6b8 100644 --- a/examples/transcribe.rs +++ b/examples/transcribe.rs @@ -441,7 +441,13 @@ fn validate_capability( return Ok(()); } - bail!("{option_name} is unsupported for provider '{provider:?}'") + bail!( + "{option_name} is unsupported for provider '{}'", + provider + .to_possible_value() + .expect("Provider has a possible value") + .get_name() + ) } fn validate_provider_args( diff --git a/services/deepgram/src/transcribe.rs b/services/deepgram/src/transcribe.rs index 7dfbd87..a2d584f 100644 --- a/services/deepgram/src/transcribe.rs +++ b/services/deepgram/src/transcribe.rs @@ -264,12 +264,10 @@ fn normalize_endpoint(endpoint: &str) -> Result { } if trimmed.contains("/listen") { - bail!( - "Invalid Deepgram endpoint: expected terminal /v1/listen or /v2/listen path ({endpoint})" - ); + bail!("invalid endpoint: expected terminal /v1/listen or /v2/listen path ({endpoint})"); } bail!( - "Invalid DEEPGRAM_ENDPOINT: expected full listen path (for example wss://api.deepgram.com/v2/listen), got base URL ({endpoint})" + "invalid endpoint: expected full listen path (for example wss://api.deepgram.com/v2/listen), got base URL ({endpoint})" ) }