foundry_local_sdk/openai/
live_audio_session.rs

1//! Live audio transcription streaming session.
2//!
3//! Provides real-time audio streaming ASR (Automatic Speech Recognition).
4//! Audio data from a microphone (or other source) is pushed in as PCM chunks
5//! and transcription results are returned as an async [`Stream`](futures_core::Stream).
6//!
7//! # Example
8//!
9//! ```ignore
10//! let audio_client = model.create_audio_client();
11//! let mut session = audio_client.create_live_transcription_session();
12//! session.settings.sample_rate = 16000;
13//! session.settings.channels = 1;
14//! session.settings.language = Some("en".into());
15//!
16//! session.start(None).await?;
17//!
18//! // Push audio from microphone callback
19//! session.append(&pcm_bytes, None).await?;
20//!
21//! // Read results as async stream
22//! use tokio_stream::StreamExt;
23//! let mut stream = session.get_stream().await?;
24//! while let Some(result) = stream.next().await {
25//!     let result = result?;
26//!     print!("{}", result.content[0].text);
27//! }
28//!
29//! session.stop(None).await?;
30//! ```
31#![allow(deprecated)] // this module implements the deprecated OpenAI facade
32
33use std::os::raw::c_int;
34use std::panic::{catch_unwind, AssertUnwindSafe};
35use std::pin::Pin;
36use std::sync::Arc;
37use std::task::{Context, Poll};
38
39use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
40use tokio_util::sync::CancellationToken;
41
42use crate::detail::api::Api;
43use crate::detail::ffi::{flItem, flStreamingCallbackData, FOUNDRY_LOCAL_ITEM_BYTES};
44use crate::detail::items::{
45    make_audio_item, read_speech_segment, read_text_item, SpeechSegmentText,
46};
47use crate::detail::native::NativeModel;
48use crate::detail::session::{NativeItemQueue, NativeRequest, NativeSession};
49use crate::detail::task::spawn_blocking;
50use crate::error::{FoundryLocalError, Result};
51
52// ── Types ────────────────────────────────────────────────────────────────────
53
54/// Audio format settings for a live transcription session.
55///
56/// Must be configured before calling [`LiveAudioTranscriptionSession::start`].
57/// Settings are frozen once the session starts.
58#[derive(Debug, Clone)]
59pub struct LiveAudioTranscriptionOptions {
60    /// PCM sample rate in Hz. Default: 16000.
61    pub sample_rate: u32,
62    /// Number of audio channels. Default: 1 (mono).
63    pub channels: u32,
64    /// Optional BCP-47 language hint (e.g., `"en"`, `"zh"`).
65    pub language: Option<String>,
66}
67
68impl Default for LiveAudioTranscriptionOptions {
69    fn default() -> Self {
70        Self {
71            sample_rate: 16000,
72            channels: 1,
73            language: None,
74        }
75    }
76}
77
78/// Internal raw deserialization target matching the native core's JSON format.
79#[derive(Debug, Clone, serde::Deserialize)]
80struct LiveAudioTranscriptionRaw {
81    #[serde(default)]
82    is_final: bool,
83    #[serde(default)]
84    text: String,
85    start_time: Option<f64>,
86    end_time: Option<f64>,
87    id: Option<String>,
88}
89
90/// A content part within a [`LiveAudioTranscriptionResponse`].
91///
92/// Mirrors the C# `ContentPart` shape from the OpenAI Realtime API so that
93/// callers can access `result.content[0].text` or `result.content[0].transcript`
94/// consistently across SDKs.
95#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
96pub struct ContentPart {
97    /// The transcribed text.
98    pub text: String,
99    /// Same as `text` — provided for OpenAI Realtime API compatibility.
100    pub transcript: String,
101}
102
103/// Transcription result from a live audio streaming session.
104///
105/// Shaped to match the C# `LiveAudioTranscriptionResponse : ConversationItem`
106/// so that callers access text via `result.content[0].text` or
107/// `result.content[0].transcript`.
108#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
109pub struct LiveAudioTranscriptionResponse {
110    /// Content parts — typically a single element. Access text via
111    /// `result.content[0].text` or `result.content[0].transcript`.
112    pub content: Vec<ContentPart>,
113    /// Whether this is a final or partial (interim) result.
114    /// Nemotron models always return `true`; other models may return `false`
115    /// for interim hypotheses that will be replaced by a subsequent final result.
116    pub is_final: bool,
117    /// Start time offset of this segment in the audio stream (seconds).
118    pub start_time: Option<f64>,
119    /// End time offset of this segment in the audio stream (seconds).
120    pub end_time: Option<f64>,
121    /// Unique identifier for this result (if available).
122    pub id: Option<String>,
123}
124
125impl LiveAudioTranscriptionResponse {
126    /// Parse a transcription response from the native core's JSON format.
127    pub fn from_json(json: &str) -> Result<Self> {
128        serde_json::from_str::<LiveAudioTranscriptionRaw>(json)
129            .map(Self::from_raw)
130            .map_err(FoundryLocalError::from)
131    }
132
133    fn from_raw(raw: LiveAudioTranscriptionRaw) -> Self {
134        Self {
135            content: vec![ContentPart {
136                transcript: raw.text.clone(),
137                text: raw.text,
138            }],
139            is_final: raw.is_final,
140            start_time: raw.start_time,
141            end_time: raw.end_time,
142            id: raw.id,
143        }
144    }
145
146    /// Build a response from a plain transcript string.
147    fn from_text(text: String, is_final: bool) -> Self {
148        Self {
149            content: vec![ContentPart {
150                transcript: text.clone(),
151                text,
152            }],
153            is_final,
154            start_time: None,
155            end_time: None,
156            id: None,
157        }
158    }
159
160    /// Build a response from a SPEECH_SEGMENT item's content.
161    fn from_segment(seg: SpeechSegmentText) -> Self {
162        Self {
163            content: vec![ContentPart {
164                transcript: seg.text.clone(),
165                text: seg.text,
166            }],
167            is_final: seg.is_final,
168            start_time: seg.start_time_s,
169            end_time: seg.end_time_s,
170            id: None,
171        }
172    }
173}
174
175/// Structured error response from the native core.
176#[derive(Debug, Clone, serde::Deserialize)]
177pub struct CoreErrorResponse {
178    /// Error code (e.g. `"ASR_SESSION_NOT_FOUND"`).
179    pub code: String,
180    /// Human-readable error message.
181    pub message: String,
182    /// Whether this error is transient (retryable).
183    #[serde(rename = "isTransient", default)]
184    pub is_transient: bool,
185}
186
187impl CoreErrorResponse {
188    /// Attempt to parse a native error string as structured JSON.
189    /// Returns `None` if the error is not valid JSON or doesn't match the schema.
190    pub fn try_parse(error_string: &str) -> Option<Self> {
191        serde_json::from_str(error_string).ok()
192    }
193}
194
195// ── Stream type ──────────────────────────────────────────────────────────────
196
197/// An async stream of [`LiveAudioTranscriptionResponse`] items.
198///
199/// Returned by [`LiveAudioTranscriptionSession::get_stream`].
200/// Implements [`futures_core::Stream`].
201pub struct LiveAudioTranscriptionStream {
202    rx: UnboundedReceiver<Result<LiveAudioTranscriptionResponse>>,
203}
204
205impl futures_core::Stream for LiveAudioTranscriptionStream {
206    type Item = Result<LiveAudioTranscriptionResponse>;
207
208    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
209        self.rx.poll_recv(cx)
210    }
211}
212
213// ── Session state ────────────────────────────────────────────────────────────
214
215#[derive(Default)]
216struct SessionState {
217    started: bool,
218    stopped: bool,
219    queue: Option<Arc<NativeItemQueue>>,
220    output_rx: Option<UnboundedReceiver<Result<LiveAudioTranscriptionResponse>>>,
221    worker: Option<tokio::task::JoinHandle<()>>,
222}
223
224// ── Session ──────────────────────────────────────────────────────────────────
225
226/// Session for real-time audio streaming ASR (Automatic Speech Recognition).
227///
228/// Audio data from a microphone (or other source) is pushed in as PCM chunks
229/// via [`append`](Self::append), and transcription results are returned as an
230/// async [`Stream`](futures_core::Stream) via [`get_stream`](Self::get_stream).
231///
232/// Created via [`AudioClient::create_live_transcription_session`](super::AudioClient::create_live_transcription_session).
233///
234/// # Cancellation
235///
236/// All lifecycle methods accept an optional [`CancellationToken`]. Pass `None`
237/// to use the default (no cancellation).
238#[deprecated(
239    since = "2.0.0",
240    note = "The OpenAI direct clients are deprecated; use the Session API instead \
241            (`AudioSession::new(&model)` with streaming)."
242)]
243pub struct LiveAudioTranscriptionSession {
244    model: NativeModel,
245    /// Audio format settings. Must be configured before calling [`start`](Self::start).
246    /// Settings are frozen once the session starts.
247    pub settings: LiveAudioTranscriptionOptions,
248    state: tokio::sync::Mutex<SessionState>,
249}
250
251impl LiveAudioTranscriptionSession {
252    pub(crate) fn new(_model_id: &str, model: NativeModel) -> Self {
253        Self {
254            model,
255            settings: LiveAudioTranscriptionOptions::default(),
256            state: tokio::sync::Mutex::new(SessionState::default()),
257        }
258    }
259
260    /// Start a real-time audio streaming session.
261    ///
262    /// Must be called before [`append`](Self::append) or
263    /// [`get_stream`](Self::get_stream). Settings are frozen after this call.
264    pub async fn start(&self, ct: Option<CancellationToken>) -> Result<()> {
265        let mut state = self.state.lock().await;
266
267        if state.started {
268            return Err(FoundryLocalError::Validation {
269                reason: "Streaming session already started. Call stop() first.".into(),
270            });
271        }
272
273        if let Some(token) = &ct {
274            if token.is_cancelled() {
275                return Err(FoundryLocalError::CommandExecution {
276                    reason: "Start cancelled".into(),
277                });
278            }
279        }
280
281        let settings = self.settings.clone();
282        let model = self.model.clone();
283        let api = Arc::clone(&model.api);
284
285        // Create the shared input queue up front so `append` can push into it.
286        let queue = Arc::new(NativeItemQueue::new(api)?);
287
288        let (output_tx, output_rx) =
289            tokio::sync::mpsc::unbounded_channel::<Result<LiveAudioTranscriptionResponse>>();
290
291        let worker_queue = Arc::clone(&queue);
292        let worker = tokio::task::spawn_blocking(move || {
293            run_worker(model, settings, worker_queue, output_tx);
294        });
295
296        state.started = true;
297        state.stopped = false;
298        state.queue = Some(queue);
299        state.output_rx = Some(output_rx);
300        state.worker = Some(worker);
301
302        Ok(())
303    }
304
305    /// Push a chunk of raw PCM audio data to the streaming session.
306    ///
307    /// The data is copied internally so the caller can reuse the buffer.
308    pub async fn append(&self, pcm_data: &[u8], ct: Option<CancellationToken>) -> Result<()> {
309        if let Some(token) = &ct {
310            if token.is_cancelled() {
311                return Err(FoundryLocalError::CommandExecution {
312                    reason: "Append cancelled".into(),
313                });
314            }
315        }
316
317        let queue = {
318            let state = self.state.lock().await;
319            if !state.started || state.stopped {
320                return Err(FoundryLocalError::Validation {
321                    reason: "No active streaming session. Call start() first.".into(),
322                });
323            }
324            state
325                .queue
326                .clone()
327                .ok_or_else(|| FoundryLocalError::Internal {
328                    reason: "Input queue not available — session may be in an invalid state".into(),
329                })?
330        };
331
332        let data = pcm_data.to_vec();
333        spawn_blocking(move || queue.push_bytes(&data, FOUNDRY_LOCAL_ITEM_BYTES)).await
334    }
335
336    /// Get the async stream of transcription results.
337    ///
338    /// Results arrive as the native ASR engine processes audio data.
339    /// Can only be called once per session (the receiver is moved out).
340    pub async fn get_stream(&self) -> Result<LiveAudioTranscriptionStream> {
341        let mut state = self.state.lock().await;
342        let rx = state
343            .output_rx
344            .take()
345            .ok_or_else(|| FoundryLocalError::Validation {
346                reason: "No active streaming session, or stream already taken. \
347                         Call start() first and only call get_stream() once."
348                    .into(),
349            })?;
350        Ok(LiveAudioTranscriptionStream { rx })
351    }
352
353    /// Signal end-of-audio and stop the streaming session.
354    ///
355    /// Any remaining buffered audio is drained to the native engine first;
356    /// final results are delivered through the transcription stream before it
357    /// completes. The native stop always completes to avoid session leaks,
358    /// even if the provided [`CancellationToken`] fires.
359    pub async fn stop(&self, _ct: Option<CancellationToken>) -> Result<()> {
360        let worker = {
361            let mut state = self.state.lock().await;
362            if !state.started || state.stopped {
363                return Ok(());
364            }
365            state.stopped = true;
366            if let Some(queue) = &state.queue {
367                queue.mark_finished();
368            }
369            state.worker.take()
370        };
371
372        if let Some(handle) = worker {
373            let _ = handle.await;
374        }
375        Ok(())
376    }
377}
378
379impl Drop for LiveAudioTranscriptionSession {
380    /// Best-effort cleanup if the session is dropped without calling
381    /// [`stop`](Self::stop): mark the input queue finished so the blocking
382    /// background worker drains and returns (releasing the native session),
383    /// rather than leaking the worker thread. We hold `&mut self`, so the inner
384    /// state is reachable synchronously via `get_mut` without locking.
385    fn drop(&mut self) {
386        let state = self.state.get_mut();
387        if state.started && !state.stopped {
388            state.stopped = true;
389            if let Some(queue) = &state.queue {
390                queue.mark_finished();
391            }
392            // Detach the worker: marking the queue finished lets its blocking
393            // ProcessRequest return on its own, so we don't need to (and can't)
394            // await it here.
395            state.worker.take();
396        }
397    }
398}
399
400/// Streaming-callback context: forwards interim transcripts to the output channel.
401struct LiveCtx {
402    api: Arc<Api>,
403    tx: UnboundedSender<Result<LiveAudioTranscriptionResponse>>,
404}
405
406unsafe extern "C" fn live_trampoline(
407    data: flStreamingCallbackData,
408    user_data: *mut std::ffi::c_void,
409) -> c_int {
410    if user_data.is_null() {
411        return 0;
412    }
413    let result = catch_unwind(AssertUnwindSafe(|| {
414        let ctx = &*(user_data as *const LiveCtx);
415        let queue = data.item_queue;
416        if queue.is_null() {
417            return 0;
418        }
419        let item_api = ctx.api.item_api();
420        loop {
421            let mut item: *mut flItem = std::ptr::null_mut();
422            if !(item_api.ItemQueue_TryPop)(queue, &mut item) {
423                break;
424            }
425            if item.is_null() {
426                continue;
427            }
428            // The native audio path now streams SPEECH_SEGMENT items; older cores
429            // (and the OpenAI-JSON path) stream plain TEXT items. Handle both, and
430            // propagate a genuine read failure from either getter rather than
431            // silently dropping the item.
432            let response = (|| -> Result<Option<LiveAudioTranscriptionResponse>> {
433                if let Some(text) = read_text_item(&ctx.api, item)? {
434                    return Ok((!text.is_empty())
435                        .then(|| LiveAudioTranscriptionResponse::from_text(text, false)));
436                }
437
438                Ok(read_speech_segment(&ctx.api, item)?
439                    .filter(|seg| !seg.text.is_empty())
440                    .map(LiveAudioTranscriptionResponse::from_segment))
441            })();
442
443            (item_api.Item_Release)(item);
444
445            let response = match response {
446                Ok(response) => response,
447                Err(error) => {
448                    let _ = ctx.tx.send(Err(error));
449                    return 1; // read failure — stop streaming and surface the error
450                }
451            };
452
453            if let Some(response) = response {
454                if ctx.tx.send(Ok(response)).is_err() {
455                    return 1; // receiver dropped — cancel
456                }
457            }
458        }
459        0
460    }));
461    result.unwrap_or(1)
462}
463
464/// Blocking worker: builds the session/request, installs the streaming callback,
465/// processes the audio queue to completion, then emits the final transcript.
466fn run_worker(
467    model: NativeModel,
468    settings: LiveAudioTranscriptionOptions,
469    queue: Arc<NativeItemQueue>,
470    output_tx: UnboundedSender<Result<LiveAudioTranscriptionResponse>>,
471) {
472    let api = Arc::clone(&model.api);
473
474    let run = (|| -> Result<()> {
475        let session = NativeSession::create(&model)?;
476
477        // Serialise this session's install→process→uninstall critical section.
478        // The session is owned per-worker (un-shared), so the guard is
479        // uncontended, but taking it keeps the "every native session op runs
480        // under op_lock" invariant uniform across all streaming paths.
481        let guard = session.lock_ops();
482
483        let mut ctx = Box::new(LiveCtx {
484            api: Arc::clone(&api),
485            tx: output_tx.clone(),
486        });
487        let ctx_ptr = &mut *ctx as *mut LiveCtx as *mut std::ffi::c_void;
488        session.set_streaming_callback(Some(live_trampoline), ctx_ptr)?;
489
490        let request = NativeRequest::new(Arc::clone(&api))?;
491        let format = make_audio_item(
492            &api,
493            &[],
494            Some("pcm"),
495            settings.sample_rate as i32,
496            settings.channels as i32,
497        )?;
498        request.add_item(format, true)?;
499        // The input queue stays owned by us (append pushes into it).
500        request.add_item(queue.as_item_ptr(), false)?;
501
502        let response = session.process_request(&request);
503
504        // Always uninstall the streaming callback before `ctx` can be dropped —
505        // on the error path too — so the native session never retains a dangling
506        // `user_data` pointer into the freed context.
507        let _ = session.set_streaming_callback(None, std::ptr::null_mut());
508        drop(guard);
509        let response = response?;
510
511        // Aggregate the terminal transcript from the final response items. The
512        // native audio path returns a SPEECH_RESULT item; the OpenAI-JSON path
513        // (and older cores) return TEXT items.
514        let mut final_text = String::new();
515        for i in 0..response.item_count() {
516            let text = match response.item_text(i)? {
517                Some(text) => Some(text),
518                None => response.item_speech_result_text(i)?,
519            };
520
521            if let Some(text) = text {
522                final_text.push_str(&text);
523            }
524        }
525
526        drop(ctx);
527
528        if !final_text.is_empty() {
529            let _ = output_tx.send(Ok(LiveAudioTranscriptionResponse::from_text(
530                final_text, true,
531            )));
532        }
533        Ok(())
534    })();
535
536    if let Err(e) = run {
537        let _ = output_tx.send(Err(e));
538    }
539    // `queue` Arc clone and `session`/`request` drop here.
540    drop(queue);
541}
foundry_local_sdk/openai/live_audio_session.rs

foundry_local_sdk/openai/
live_audio_session.rs
