diff --git a/Cargo.lock b/Cargo.lock index dfea5d5c9c..c972c3c043 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1507,6 +1507,12 @@ dependencies = [ "dasp_sample", ] +[[package]] +name = "data-encoding" +version = "2.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4ae5f15dda3c708c0ade84bfee31ccab44a3da4f88015ed22f63732abe300c8" + [[package]] name = "dbus" version = "0.9.11" @@ -3615,6 +3621,7 @@ version = "0.4.0" dependencies = [ "anyhow", "async-trait", + "base64 0.22.1", "bytemuck", "bytes", "chrono", @@ -3673,6 +3680,7 @@ dependencies = [ "thiserror 2.0.18", "time", "tokio", + "tokio-tungstenite", "tokio-util", "tracing", "tracing-subscriber", @@ -7239,6 +7247,20 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edc5f74e248dc973e0dbb7b74c7e0d6fcc301c694ff50049504004ef4d0cdcd9" +dependencies = [ + "futures-util", + "log", + "native-tls", + "tokio", + "tokio-native-tls", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.18" @@ -7513,6 +7535,25 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.24.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18e5b8366ee7a95b16d32197d0b2604b43a0be89dc5fac9f8e96ccafbaedda8a" +dependencies = [ + "byteorder", + "bytes", + "data-encoding", + "http 1.4.0", + "httparse", + "log", + "native-tls", + "rand 0.8.6", + "sha1", + "thiserror 1.0.69", + "utf-8", +] + [[package]] name = "typeid" version = "1.0.3" diff --git a/README.md b/README.md index a62e670555..3b3527f21a 100644 --- a/README.md +++ b/README.md @@ -100,6 +100,7 @@ Whether you're a defense consultant, enterprise executive, legal professional, o - **Multi-Platform:** Works on macOS, Windows, and Linux. - **Open Source:** Meetily is open source and free to use. - **Flexible AI Provider Support:** Choose from Ollama (local), Claude, Groq, OpenRouter, or use your own OpenAI-compatible endpoint. +- **Self-Hosted Streaming Transcription (optional):** Point Meetily at your own realtime ASR websocket instead of the built-in engine — see [Streaming Transcription](docs/streaming-transcription.md). ## Installation diff --git a/docs/streaming-transcription.md b/docs/streaming-transcription.md new file mode 100644 index 0000000000..5040f516ef --- /dev/null +++ b/docs/streaming-transcription.md @@ -0,0 +1,157 @@ +# Streaming Transcription (self-hosted realtime ASR) + +Meetily's built-in transcription is **one-shot per chunk**: the audio pipeline +applies voice activity detection, cuts the speech into segments, and asks the +provider to transcribe each segment on its own. That is the right shape for +Whisper and Parakeet, and it is what every local provider uses. + +Some ASR models are built the other way around. They expect a **persistent +connection**, a continuous stream of audio, and they do their own segmentation +server-side — emitting partial hypotheses that get revised as more audio arrives. +Mistral's `Voxtral-Mini-Realtime` served by vLLM is the worked example. + +The **Custom Realtime (WebSocket)** provider covers that second shape. It is +entirely optional: if you don't configure it, nothing about local transcription +changes. + +> This connects Meetily to a server **you run**. Audio leaves the app over the +> network to whatever endpoint you configure, so point it at your own machine or +> your own infrastructure — not a third party you don't control. + +## Configuring it + +Settings → **Transcription** → provider **Custom Realtime (WebSocket)**: + +| Field | Meaning | +| --- | --- | +| **Endpoint** | `ws://host:port`, `wss://…`, or a full path. A bare host gets `/v1/realtime` appended; `http(s)://` is rewritten to `ws(s)://`. | +| **Model** | Model id the server should load, e.g. `voxtral-mini-transcribe-realtime-2602`. Validated server-side — a wrong name fails the connection test. | +| **API key** | Optional. Sent as `Authorization: Bearer …`. Leave empty for a server with no auth. | +| **Max session length** | Seconds of audio per server session before rolling over to a fresh one. Blank uses the 300 s default; **Detect** reads it from the endpoint; `0` never rolls over. See [Long meetings](#long-meetings-and-session-rollover). | + +**Test Connection** opens the socket and performs the handshake (which validates +the model) before you commit the settings, so a typo surfaces here rather than at +the start of a meeting. + +Language selection is disabled for this provider: the server detects the language +itself, the same way Parakeet does. + +## Running a Voxtral Realtime endpoint + +```bash +vllm serve mistralai/Voxtral-Mini-4B-Realtime-2602 \ + --tokenizer_mode mistral --config_format mistral --load_format mistral +``` + +Then point Meetily at `ws://localhost:8000`. Transcription delay is a +**server-side** knob (the model's `tekken.json`, 80–1200 ms in 80 ms steps; +Mistral recommends 480), not something the client negotiates per session. + +## What happens during a recording + +The pipeline gains a second tap that carries the **continuous, pre-VAD** mixed +audio — the same mix that gets recorded, before speech segmentation. That stream +is resampled to 16 kHz mono and pushed up the socket; the VAD chunk path is left +running but its output is discarded, since the server segments for itself. + +Results come back as a growing transcript. Meetily splits it into sentences: the +in-progress sentence updates in place as you speak, and settles into the +transcript when it completes. Everything is emitted on the same `transcript-update` +event the local providers use, so transcript history, persistence, reload +recovery, and summarization all work unchanged. + +**On a long meeting** the session is rolled over periodically so the server's +context can't fill — see [Long meetings](#long-meetings-and-session-rollover). + +**If the connection drops mid-recording**, the provider closes out the text it +had, reconnects with backoff (3 attempts), and carries on. Audio from the outage +window is discarded rather than replayed, so the live transcript doesn't fall +permanently behind. If all attempts fail you get an explicit error — transcription +stops for the rest of that recording, but **the recording itself keeps going**, so +the audio can be transcribed afterwards. + +## Long meetings and session rollover + +A realtime server holds an entire session in **one bounded context window** — +both the audio it has ingested and the transcript it has written. A meeting long +enough to fill that window doesn't produce an error; the session simply stops +emitting transcripts, and the rest of the meeting goes untranscribed. + +Meetily avoids this by giving each server session a fixed budget of audio. When +the budget is spent, the session is finalized and the recording continues on a +fresh one. The rollover is invisible in the transcript: the outgoing session's +final text lands before the new session takes over, so a three-hour meeting reads +the same as a three-minute one. + +**Max session length** controls that budget: + +- **Blank** — use the default of 300 seconds. Deliberately conservative: it fits + comfortably inside the 8k-token context of a typical self-hosted Voxtral-Mini + deployment, which in practice stops transcribing somewhere past 8 minutes. +- **Detect** — read `max_model_len` from the endpoint's `/v1/models` and derive a + session length from it, keeping 15% headroom. vLLM and several other + OpenAI-compatible servers publish this; ones that don't aren't an error, you + just set the value yourself. +- **A number** — seconds, at least 30. +- **`0`** — never roll over. One session for the entire recording. Only sensible + if your backend's context genuinely covers your longest meeting. + +The estimate behind Detect assumes Voxtral's encoder rate of one token per 80 ms +of audio (12.5 tokens/second), plus roughly 3.5 tokens/second for the transcript +the model writes. A different model with a different encoder rate will need the +value set by hand. + +Rollover is independent of reconnection: rolling over is *planned* and keeps the +transcript intact, whereas a reconnect is a *reaction* to a socket that died and +discards the audio from the outage window. + +## The wire contract + +The provider abstraction is **not** Voxtral-specific. A streaming provider only +has to satisfy the `StreamingTranscriptionProvider` trait +(`src-tauri/src/audio/transcription/streaming_provider.rs`): + +- accept 16 kHz mono `f32` audio frames pushed at it for the life of a recording +- emit `Partial { text }` for interim hypotheses and `Final { text, confidence }` + for settled ones — both carrying the **cumulative** text of the current utterance +- emit `Error { message, fatal }`, where `fatal` means transcription has ended +- verify itself on demand for the "Test Connection" button + +The persisted config carries a `protocol` discriminator (default +`voxtral-realtime`). Adding another dialect — Deepgram live, a Whisper streaming +server, your own — is a new module plus one match arm in +`build_streaming_provider`; no changes to the pipeline, worker, events, or UI. + +The `voxtral-realtime` dialect itself, verified against a live vLLM endpoint: + +```text + connect ws(s)://{host}/v1/realtime (wss:// adds Authorization: Bearer) + → client: {"type":"session.update","model":…} (model REQUIRED, flat) + → client: {"type":"input_audio_buffer.commit"} (REQUIRED — opens the buffer) + ← server: {"type":"session.created","id":…} (ignored; may arrive late) + → client: {"type":"input_audio_buffer.append","audio":""} (repeated) + → client: {"type":"input_audio_buffer.commit","final":true} (on finish) + ← server: {"type":"transcription.delta","delta":…} (INCREMENTAL, not cumulative) + ← server: {"type":"transcription.done","text":…} + ← server: {"type":"error","error":…} +``` + +Audio frames are base64-encoded 16 kHz mono PCM16-LE. Two things that are easy to +get wrong and cost real debugging time: + +- **The leading `commit` is required.** Omit it and the server ingests every + append but emits no delta and no `done` — the session just hangs silently. +- **`session.update` accepts only `model`**, top-level. Unknown fields are + silently ignored, so `language` and delay settings are *not* part of this + contract. + +## Troubleshooting + +| Symptom | Likely cause | +| --- | --- | +| Test Connection fails with `model_not_found` | Model id doesn't match what the server loaded. | +| Test Connection fails with a 502 | Endpoint reachable but the ASR backend is still booting. | +| Recording starts, transcript stays empty | Server accepted audio but never sent a delta — usually a dialect mismatch, not a Meetily bug. Check the server log. | +| "Reconnecting" warnings during a meeting | Server restarted or the network blipped; transcription resumes on its own. | +| Transcript stops partway through a long meeting and never resumes | The server's context filled. Lower **Max session length**, or press **Detect** to fit it to the endpoint. | +| Detect reports no limit | The endpoint doesn't publish `max_model_len`. Set the value by hand from what you know of the backend. | diff --git a/frontend/README.md b/frontend/README.md index bb8d36f892..f9b555c10d 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -137,6 +137,11 @@ For build and acceleration details, see: - [GPU Acceleration](../docs/GPU_ACCELERATION.md) - [Architecture](../docs/architecture.md) +Optionally, Meetily can transcribe against a self-hosted realtime ASR websocket +instead (e.g. vLLM serving Voxtral Realtime) — see +[Streaming Transcription](../docs/streaming-transcription.md). Local +transcription is unaffected if you don't configure it. + ## Development ### Frontend (Next.js) diff --git a/frontend/src-tauri/Cargo.toml b/frontend/src-tauri/Cargo.toml index b923d3c4ef..7e3f1546c0 100644 --- a/frontend/src-tauri/Cargo.toml +++ b/frontend/src-tauri/Cargo.toml @@ -111,6 +111,11 @@ tokio = { version = "1.32.0", features = ["full", "tracing"] } tokio-util = "0.7" # Utilities for tokio including CancellationToken async-trait = "0.1" # Trait abstraction for async methods +# Streaming (websocket) transcription — self-hosted realtime ASR (e.g. Voxtral). +# native-tls reuses the TLS stack reqwest already links (no second stack). +tokio-tungstenite = { version = "0.24", default-features = false, features = ["connect", "native-tls"] } +base64 = "0.22" + reqwest = { version = "0.11", features = ["blocking", "multipart", "json", "stream"] } # crossbeam diff --git a/frontend/src-tauri/migrations/20260720000000_add_custom_transcription_config.sql b/frontend/src-tauri/migrations/20260720000000_add_custom_transcription_config.sql new file mode 100644 index 0000000000..c309e68aca --- /dev/null +++ b/frontend/src-tauri/migrations/20260720000000_add_custom_transcription_config.sql @@ -0,0 +1,5 @@ +-- Migration: Add custom streaming transcription endpoint configuration +-- Stores JSON: {endpoint, apiKey, model, protocol, delayMs} +-- Used to connect meetily to a self-hosted realtime ASR websocket server +-- (e.g. vLLM serving Voxtral-Mini-Realtime) as a "custom" transcription provider. +ALTER TABLE transcript_settings ADD COLUMN customTranscriptionConfig TEXT; diff --git a/frontend/src-tauri/src/api/api.rs b/frontend/src-tauri/src/api/api.rs index e7f7c5db0c..273feb89b0 100644 --- a/frontend/src-tauri/src/api/api.rs +++ b/frontend/src-tauri/src/api/api.rs @@ -713,6 +713,176 @@ pub async fn api_get_transcript_api_key( } } +#[tauri::command] +pub async fn api_get_custom_transcription_config( + _app: AppHandle, + state: tauri::State<'_, AppState>, + _auth_token: Option, +) -> Result, String> { + log_info!("api_get_custom_transcription_config called (native)"); + match SettingsRepository::get_custom_transcription_config(state.db_manager.pool()).await { + Ok(config) => Ok(config), + Err(e) => { + log_error!("Failed to get custom transcription config: {}", e); + Err(e.to_string()) + } + } +} + +#[tauri::command] +pub async fn api_save_custom_transcription_config( + _app: AppHandle, + state: tauri::State<'_, AppState>, + endpoint: String, + model: String, + api_key: Option, + protocol: Option, + delay_ms: Option, + max_session_seconds: Option, + _auth_token: Option, +) -> Result { + log_info!("api_save_custom_transcription_config called (native)"); + + let endpoint = endpoint.trim().to_string(); + if endpoint.is_empty() { + return Err("Endpoint URL cannot be empty".to_string()); + } + if !(endpoint.starts_with("ws://") + || endpoint.starts_with("wss://") + || endpoint.starts_with("http://") + || endpoint.starts_with("https://")) + { + return Err("Endpoint must start with ws://, wss://, http:// or https://".to_string()); + } + if model.trim().is_empty() { + return Err("Model cannot be empty".to_string()); + } + // 0 is meaningful (never roll over); anything positive must be long enough to + // be worth a session. + if matches!(max_session_seconds, Some(secs) if secs > 0 && secs < 30) { + return Err("Max session length must be at least 30 seconds (or 0 to disable)".to_string()); + } + + let config = crate::audio::transcription::CustomTranscriptionConfig { + endpoint, + api_key: api_key.filter(|k| !k.is_empty()), + model: model.trim().to_string(), + protocol: protocol.unwrap_or_else(|| "voxtral-realtime".to_string()), + delay_ms, + max_session_seconds, + }; + + match SettingsRepository::save_custom_transcription_config(state.db_manager.pool(), &config) + .await + { + Ok(()) => { + log_info!("Successfully saved custom transcription configuration."); + Ok(serde_json::json!({ "status": "success", "message": "Custom transcription configuration saved successfully" })) + } + Err(e) => { + log_error!("Failed to save custom transcription config: {}", e); + Err(e.to_string()) + } + } +} + +#[tauri::command] +pub async fn api_test_custom_transcription_connection( + _app: AppHandle, + endpoint: String, + model: String, + api_key: Option, + protocol: Option, + delay_ms: Option, +) -> Result { + log_info!( + "api_test_custom_transcription_connection called: endpoint='{}', model='{}'", + &endpoint, + &model + ); + + let endpoint = endpoint.trim().to_string(); + if !(endpoint.starts_with("ws://") + || endpoint.starts_with("wss://") + || endpoint.starts_with("http://") + || endpoint.starts_with("https://")) + { + return Err("Endpoint must start with ws://, wss://, http:// or https://".to_string()); + } + if model.trim().is_empty() { + return Err("Model cannot be empty".to_string()); + } + + let config = crate::audio::transcription::CustomTranscriptionConfig { + endpoint, + api_key: api_key.filter(|k| !k.trim().is_empty()), + model: model.trim().to_string(), + protocol: protocol.unwrap_or_else(|| "voxtral-realtime".to_string()), + delay_ms, + max_session_seconds: None, + }; + + let provider = crate::audio::transcription::build_streaming_provider(config) + .map_err(|e| e.to_string())?; + + match provider.test_connection().await { + Ok(()) => { + log_info!("Custom transcription connection test succeeded."); + Ok(serde_json::json!({ + "status": "success", + "message": "Connected to the realtime transcription endpoint successfully" + })) + } + Err(e) => { + log_error!("Custom transcription connection test failed: {}", e); + Err(format!("Connection test failed: {}", e)) + } + } +} + +/// Ask a realtime transcription endpoint how much audio it can hold in one +/// session, so the settings UI can propose a session length instead of making the +/// user guess. Servers that announce nothing are reported as such, not as an error. +#[tauri::command] +pub async fn api_detect_custom_transcription_limits( + _app: AppHandle, + endpoint: String, + model: String, + api_key: Option, + protocol: Option, +) -> Result { + log_info!( + "api_detect_custom_transcription_limits called: endpoint='{}', model='{}'", + &endpoint, + &model + ); + + let endpoint = endpoint.trim().to_string(); + if !(endpoint.starts_with("ws://") + || endpoint.starts_with("wss://") + || endpoint.starts_with("http://") + || endpoint.starts_with("https://")) + { + return Err("Endpoint must start with ws://, wss://, http:// or https://".to_string()); + } + if model.trim().is_empty() { + return Err("Model cannot be empty".to_string()); + } + + let config = crate::audio::transcription::CustomTranscriptionConfig { + endpoint, + api_key: api_key.filter(|k| !k.trim().is_empty()), + model: model.trim().to_string(), + protocol: protocol.unwrap_or_else(|| "voxtral-realtime".to_string()), + delay_ms: None, + max_session_seconds: None, + }; + + crate::audio::transcription::detect_session_limit(&config) + .await + .map_err(|e| format!("Could not read the endpoint's limits: {}", e)) +} + #[tauri::command] pub async fn api_delete_api_key( _app: AppHandle, diff --git a/frontend/src-tauri/src/audio/pipeline.rs b/frontend/src-tauri/src/audio/pipeline.rs index cd344ddb57..91ab0fdc72 100644 --- a/frontend/src-tauri/src/audio/pipeline.rs +++ b/frontend/src-tauri/src/audio/pipeline.rs @@ -694,6 +694,9 @@ pub struct AudioPipeline { mixer: ProfessionalAudioMixer, // Recording sender for pre-mixed audio recording_sender_for_mixed: Option>, + // Streaming sender for pre-mixed audio (continuous, pre-VAD): feeds a + // realtime transcription provider. `None` unless a streaming provider is active. + streaming_sender_for_mixed: Option>, } impl AudioPipeline { @@ -760,6 +763,7 @@ impl AudioPipeline { ring_buffer, mixer, recording_sender_for_mixed: None, // Will be set by manager + streaming_sender_for_mixed: None, // Will be set by manager when streaming is active } } @@ -876,6 +880,20 @@ impl AudioPipeline { }; let _ = sender.send(recording_chunk); } + + // STEP 5: Send continuous (pre-VAD) mixed audio to a streaming + // transcription provider, when one is active. The provider does + // its own segmentation, so this bypasses VAD entirely. + if let Some(ref sender) = self.streaming_sender_for_mixed { + let streaming_chunk = AudioChunk { + data: mixed_with_gain.clone(), + sample_rate: self.sample_rate, + timestamp: chunk.timestamp, + chunk_id: self.chunk_id_counter, + device_type: DeviceType::Microphone, // Mixed audio + }; + let _ = sender.send(streaming_chunk); + } } } } @@ -962,6 +980,7 @@ impl AudioPipelineManager { target_chunk_duration_ms: u32, sample_rate: u32, recording_sender: Option>, + streaming_sender: Option>, mic_device_name: String, mic_device_kind: super::device_detection::InputDeviceKind, system_device_name: String, @@ -994,6 +1013,7 @@ impl AudioPipelineManager { // CRITICAL FIX: Connect recording sender to receive pre-mixed audio // This ensures both mic AND system audio are captured in recordings pipeline.recording_sender_for_mixed = recording_sender; + pipeline.streaming_sender_for_mixed = streaming_sender; let handle = tokio::spawn(async move { pipeline.run().await diff --git a/frontend/src-tauri/src/audio/recording_commands.rs b/frontend/src-tauri/src/audio/recording_commands.rs index 31061e6f6d..d5f6785d19 100644 --- a/frontend/src-tauri/src/audio/recording_commands.rs +++ b/frontend/src-tauri/src/audio/recording_commands.rs @@ -11,6 +11,7 @@ use std::sync::{ Arc, Mutex, }; use tauri::{AppHandle, Emitter, Manager, Runtime}; +use tokio::sync::mpsc; use tokio::task::JoinHandle; use super::{ @@ -61,6 +62,71 @@ pub struct TranscriptionStatus { pub last_activity_ms: u64, } +// ============================================================================ +// TRANSCRIPTION ROUTING (chunk vs streaming) +// ============================================================================ + +/// Resolve the active streaming transcription provider, if the configured +/// transcript provider is a custom streaming (websocket) endpoint. Returns +/// `None` for the standard local (Whisper/Parakeet) chunk providers. +async fn resolve_streaming_provider( + app: &AppHandle, +) -> Option> { + let config = crate::api::api::api_get_transcript_config(app.clone(), app.clone().state(), None) + .await + .ok() + .flatten()?; + + if config.provider != transcription::CUSTOM_STREAMING_PROVIDER { + return None; + } + + let custom = + crate::api::api::api_get_custom_transcription_config(app.clone(), app.clone().state(), None) + .await + .ok() + .flatten()?; + + match transcription::build_streaming_provider(custom) { + Ok(provider) => Some(provider), + Err(e) => { + warn!("Failed to build streaming transcription provider: {}", e); + None + } + } +} + +/// Spawn the transcription workers and store the handle in `TRANSCRIPTION_TASK`. +/// +/// When a streaming provider is active it drives the continuous tap; the VAD +/// chunk channel is drained (not transcribed) so the pipeline's unbounded sender +/// cannot accumulate for the whole recording. Otherwise the standard parallel +/// chunk worker runs as before. +fn spawn_transcription( + app: &AppHandle, + transcription_receiver: mpsc::UnboundedReceiver, + streaming: Option<( + Arc, + mpsc::UnboundedReceiver, + )>, +) { + let handle = match streaming { + Some((provider, streaming_receiver)) => { + info!( + "🌊 Streaming transcription active via '{}' — VAD chunk path disabled", + provider.provider_name() + ); + // The pipeline still produces VAD chunks; drain + discard them. + let mut chunk_rx = transcription_receiver; + tokio::spawn(async move { while chunk_rx.recv().await.is_some() {} }); + transcription::run_streaming_session(app.clone(), streaming_receiver, provider) + } + None => transcription::start_transcription_task(app.clone(), transcription_receiver), + }; + let mut global_task = TRANSCRIPTION_TASK.lock().unwrap(); + *global_task = Some(handle); +} + // ============================================================================ // RECORDING COMMANDS // ============================================================================ @@ -232,9 +298,19 @@ pub async fn start_recording_with_meeting_name( let _ = app_for_error.emit("recording-error", error.user_message()); }); + // Resolve a streaming transcription provider (if configured) and create the + // continuous pre-VAD tap it will consume. + let streaming_provider = resolve_streaming_provider(&app).await; + let (streaming_sender, streaming_receiver) = if streaming_provider.is_some() { + let (tx, rx) = mpsc::unbounded_channel::(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + // Start recording with resolved devices (replaces start_recording_with_defaults_and_auto_save call) let transcription_receiver = manager - .start_recording(microphone_device, system_device, auto_save) + .start_recording(microphone_device, system_device, auto_save, streaming_sender) .await .map_err(|e| format!("Failed to start recording: {}", e))?; @@ -250,12 +326,8 @@ pub async fn start_recording_with_meeting_name( drop(engine_lifecycle_guard); reset_speech_detected_flag(); // Reset for new recording session - // Start optimized parallel transcription task and store handle - let task_handle = transcription::start_transcription_task(app.clone(), transcription_receiver); - { - let mut global_task = TRANSCRIPTION_TASK.lock().unwrap(); - *global_task = Some(task_handle); - } + // Start transcription: streaming provider (if configured) or the parallel chunk worker. + spawn_transcription(&app, transcription_receiver, streaming_provider.zip(streaming_receiver)); // CRITICAL: Listen for transcript-update events and save to recording manager // This enables transcript history persistence for page reload sync @@ -403,9 +475,19 @@ pub async fn start_recording_with_devices_and_meeting( let _ = app_for_error.emit("recording-error", error.user_message()); }); + // Resolve a streaming transcription provider (if configured) and create the + // continuous pre-VAD tap it will consume. + let streaming_provider = resolve_streaming_provider(&app).await; + let (streaming_sender, streaming_receiver) = if streaming_provider.is_some() { + let (tx, rx) = mpsc::unbounded_channel::(); + (Some(tx), Some(rx)) + } else { + (None, None) + }; + // Start recording with specified devices and auto_save setting let transcription_receiver = manager - .start_recording(mic_device, system_device, auto_save) + .start_recording(mic_device, system_device, auto_save, streaming_sender) .await .map_err(|e| format!("Failed to start recording: {}", e))?; @@ -421,12 +503,8 @@ pub async fn start_recording_with_devices_and_meeting( drop(engine_lifecycle_guard); reset_speech_detected_flag(); // Reset for new recording session - // Start optimized parallel transcription task and store handle - let task_handle = transcription::start_transcription_task(app.clone(), transcription_receiver); - { - let mut global_task = TRANSCRIPTION_TASK.lock().unwrap(); - *global_task = Some(task_handle); - } + // Start transcription: streaming provider (if configured) or the parallel chunk worker. + spawn_transcription(&app, transcription_receiver, streaming_provider.zip(streaming_receiver)); // CRITICAL: Listen for transcript-update events and save to recording manager // This enables transcript history persistence for page reload sync diff --git a/frontend/src-tauri/src/audio/recording_manager.rs b/frontend/src-tauri/src/audio/recording_manager.rs index ce9b95e750..106c25220b 100644 --- a/frontend/src-tauri/src/audio/recording_manager.rs +++ b/frontend/src-tauri/src/audio/recording_manager.rs @@ -65,6 +65,7 @@ impl RecordingManager { microphone_device: Option>, system_device: Option>, auto_save: bool, + streaming_sender: Option>, ) -> Result> { info!("Starting recording manager (auto_save: {})", auto_save); @@ -112,6 +113,7 @@ impl RecordingManager { 0, // Ignored - using dynamic sizing internally 48000, // 48kHz sample rate Some(recording_sender), // CRITICAL: Pass recording sender to receive pre-mixed audio + streaming_sender, // Optional: continuous pre-VAD tap for a streaming provider mic_name, mic_kind, sys_name, @@ -186,7 +188,7 @@ impl RecordingManager { } // Start recording with selected devices and auto_save setting - self.start_recording(microphone_device, system_device, auto_save).await + self.start_recording(microphone_device, system_device, auto_save, None).await } #[cfg(not(target_os = "macos"))] @@ -221,7 +223,7 @@ impl RecordingManager { return Err(anyhow::anyhow!("No microphone device available")); } - self.start_recording(microphone_device, system_device, auto_save).await + self.start_recording(microphone_device, system_device, auto_save, None).await } } diff --git a/frontend/src-tauri/src/audio/transcription/engine.rs b/frontend/src-tauri/src/audio/transcription/engine.rs index 415cdc1f1d..c29192589e 100644 --- a/frontend/src-tauri/src/audio/transcription/engine.rs +++ b/frontend/src-tauri/src/audio/transcription/engine.rs @@ -135,6 +135,36 @@ pub async fn validate_transcription_model_ready(app: &AppHandle) } } } + p if p == super::CUSTOM_STREAMING_PROVIDER => { + info!("🔍 Validating custom streaming transcription endpoint..."); + let custom = crate::api::api::api_get_custom_transcription_config( + app.clone(), + app.clone().state(), + None, + ) + .await + .map_err(|e| format!("Failed to load streaming transcription config: {}", e))? + .ok_or_else(|| { + "No custom streaming transcription endpoint is configured. Please set one in the transcription settings.".to_string() + })?; + + let provider = super::build_streaming_provider(custom) + .map_err(|e| format!("Invalid streaming transcription config: {}", e))?; + + match provider.test_connection().await { + Ok(()) => { + info!("✅ Streaming transcription endpoint reachable"); + Ok(()) + } + Err(e) => { + warn!("❌ Streaming transcription endpoint validation failed: {}", e); + Err(format!( + "Could not reach the realtime transcription endpoint: {}", + e + )) + } + } + } other => { warn!("❌ Unsupported transcription provider for local recording: {}", other); Err(format!( diff --git a/frontend/src-tauri/src/audio/transcription/mod.rs b/frontend/src-tauri/src/audio/transcription/mod.rs index 10bd7ca23d..5e5fa234bd 100644 --- a/frontend/src-tauri/src/audio/transcription/mod.rs +++ b/frontend/src-tauri/src/audio/transcription/mod.rs @@ -2,11 +2,73 @@ // // Transcription module: Provider abstraction, engine management, and worker pool. +use serde::{Deserialize, Serialize}; + pub mod provider; pub mod whisper_provider; pub mod parakeet_provider; pub mod engine; pub mod worker; +pub mod streaming_provider; +pub mod voxtral_realtime; +pub mod streaming_worker; + +/// Provider identifier for the custom streaming (websocket) transcription backend. +pub const CUSTOM_STREAMING_PROVIDER: &str = "customStreaming"; + +/// Configuration for a self-hosted realtime transcription websocket endpoint. +/// +/// Stored as JSON in the `transcript_settings.customTranscriptionConfig` column and +/// used to connect to an OpenAI/Voxtral-compatible realtime ASR server (e.g. vLLM +/// serving `Voxtral-Mini-Realtime`). Mirrors the summary-side `CustomOpenAIConfig`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct CustomTranscriptionConfig { + /// Base URL of the websocket endpoint (e.g. "ws://localhost:8000" or a full path). + pub endpoint: String, + /// API key for authentication (optional if the server doesn't require it). + #[serde(rename = "apiKey")] + pub api_key: Option, + /// Model identifier to request (e.g. "voxtral-mini-transcribe-realtime-2602"). + pub model: String, + /// Streaming protocol dialect. Defaults to "voxtral-realtime". + #[serde(default = "default_streaming_protocol")] + pub protocol: String, + /// Requested transcription delay in milliseconds (protocol-specific, optional). + #[serde(rename = "delayMs")] + pub delay_ms: Option, + /// Longest stretch of audio, in seconds, to feed to a single server session + /// before rolling over to a fresh one. + /// + /// A realtime ASR server holds the whole session in one bounded context; once + /// that fills, the session stops producing transcripts for the rest of the + /// recording. Rolling over on a schedule keeps a multi-hour meeting inside + /// whatever the backend can actually take. `None` uses + /// [`DEFAULT_MAX_SESSION_SECONDS`]; `Some(0)` disables rollover entirely (one + /// session for the whole recording — the pre-0.4 behaviour). + #[serde(rename = "maxSessionSeconds", default)] + pub max_session_seconds: Option, +} + +/// Rollover interval used when the user hasn't set one and the endpoint didn't +/// announce a context size. Conservative on purpose: it comfortably fits the +/// 8k-token context typical of a self-hosted Voxtral-Mini deployment, which in +/// practice dies somewhere past the 8-minute mark. +pub const DEFAULT_MAX_SESSION_SECONDS: u32 = 300; + +impl CustomTranscriptionConfig { + /// Seconds of audio per server session, or `None` when rollover is disabled. + pub fn session_limit_seconds(&self) -> Option { + match self.max_session_seconds { + Some(0) => None, + Some(secs) => Some(secs), + None => Some(DEFAULT_MAX_SESSION_SECONDS), + } + } +} + +fn default_streaming_protocol() -> String { + "voxtral-realtime".to_string() +} // Re-export commonly used types pub use provider::{TranscriptionError, TranscriptionProvider, TranscriptResult}; @@ -23,3 +85,11 @@ pub use worker::{ reset_speech_detected_flag, TranscriptUpdate }; +pub use streaming_provider::{ + build_streaming_provider, + StreamSession, + StreamTranscriptEvent, + StreamingTranscriptionProvider, +}; +pub use streaming_worker::run_streaming_session; +pub use voxtral_realtime::{detect_session_limit, DetectedSessionLimit}; diff --git a/frontend/src-tauri/src/audio/transcription/streaming_provider.rs b/frontend/src-tauri/src/audio/transcription/streaming_provider.rs new file mode 100644 index 0000000000..5ac6035ae7 --- /dev/null +++ b/frontend/src-tauri/src/audio/transcription/streaming_provider.rs @@ -0,0 +1,99 @@ +// audio/transcription/streaming_provider.rs +// +// Streaming transcription abstraction — a sibling to the one-shot +// `TranscriptionProvider` trait for providers that hold a persistent connection +// and stream partial + final results as audio flows in (e.g. a self-hosted +// Voxtral-realtime websocket served by vLLM). +// +// The one-shot trait is request/response per VAD-gated speech segment +// (`provider.rs`). A realtime server is the opposite: one long-lived session per +// recording, continuous PCM pushed up, partial/final transcript events streamed +// back. That doesn't fit the one-shot seam, so streaming providers live behind +// this separate trait and are driven by the streaming worker rather than the +// chunk worker. + +use std::sync::Arc; + +use async_trait::async_trait; +use tokio::sync::mpsc; +use tokio_util::sync::CancellationToken; + +use super::provider::TranscriptionError; +use super::CustomTranscriptionConfig; + +/// A transcript event streamed from a realtime provider. +#[derive(Debug, Clone)] +pub enum StreamTranscriptEvent { + /// Interim hypothesis — may be revised by later partials or superseded by the + /// final for the same segment. + Partial { text: String }, + /// Stable transcript for a completed segment. `confidence` is `None` when the + /// provider does not report per-segment confidence (Voxtral-realtime does not). + Final { + text: String, + confidence: Option, + }, + /// An error surfaced from inside the session. + /// + /// `fatal` distinguishes a transient hiccup the provider is still recovering + /// from (a dropped socket it is reconnecting to) from a terminal one that has + /// ended transcription for the rest of the recording. The worker surfaces the + /// former as a warning and the latter as an actionable `transcription-error`, + /// so a session that dies mid-recording can never fail silently. + Error { message: String, fatal: bool }, +} + +/// Handle to a live streaming transcription session. +pub struct StreamSession { + /// Push 16 kHz mono f32 audio frames to the provider. Dropping this sender + /// (closing the channel) signals end-of-audio: the provider flushes the tail, + /// requests the final transcript, and then the worker task exits. + pub audio_tx: mpsc::UnboundedSender>, + /// Hard-cancel the session — closes the socket and ends the worker task + /// without waiting for a final flush. Prefer dropping `audio_tx` for a clean + /// stop; use this to abort. + pub cancel: CancellationToken, +} + +/// A transcription provider that streams results over a persistent connection. +#[async_trait] +pub trait StreamingTranscriptionProvider: Send + Sync { + /// Open a streaming session. Audio pushed to the returned + /// [`StreamSession::audio_tx`] (16 kHz mono f32) is transcribed and results + /// delivered on `events` until the audio channel is closed, the session is + /// cancelled, or the remote closes the connection. + /// + /// `language` is a best-effort hint; providers that auto-detect (Voxtral) may + /// ignore it. + async fn start_session( + &self, + language: Option, + events: mpsc::UnboundedSender, + ) -> Result; + + /// Verify the endpoint is reachable and correctly configured. Connects, + /// performs the protocol handshake (which validates the model server-side), + /// and disconnects. Used by the "Test Connection" settings command. + async fn test_connection(&self) -> Result<(), TranscriptionError>; + + /// Provider name for logging/debugging. + fn provider_name(&self) -> &'static str; +} + +/// Build a streaming provider from persisted config, dispatching on `protocol`. +/// +/// New websocket dialects (Deepgram live, etc.) plug in here without touching the +/// rest of the streaming plumbing. +pub fn build_streaming_provider( + config: CustomTranscriptionConfig, +) -> Result, TranscriptionError> { + match config.protocol.as_str() { + "voxtral-realtime" | "" => Ok(Arc::new( + super::voxtral_realtime::VoxtralRealtimeProvider::new(config), + )), + other => Err(TranscriptionError::EngineFailed(format!( + "Unknown streaming transcription protocol '{}'. Supported: voxtral-realtime.", + other + ))), + } +} diff --git a/frontend/src-tauri/src/audio/transcription/streaming_worker.rs b/frontend/src-tauri/src/audio/transcription/streaming_worker.rs new file mode 100644 index 0000000000..ec5d5e7907 --- /dev/null +++ b/frontend/src-tauri/src/audio/transcription/streaming_worker.rs @@ -0,0 +1,410 @@ +// audio/transcription/streaming_worker.rs +// +// Drives a live streaming transcription session for the duration of a recording. +// +// Unlike the chunk worker (`worker.rs`), which VAD-gates audio and calls a +// one-shot `transcribe()` per segment, this runner holds a persistent provider +// session: it pushes the continuous (pre-VAD) mixed stream up and forwards the +// provider's partial/final events onto the SAME `transcript-update` event the +// chunk worker uses — so history persistence and the UI work unchanged. +// +// Lifecycle / clean shutdown is by channel drop-chain: +// pipeline stops → `streaming_receiver` closes → audio task drops `audio_tx` +// → provider flushes the tail + requests the final transcript → provider drops +// the events sender → the event loop ends → this task's JoinHandle completes. +// `stop_recording` awaits that handle (stored in `TRANSCRIPTION_TASK`), so the +// final transcript is guaranteed to land before the recording is finalized. + +use std::sync::Arc; +use std::time::Instant; + +use log::{error, info, warn}; +use tauri::{AppHandle, Emitter, Runtime}; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; + +use super::streaming_provider::{StreamTranscriptEvent, StreamingTranscriptionProvider}; +use super::worker::TranscriptUpdate; +use crate::audio::AudioChunk; + +/// Sample rate the streaming providers expect (16 kHz mono). +const STREAMING_SAMPLE_RATE: u32 = 16000; + +/// Start a streaming transcription session. Returns a handle that completes once +/// the provider session has fully finalized (final transcript delivered). +/// +/// * `streaming_receiver` — continuous mixed audio from the pipeline tap (48 kHz). +/// * `provider` — the configured streaming provider (e.g. Voxtral realtime). +pub fn run_streaming_session( + app: AppHandle, + streaming_receiver: mpsc::UnboundedReceiver, + provider: Arc, +) -> JoinHandle<()> { + tokio::spawn(async move { + info!( + "🌊 Starting streaming transcription session via '{}'", + provider.provider_name() + ); + + // Open the session. `language: None` — Voxtral auto-detects; other + // providers may honor a hint later. + let (events_tx, events_rx) = mpsc::unbounded_channel::(); + let session = match provider.start_session(None, events_tx).await { + Ok(s) => s, + Err(e) => { + error!("Failed to start streaming transcription session: {}", e); + let _ = app.emit( + "transcription-error", + serde_json::json!({ + "error": e.to_string(), + "userMessage": "Recording failed: could not connect to the realtime transcription endpoint. Please check the endpoint in transcription settings.", + "actionable": true, + }), + ); + // Drain the audio tap so the pipeline's unbounded channel can't grow. + drain_audio(streaming_receiver).await; + return; + } + }; + + // Task A: resample the 48 kHz mixed tap to 16 kHz mono and push it up. + let audio_task = tokio::spawn(feed_audio(streaming_receiver, session.audio_tx)); + + // Task B (this task): forward provider events onto `transcript-update`. + forward_events(&app, events_rx).await; + + // The event loop ends when the provider drops its events sender (session + // fully closed). Join the audio task for a clean teardown. + let _ = audio_task.await; + info!("🌊 Streaming transcription session finished"); + }) +} + +/// Resample the continuous mixed stream to 16 kHz mono and forward it to the +/// provider until the tap closes (end of recording), then drop `audio_tx` to +/// signal end-of-audio. +async fn feed_audio( + mut streaming_receiver: mpsc::UnboundedReceiver, + audio_tx: mpsc::UnboundedSender>, +) { + while let Some(chunk) = streaming_receiver.recv().await { + let samples = if chunk.sample_rate != STREAMING_SAMPLE_RATE { + crate::audio::audio_processing::resample_audio( + &chunk.data, + chunk.sample_rate, + STREAMING_SAMPLE_RATE, + ) + } else { + chunk.data + }; + if samples.is_empty() { + continue; + } + if audio_tx.send(samples).is_err() { + // Provider worker gone — nothing more we can do. + warn!("Streaming provider closed the audio channel; stopping audio feed"); + break; + } + } + // `audio_tx` dropped here → provider flushes the tail and finalizes. +} + +/// Consume provider events and emit `transcript-update`s. +/// +/// The provider streams a single, ever-growing cumulative text (partials) plus a +/// final. We break it into sentences: the **in-progress** sentence is emitted as +/// a live partial that keeps updating under one `sequence_id` (word-for-word), +/// and when it completes (sentence-final punctuation, soft cap, or end of stream) +/// it is re-emitted as final under that same id and the next sentence takes the +/// next id. The live UI upserts by `sequence_id`, so a segment grows in place and +/// then locks — like a realtime caption that settles into the transcript. +async fn forward_events( + app: &AppHandle, + mut events_rx: mpsc::UnboundedReceiver, +) { + let session_start = Instant::now(); + let mut segmenter = Segmenter::new(); + // Recording-relative start time of the sentence currently being built. + let mut segment_start: f64 = 0.0; + let mut speech_announced = false; + + while let Some(ev) = events_rx.recv().await { + let (text, is_final) = match ev { + StreamTranscriptEvent::Partial { text } => (text, false), + StreamTranscriptEvent::Final { text, .. } => (text, true), + StreamTranscriptEvent::Error { message, fatal } => { + if fatal { + // Transcription is over for this recording. `transcription-error` + // is the event the UI actually surfaces to the user (see + // useModalState); a `transcription-warning` would be swallowed. + error!("Streaming transcription failed: {}", message); + let _ = app.emit( + "transcription-error", + serde_json::json!({ + "error": message, + "userMessage": message, + "actionable": true, + }), + ); + } else { + warn!("Streaming transcription warning: {}", message); + let _ = app.emit( + "transcription-warning", + serde_json::json!({ "error": message }), + ); + } + continue; + } + }; + + for e in segmenter.advance(&text, is_final) { + announce_speech(app, &mut speech_announced); + let now = session_start.elapsed().as_secs_f64(); + if !e.is_partial { + info!("🌊 streaming segment {}: {}", e.seq, e.text); + } + emit_update(app, e.text, e.is_partial, 1.0, e.seq, segment_start, now); + // A finalized sentence closes its block; the next one starts now. + if !e.is_partial { + segment_start = now; + } + } + + // A final closes the utterance; the next one starts fresh cumulative text. + if is_final { + segmenter.reset(); + } + } +} + +/// Soft cap: finalize an in-progress run of speech that hasn't hit sentence-ending +/// punctuation yet, so spontaneous speech without periods still settles into the +/// transcript instead of growing unboundedly. +const SOFT_FLUSH_CHARS: usize = 200; + +/// One emission destined for a `transcript-update`: a `sequence_id`, its text, and +/// whether it's still provisional (`is_partial`) or the settled sentence. +#[derive(Debug, PartialEq)] +struct Emission { + seq: u64, + text: String, + is_partial: bool, +} + +/// Turns a provider's growing cumulative transcript into per-sentence emissions. +/// +/// `committed` is how far into the cumulative text has already been *finalized*; +/// `seq` is the id of the sentence currently being built. On each update, any +/// sentences that completed (at `.?!…`, or a word boundary past +/// [`SOFT_FLUSH_CHARS`]) are emitted as finals (incrementing `seq`), and the +/// remaining tail is emitted as a live partial under the current `seq`. +struct Segmenter { + committed: usize, + seq: u64, +} + +impl Segmenter { + fn new() -> Self { + Self { committed: 0, seq: 0 } + } + + fn advance(&mut self, text: &str, is_final: bool) -> Vec { + // Cumulative text only grows within an utterance; if it shrank (a new + // utterance began), restart the byte offset. `seq` keeps climbing so ids + // stay globally unique. + if self.committed > text.len() { + self.committed = 0; + } + let mut out = Vec::new(); + + // Finalize every sentence that has completed. + while let Some(b) = next_break(&text[self.committed..]) { + let seg = text[self.committed..self.committed + b].trim(); + if !seg.is_empty() { + out.push(Emission { seq: self.seq, text: seg.to_string(), is_partial: false }); + self.seq += 1; + } + self.committed += b; + } + + // The unfinished remainder: a live partial mid-stream, or a final flush. + let tail = text[self.committed..].trim(); + if !tail.is_empty() { + out.push(Emission { seq: self.seq, text: tail.to_string(), is_partial: !is_final }); + if is_final { + self.seq += 1; + } + } + if is_final { + self.committed = text.len(); + } + out + } + + /// End of an utterance: the next one restarts the byte offset (ids keep going). + fn reset(&mut self) { + self.committed = 0; + } +} + +/// Byte offset (within `tail`) to cut the next segment: right after the first +/// sentence-final punctuation, or—if the tail is longer than [`SOFT_FLUSH_CHARS`] +/// with none—at the last word boundary within that window. `None` if the tail +/// isn't ready to flush yet. The returned offset is always ≥ 1 and lands on a +/// UTF-8 char boundary (cuts are at ASCII punctuation or spaces). +fn next_break(tail: &str) -> Option { + for (i, c) in tail.char_indices() { + if matches!(c, '.' | '!' | '?' | '…') { + return Some(i + c.len_utf8()); + } + } + if tail.len() > SOFT_FLUSH_CHARS { + if let Some(sp) = tail[..SOFT_FLUSH_CHARS].rfind(' ') { + if sp > 0 { + return Some(sp + 1); + } + } + return Some(tail.len()); + } + None +} + +/// Emit the first `speech-detected` event of the session (UI feedback), once. +fn announce_speech(app: &AppHandle, announced: &mut bool) { + if *announced { + return; + } + *announced = true; + let _ = app.emit( + "speech-detected", + serde_json::json!({ "message": "Speech activity detected" }), + ); +} + +/// Build and emit a `transcript-update` matching the chunk worker's shape. +fn emit_update( + app: &AppHandle, + text: String, + is_partial: bool, + confidence: f32, + sequence_id: u64, + audio_start_time: f64, + audio_end_time: f64, +) { + let update = TranscriptUpdate { + text, + timestamp: format_current_timestamp(), + source: "Audio".to_string(), + sequence_id, + chunk_start_time: audio_start_time, + is_partial, + confidence, + audio_start_time, + audio_end_time, + duration: (audio_end_time - audio_start_time).max(0.0), + }; + if let Err(e) = app.emit("transcript-update", &update) { + error!("Failed to emit streaming transcript update: {}", e); + } +} + +/// Drain and discard the audio tap (used when the session failed to start), so +/// the pipeline's unbounded sender does not accumulate for the whole recording. +async fn drain_audio(mut streaming_receiver: mpsc::UnboundedReceiver) { + while streaming_receiver.recv().await.is_some() {} +} + +/// Wall-clock HH:MM:SS for the display timestamp (matches `worker.rs`). +fn format_current_timestamp() -> String { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + let hours = (now.as_secs() / 3600) % 24; + let minutes = (now.as_secs() / 60) % 60; + let seconds = now.as_secs() % 60; + format!("{:02}:{:02}:{:02}", hours, minutes, seconds) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn partial(seq: u64, text: &str) -> Emission { + Emission { seq, text: text.to_string(), is_partial: true } + } + fn final_(seq: u64, text: &str) -> Emission { + Emission { seq, text: text.to_string(), is_partial: false } + } + + #[test] + fn in_progress_sentence_updates_in_place_then_finalizes() { + let mut s = Segmenter::new(); + // The current sentence grows as a live partial under one id (0). + assert_eq!(s.advance("Hello", false), vec![partial(0, "Hello")]); + assert_eq!(s.advance("Hello world", false), vec![partial(0, "Hello world")]); + // It completes → finalized under the SAME id 0. + assert_eq!(s.advance("Hello world.", false), vec![final_(0, "Hello world.")]); + // The next sentence takes id 1, again as a growing partial. + assert_eq!(s.advance("Hello world. And", false), vec![partial(1, "And")]); + assert_eq!( + s.advance("Hello world. And more!", false), + vec![final_(1, "And more!")] + ); + } + + #[test] + fn sentence_completing_with_more_after_it_finalizes_and_starts_next_partial() { + let mut s = Segmenter::new(); + // "One." finalizes as id 0; " Two" starts as partial id 1 in the same update. + assert_eq!( + s.advance("One. Two", false), + vec![final_(0, "One."), partial(1, "Two")] + ); + } + + #[test] + fn multiple_sentences_in_one_update_all_finalize() { + let mut s = Segmenter::new(); + assert_eq!( + s.advance("One. Two. Three.", false), + vec![final_(0, "One."), final_(1, "Two."), final_(2, "Three.")] + ); + } + + #[test] + fn final_flushes_unterminated_tail_as_final() { + let mut s = Segmenter::new(); + assert_eq!(s.advance("A complete one.", false), vec![final_(0, "A complete one.")]); + // No terminal punctuation on the trailing bit; end-of-stream finalizes it. + assert_eq!( + s.advance("A complete one. trailing words", true), + vec![final_(1, "trailing words")] + ); + } + + #[test] + fn soft_flush_finalizes_long_punctuationless_runs_at_a_word_boundary() { + let mut s = Segmenter::new(); + let long = "word ".repeat(60); // 300 chars, no punctuation + let out = s.advance(&long, false); + assert!(!out.is_empty()); + assert!(!out[0].is_partial, "long run should finalize, not stay partial"); + assert!(out[0].text.len() <= SOFT_FLUSH_CHARS); + assert!(out[0].text.starts_with("word")); + } + + #[test] + fn next_break_finds_terminal_punctuation() { + assert_eq!(next_break("Hi there. rest"), Some("Hi there.".len())); + assert_eq!(next_break("Was?! next"), Some("Was?".len())); + assert_eq!(next_break("no boundary yet"), None); + } + + #[test] + fn utf8_is_not_split_mid_char() { + // German umlauts are multi-byte; cuts must land on char boundaries. + let mut s = Segmenter::new(); + let out = s.advance("Schöne Grüße. Nächster", false); + assert_eq!(out, vec![final_(0, "Schöne Grüße."), partial(1, "Nächster")]); + } +} diff --git a/frontend/src-tauri/src/audio/transcription/voxtral_realtime.rs b/frontend/src-tauri/src/audio/transcription/voxtral_realtime.rs new file mode 100644 index 0000000000..b500dbd46a --- /dev/null +++ b/frontend/src-tauri/src/audio/transcription/voxtral_realtime.rs @@ -0,0 +1,1307 @@ +// audio/transcription/voxtral_realtime.rs +// +// Streaming transcription over a vLLM/OpenAI-style `/v1/realtime` websocket, the +// worked example being Mistral's `Voxtral-Mini-4B-Realtime-2602` served by vLLM. +// +// ## Wire protocol — VERIFIED against a live vLLM/Voxtral endpoint (2026-07-12) +// +// ```text +// connect ws(s)://{host}/v1/realtime (wss:// adds Authorization: Bearer) +// → client: {"type":"session.update","model":…} (model REQUIRED, flat) +// → client: {"type":"input_audio_buffer.commit"} (REQUIRED — opens the buffer) +// ← server: {"type":"session.created","id":…} (ignored; may arrive late) +// → client: {"type":"input_audio_buffer.append","audio":""} (repeated) +// → client: {"type":"input_audio_buffer.commit","final":true} (on finish) +// ← server: {"type":"transcription.delta","delta":…} → Partial (delta is INCREMENTAL) +// ← server: {"type":"transcription.done","text":…} → Final +// ← server: {"type":"error","error":…} → Error +// ``` +// +// Audio is 16 kHz mono PCM16-LE (`i16 = f32 * i16::MAX`), base64-encoded. +// +// Findings that shaped this (all probed against the running server): +// - The **leading `commit` is required.** Omit it and the server ingests every +// append but emits no delta and no `done` — the session silently hangs. +// - **`session.update` accepts only `model`.** It is validated (a bad name → +// `model_not_found`) and must be **top-level**. Unknown fields are silently +// ignored, so `language` / delay are *not* part of this contract: Voxtral +// auto-detects language, and the transcription delay is a **server-side** knob +// (the model's `tekken.json`, 80–1200 ms in 80 ms steps, Mistral recommends +// 480), not per-session. `CustomTranscriptionConfig::delay_ms` is therefore +// persisted for UX/forward-compat but intentionally **not** sent here. +// - `transcription.delta` is **incremental** (one token per frame, often empty), +// not cumulative — this client accumulates and emits the running text. + +use base64::Engine as _; +use futures_util::{Sink, SinkExt, StreamExt}; +use tokio::sync::mpsc; +use tokio_tungstenite::connect_async; +use tokio_tungstenite::tungstenite::client::IntoClientRequest; +use tokio_tungstenite::tungstenite::http::header::AUTHORIZATION; +use tokio_tungstenite::tungstenite::{Error as WsError, Message}; +use tokio_util::sync::CancellationToken; + +use super::provider::TranscriptionError; +use super::streaming_provider::{ + StreamSession, StreamTranscriptEvent, StreamingTranscriptionProvider, +}; +use super::CustomTranscriptionConfig; +use async_trait::async_trait; +use log::{info, warn}; + +/// Raw PCM16 bytes per `input_audio_buffer.append` frame (~128 ms @ 16 kHz). +const FRAME_BYTES: usize = 4096; +/// Sample rate the provider is fed at, used to turn the configured session limit +/// into a sample budget. Set by the streaming worker's resampler. +const SAMPLE_RATE: u64 = 16_000; +/// How long a planned rollover waits for the outgoing session's final transcript. +/// Deliberately short: live audio is buffering the whole time, and whatever the +/// old session already transcribed is emitted regardless. +const ROLLOVER_FINISH_SECS: u64 = 5; +/// How long to wait for the terminal `transcription.done` after end-of-audio +/// before giving up, so a stalled server can't hang teardown forever. +const FINISH_TIMEOUT_SECS: u64 = 30; +/// How long `test_connection` waits for the handshake to settle. +const TEST_TIMEOUT_SECS: u64 = 10; +/// Reconnect attempts allowed per session (total, not per drop) before giving up. +/// A server restart mid-meeting should recover; a server that is gone for good, or +/// one that flaps endlessly, must stop and tell the user rather than retry forever. +const MAX_RECONNECT_ATTEMPTS: u32 = 3; +/// Backoff before the first reconnect attempt; doubles for each further attempt. +const RECONNECT_BACKOFF_MS: u64 = 500; +/// How long to keep reading a dropped socket for transcript frames that arrived +/// but hadn't been consumed yet, before writing the session off. +const DROP_DRAIN_MS: u64 = 50; + +/// Streaming provider backed by a `/v1/realtime` websocket. +pub struct VoxtralRealtimeProvider { + config: CustomTranscriptionConfig, +} + +impl VoxtralRealtimeProvider { + pub fn new(config: CustomTranscriptionConfig) -> Self { + Self { config } + } +} + +#[async_trait] +impl StreamingTranscriptionProvider for VoxtralRealtimeProvider { + async fn start_session( + &self, + _language: Option, + events: mpsc::UnboundedSender, + ) -> Result { + // Connect eagerly so a bad endpoint / unreachable server fails loudly at + // record-start rather than silently swallowing audio. + let stream = connect_and_handshake(&self.config).await?; + + let (audio_tx, audio_rx) = mpsc::unbounded_channel::>(); + let cancel = CancellationToken::new(); + let worker_cancel = cancel.clone(); + let config = self.config.clone(); + + tokio::spawn(async move { + run_session(config, stream, audio_rx, events, worker_cancel).await; + }); + + Ok(StreamSession { audio_tx, cancel }) + } + + async fn test_connection(&self) -> Result<(), TranscriptionError> { + let mut stream = connect_and_handshake(&self.config).await?; + + // Read until we either see a server error (fail — e.g. model_not_found) or + // the handshake clearly settled. Absence of an error within the window is + // treated as success: some servers stay quiet until audio arrives. + let deadline = tokio::time::Duration::from_secs(TEST_TIMEOUT_SECS); + loop { + match tokio::time::timeout(deadline, stream.next()).await { + Ok(Some(Ok(Message::Text(t)))) => { + match serde_json::from_str::(t.as_str()) { + Ok(ServerEvent::Error { error }) => { + let _ = stream.close(None).await; + return Err(TranscriptionError::EngineFailed(error)); + } + // Any non-error frame means the socket + handshake work. + Ok(_) => break, + Err(_) => break, + } + } + Ok(Some(Ok(Message::Close(_)))) | Ok(None) => { + return Err(TranscriptionError::EngineFailed( + "server closed the connection during handshake".to_string(), + )); + } + Ok(Some(Ok(_))) => continue, // ping/pong/binary + Ok(Some(Err(e))) => { + return Err(TranscriptionError::EngineFailed(format!("ws: {e}"))); + } + // Quiet server — connection + handshake succeeded, good enough. + Err(_) => break, + } + } + let _ = stream.close(None).await; + Ok(()) + } + + fn provider_name(&self) -> &'static str { + "Voxtral Realtime (streaming)" + } +} + +// === Session worker ======================================================== + +async fn run_session( + config: CustomTranscriptionConfig, + stream: WsStream, + mut audio_rx: mpsc::UnboundedReceiver>, + events: mpsc::UnboundedSender, + cancel: CancellationToken, +) { + let mut current = stream; + let mut reconnects_used: u32 = 0; + // Audio budget for one server session. Exceeding whatever the backend can hold + // is silent — it simply stops transcribing — so the client bounds it instead. + let limit_samples = config + .session_limit_seconds() + .map(|secs| secs as u64 * SAMPLE_RATE); + + loop { + let (mut write, mut read) = current.split(); + let mut cumulative = String::new(); + let mut pending: Vec = Vec::new(); + let mut samples_sent: u64 = 0; + + // Streaming phase: forward audio and surface segments until the audio + // channel closes (clean end-of-recording), the session hits its audio + // budget, it is cancelled, or the socket drops. + let phase = loop { + tokio::select! { + _ = cancel.cancelled() => break Phase::Cancelled, + pcm = audio_rx.recv() => match pcm { + Some(s) => { + samples_sent += s.len() as u64; + append_pcm16(&mut pending, &s); + if let Err(e) = flush_frames(&mut write, &mut pending, false).await { + // A failed send means the socket is gone, not that the + // audio was bad — recoverable like any other drop. + warn!("Voxtral realtime: send failed ({e}); connection considered dropped"); + break Phase::Dropped; + } + if limit_samples.is_some_and(|limit| samples_sent >= limit) { + break Phase::LimitReached; + } + } + None => break Phase::EndOfAudio, // audio_tx dropped → end of recording + }, + msg = read.next() => match apply(msg, &events, &mut cumulative) { + Flow::Continue | Flow::Terminal => {} + // A mid-stream server error usually means this session is + // finished even though the socket is still open (a full context + // is the common cause). Ignoring it is how a recording ends up + // silently untranscribed from that point on, so treat it like a + // drop and start a fresh session. + Flow::Failed | Flow::Closed => break Phase::Dropped, + }, + } + }; + + if let Phase::EndOfAudio = phase { + // Drain any audio buffered before the channel closed, flush the tail, + // then close the buffer and read until the final transcript arrives. + while let Ok(s) = audio_rx.try_recv() { + append_pcm16(&mut pending, &s); + } + let _ = flush_frames(&mut write, &mut pending, true).await; + let _ = + send_json_split(&mut write, &ClientEvent::Commit { final_flag: Some(true) }).await; + + let deadline = tokio::time::Duration::from_secs(FINISH_TIMEOUT_SECS); + loop { + match tokio::time::timeout(deadline, read.next()).await { + Ok(msg) => match apply(msg, &events, &mut cumulative) { + Flow::Continue => {} + Flow::Terminal | Flow::Failed | Flow::Closed => break, + }, + Err(_) => { + let _ = events.send(StreamTranscriptEvent::Error { + message: "realtime finish timed out".to_string(), + fatal: false, + }); + break; + } + } + } + } + + if let Phase::LimitReached = phase { + // Close the outgoing session the same way end-of-recording does, so the + // server transcribes the audio it is still holding instead of dropping + // it, then take its final transcript. Audio recorded during this + // handover keeps queueing in `audio_rx` (unbounded) and is sent to the + // replacement session — a rollover costs latency, never words. + let _ = flush_frames(&mut write, &mut pending, true).await; + let _ = + send_json_split(&mut write, &ClientEvent::Commit { final_flag: Some(true) }).await; + + let deadline = tokio::time::Duration::from_secs(ROLLOVER_FINISH_SECS); + loop { + match tokio::time::timeout(deadline, read.next()).await { + Ok(msg) => match apply(msg, &events, &mut cumulative) { + Flow::Continue => {} + Flow::Terminal | Flow::Failed | Flow::Closed => break, + }, + // A server that won't finalize can't be allowed to stall the + // handover; the replacement session is opened regardless. + Err(_) => break, + } + } + } + + let _ = write.close().await; + + match phase { + Phase::EndOfAudio | Phase::Cancelled => break, + Phase::LimitReached => { + // Close out this session's transcript. The replacement server starts + // counting from zero, so the worker must stop treating its offset + // into the old cumulative text as valid — a `Final` says exactly + // that. (`apply` already emitted one if `transcription.done` came + // back in time, leaving `cumulative` empty.) + if !cumulative.trim().is_empty() { + let _ = events.send(StreamTranscriptEvent::Final { + text: std::mem::take(&mut cumulative), + confidence: None, + }); + } + match connect_and_handshake(&config).await { + Ok(stream) => { + info!( + "Voxtral realtime: rolled over to a fresh session after {}s of audio", + samples_sent / SAMPLE_RATE + ); + // The endpoint just proved it is healthy, so earlier drops + // shouldn't count against a meeting that may run for hours. + reconnects_used = 0; + current = stream; + } + // The rollover was planned, but the server didn't answer — that + // is an outage like any other, so hand it to the retry path. + Err(e) => { + warn!("Voxtral realtime: rollover connect failed: {e}"); + match reconnect(&config, &events, &cancel, &mut reconnects_used).await { + Some(stream) => { + // Backoff means minutes of audio may have queued; + // sending it all would leave the live transcript + // permanently trailing the speaker. + while audio_rx.try_recv().is_ok() {} + current = stream; + } + None => break, + } + } + } + } + Phase::Dropped => { + // A drop is often noticed on a failed *send*, which can win the + // select! race against a *read* whose transcript frames already + // arrived. Drain what the socket still holds before writing the + // session off, so the last words spoken before the outage aren't + // silently dropped along with the connection. + loop { + let drain = tokio::time::Duration::from_millis(DROP_DRAIN_MS); + match tokio::time::timeout(drain, read.next()).await { + Ok(Some(msg)) => match apply(Some(msg), &events, &mut cumulative) { + Flow::Continue | Flow::Terminal => {} + Flow::Failed | Flow::Closed => break, + }, + // Timed out, or the stream is exhausted — nothing left. + _ => break, + } + } + + // Close out whatever this (now dead) session had transcribed. The + // replacement server starts a fresh transcript from scratch, so the + // worker has to stop treating its offset into the old cumulative + // text as valid — a `Final` is exactly that signal. + let _ = events.send(StreamTranscriptEvent::Final { + text: std::mem::take(&mut cumulative), + confidence: None, + }); + match reconnect(&config, &events, &cancel, &mut reconnects_used).await { + Some(stream) => { + // Discard audio buffered during the outage: the new session + // can't transcribe it, and replaying it would only push the + // live transcript permanently behind the speaker. + while audio_rx.try_recv().is_ok() {} + current = stream; + } + None => break, + } + } + } + } + + info!("Voxtral realtime session ended"); +} + +/// Why the streaming phase stopped. +enum Phase { + /// `audio_tx` was dropped — clean end of recording, finalize the transcript. + EndOfAudio, + /// The session used up its audio budget — finalize it and continue the + /// recording on a fresh session. + LimitReached, + /// The session was hard-cancelled; leave without finalizing. + Cancelled, + /// The socket closed or failed mid-recording — recoverable via reconnect. + Dropped, +} + +/// Re-establish a dropped session, with backoff, until it succeeds or the +/// per-session attempt budget runs out. +/// +/// Emits a warning per attempt and, on exhaustion, a **fatal** error — a websocket +/// that dies mid-meeting must never leave the user with a silently empty transcript. +async fn reconnect( + config: &CustomTranscriptionConfig, + events: &mpsc::UnboundedSender, + cancel: &CancellationToken, + used: &mut u32, +) -> Option { + while *used < MAX_RECONNECT_ATTEMPTS { + *used += 1; + let backoff = + tokio::time::Duration::from_millis(RECONNECT_BACKOFF_MS << (*used - 1).min(6)); + let _ = events.send(StreamTranscriptEvent::Error { + message: format!( + "Realtime transcription connection lost — reconnecting (attempt {}/{})", + *used, MAX_RECONNECT_ATTEMPTS + ), + fatal: false, + }); + + tokio::select! { + _ = cancel.cancelled() => return None, + _ = tokio::time::sleep(backoff) => {} + } + + match connect_and_handshake(config).await { + Ok(stream) => { + info!("Voxtral realtime: reconnected after {} attempt(s)", *used); + return Some(stream); + } + Err(e) => warn!("Voxtral realtime: reconnect attempt {} failed: {}", *used, e), + } + } + + let _ = events.send(StreamTranscriptEvent::Error { + message: format!( + "Lost the connection to the realtime transcription endpoint and could not \ + reconnect after {MAX_RECONNECT_ATTEMPTS} attempts. Transcription has stopped \ + for the rest of this recording — audio is still being recorded and can be \ + transcribed afterwards." + ), + fatal: true, + }); + None +} + +/// What a handled inbound message means for the read loop. +enum Flow { + Continue, + /// The server finished an utterance (`transcription.done`); the session itself + /// is still usable. + Terminal, + /// The server reported an error on this session. The socket may well still be + /// open, but the session should be considered spent and replaced. + Failed, + Closed, +} + +/// Handle one inbound WS message: emit mapped events and report how the read loop +/// should proceed. Tolerant of unknown / non-text frames. +fn apply( + msg: Option>, + events: &mpsc::UnboundedSender, + cumulative: &mut String, +) -> Flow { + let msg = match msg { + Some(Ok(m)) => m, + Some(Err(e)) => { + // Not fatal on its own: the caller treats a closed socket as a + // reconnectable drop, and only gives up once retries are exhausted. + let _ = events.send(StreamTranscriptEvent::Error { + message: format!("ws: {e}"), + fatal: false, + }); + return Flow::Closed; + } + None => return Flow::Closed, + }; + let payload = match msg { + Message::Text(t) => t.as_str().to_string(), + Message::Close(_) => return Flow::Closed, + _ => return Flow::Continue, // ping/pong/binary + }; + match serde_json::from_str::(&payload) { + Ok(ServerEvent::Delta { delta }) => { + if !delta.is_empty() { + cumulative.push_str(&delta); + let _ = events.send(StreamTranscriptEvent::Partial { + text: cumulative.clone(), + }); + } + Flow::Continue + } + Ok(ServerEvent::Done { text }) => { + // Emit the delta-accumulated `cumulative` VERBATIM (same string the + // partials carried) so the worker's byte offset into it stays valid at + // end-of-stream. Sending a trimmed/normalized `done.text` here would be + // a *different* string and make the worker re-segment the whole + // transcript. `done.text` is the same content in practice; it's used + // only when we somehow received no deltas. + let final_text = if cumulative.trim().is_empty() { + text.trim().to_string() + } else { + cumulative.clone() + }; + if !final_text.trim().is_empty() { + let _ = events.send(StreamTranscriptEvent::Final { + text: final_text, + confidence: None, + }); + } + cumulative.clear(); + Flow::Terminal + } + Ok(ServerEvent::Error { error }) => { + let _ = events.send(StreamTranscriptEvent::Error { message: error, fatal: false }); + Flow::Failed + } + Ok(ServerEvent::SessionCreated) | Ok(ServerEvent::Other) => Flow::Continue, + // Tolerate an unparseable shape rather than tearing the session down. + Err(_) => Flow::Continue, + } +} + +// === Connection ============================================================ + +type WsStream = tokio_tungstenite::WebSocketStream< + tokio_tungstenite::MaybeTlsStream, +>; + +/// Connect and perform the full session handshake: `session.update` (which +/// validates the model server-side) followed by the required leading commit that +/// opens the audio buffer. Shared by record-start, "Test Connection", and +/// reconnects so all three go through the identical sequence. +async fn connect_and_handshake( + config: &CustomTranscriptionConfig, +) -> Result { + let mut stream = connect(config).await?; + send_json(&mut stream, &ClientEvent::SessionUpdate { model: config.model.clone() }) + .await + .map_err(|e| TranscriptionError::EngineFailed(format!("session.update: {e}")))?; + send_json(&mut stream, &ClientEvent::Commit { final_flag: None }) + .await + .map_err(|e| TranscriptionError::EngineFailed(format!("open buffer: {e}")))?; + Ok(stream) +} + +async fn connect(config: &CustomTranscriptionConfig) -> Result { + let url = normalize_realtime_url(&config.endpoint); + let mut request = url + .as_str() + .into_client_request() + .map_err(|e| TranscriptionError::EngineFailed(format!("bad endpoint '{url}': {e}")))?; + if let Some(key) = config.api_key.as_deref().filter(|k| !k.trim().is_empty()) { + if let Ok(val) = format!("Bearer {key}").parse() { + request.headers_mut().insert(AUTHORIZATION, val); + } + } + // Name the endpoint in the error: a handshake failure here is usually the + // server (a proxy 502 while the speech backend boots), not the client, and an + // error that says only "502" sends you hunting the wrong side. + let (stream, _resp) = connect_async(request) + .await + .map_err(|e| TranscriptionError::EngineFailed(format!("connect to '{url}': {e}")))?; + warn!("Voxtral realtime connected: {url}"); + Ok(stream) +} + +/// Derive the websocket URL from a user-entered endpoint: map `http(s)` → `ws(s)` +/// and append the default `/v1/realtime` path when the user gave only a host. +fn normalize_realtime_url(endpoint: &str) -> String { + let mut url = endpoint.trim().to_string(); + if let Some(rest) = url.strip_prefix("https://") { + url = format!("wss://{rest}"); + } else if let Some(rest) = url.strip_prefix("http://") { + url = format!("ws://{rest}"); + } + let after_scheme = url.splitn(2, "://").nth(1).unwrap_or(""); + let path = after_scheme.splitn(2, '/').nth(1).unwrap_or(""); + if path.is_empty() { + url = format!("{}/v1/realtime", url.trim_end_matches('/')); + } + url +} + +// === Endpoint capability probe ============================================= + +/// Voxtral's audio encoder emits one token per 80 ms frame, so a second of audio +/// costs 12.5 context tokens. +const AUDIO_TOKENS_PER_SEC: f32 = 12.5; +/// The transcript the model writes shares that same context. Roughly 2.5 words a +/// second of natural speech, a little over a token per word. +const TEXT_TOKENS_PER_SEC: f32 = 3.5; +/// Fraction of the context a session is allowed to fill. The token rates above +/// are estimates, and the cost of stopping short is one extra rollover — the cost +/// of overshooting is a session that stops transcribing. +const CONTEXT_HEADROOM: f32 = 0.85; + +/// What an endpoint reports about how much audio one session can hold. +#[derive(Debug, Clone, serde::Serialize)] +pub struct DetectedSessionLimit { + /// Context window the server reports for the model, in tokens; `None` when the + /// endpoint doesn't announce one. + pub max_model_len: Option, + /// Session length derived from that context, in seconds. + pub recommended_seconds: Option, + /// Plain-language account of what was found, for the settings UI. + pub detail: String, +} + +/// Ask the endpoint how big its context is and turn that into a session length. +/// +/// vLLM (and several other OpenAI-compatible servers) publish `max_model_len` per +/// model on `GET /v1/models`. Servers that don't are not an error: the caller +/// falls back to a value the user sets by hand, since only they know what their +/// backend can take. +pub async fn detect_session_limit( + config: &CustomTranscriptionConfig, +) -> Result { + let url = models_url(&config.endpoint); + let mut request = reqwest::Client::new() + .get(&url) + .timeout(std::time::Duration::from_secs(TEST_TIMEOUT_SECS)); + if let Some(key) = config.api_key.as_deref().filter(|k| !k.trim().is_empty()) { + request = request.bearer_auth(key); + } + + let response = request + .send() + .await + .map_err(|e| TranscriptionError::EngineFailed(format!("GET {url}: {e}")))?; + if !response.status().is_success() { + return Err(TranscriptionError::EngineFailed(format!( + "GET {url} returned {}", + response.status() + ))); + } + let body: serde_json::Value = response + .json() + .await + .map_err(|e| TranscriptionError::EngineFailed(format!("GET {url}: {e}")))?; + + Ok(match context_length_for_model(&body, &config.model) { + Some(tokens) => { + let seconds = seconds_for_context(tokens); + DetectedSessionLimit { + max_model_len: Some(tokens), + recommended_seconds: Some(seconds), + detail: format!( + "The endpoint reports a {tokens}-token context for this model, \ + which holds about {seconds}s of audio plus its transcript." + ), + } + } + None => DetectedSessionLimit { + max_model_len: None, + recommended_seconds: None, + detail: format!( + "Reached {url}, but it doesn't announce a context size. \ + Set the session length by hand to match your backend." + ), + }, + }) +} + +/// Derive the `/v1/models` URL from the configured realtime endpoint: back to +/// http(s), and `…/v1/realtime` → `…/v1/models` (falling back to the host root for +/// endpoints with some other path layout). +fn models_url(endpoint: &str) -> String { + let mut url = endpoint.trim().trim_end_matches('/').to_string(); + if let Some(rest) = url.strip_prefix("wss://") { + url = format!("https://{rest}"); + } else if let Some(rest) = url.strip_prefix("ws://") { + url = format!("http://{rest}"); + } + if let Some(base) = url.strip_suffix("/v1/realtime") { + return format!("{base}/v1/models"); + } + // No recognizable realtime path — assume the API lives at the host root. + let (scheme, rest) = url.split_once("://").unwrap_or(("http", url.as_str())); + let authority = rest.split('/').next().unwrap_or(rest); + format!("{scheme}://{authority}/v1/models") +} + +/// Pull the context length out of an OpenAI-style `/v1/models` payload, preferring +/// the entry matching `model`. Key names differ between servers, so several are +/// accepted. +fn context_length_for_model(body: &serde_json::Value, model: &str) -> Option { + const KEYS: [&str; 3] = ["max_model_len", "max_context_length", "context_length"]; + let entries = body.get("data").and_then(|d| d.as_array())?; + let context_of = |entry: &serde_json::Value| -> Option { + KEYS.iter() + .find_map(|k| entry.get(k).and_then(|v| v.as_u64())) + .filter(|v| *v > 0) + .map(|v| v.min(u32::MAX as u64) as u32) + }; + entries + .iter() + .find(|e| e.get("id").and_then(|i| i.as_str()) == Some(model)) + .and_then(context_of) + // A server hosting one model under a different id is still telling us + // something useful; fall back to whatever entry announces a context. + .or_else(|| entries.iter().find_map(context_of)) +} + +/// Seconds of audio that fit in a context window of `max_model_len` tokens. +fn seconds_for_context(max_model_len: u32) -> u32 { + let usable = max_model_len as f32 * CONTEXT_HEADROOM; + let seconds = usable / (AUDIO_TOKENS_PER_SEC + TEXT_TOKENS_PER_SEC); + (seconds as u32).clamp(60, 3600) +} + +// === Wire types ============================================================ + +/// Client → server messages. Internally tagged by `type`. +#[derive(serde::Serialize)] +#[serde(tag = "type")] +enum ClientEvent { + /// `model` is the only field `/v1/realtime` accepts here — required, validated, + /// top-level. See the module note. + #[serde(rename = "session.update")] + SessionUpdate { model: String }, + #[serde(rename = "input_audio_buffer.append")] + Append { audio: String }, + #[serde(rename = "input_audio_buffer.commit")] + Commit { + #[serde(rename = "final", skip_serializing_if = "Option::is_none")] + final_flag: Option, + }, +} + +/// Server → client messages. Unknown `type`s are tolerated (`Other`). +#[derive(serde::Deserialize)] +#[serde(tag = "type")] +enum ServerEvent { + #[serde(rename = "session.created")] + SessionCreated, + #[serde(rename = "transcription.delta")] + Delta { + #[serde(default)] + delta: String, + }, + #[serde(rename = "transcription.done")] + Done { + #[serde(default)] + text: String, + }, + #[serde(rename = "error")] + Error { + #[serde(default)] + error: String, + }, + #[serde(other)] + Other, +} + +// === PCM framing + send helpers ============================================ + +/// Append 16 kHz mono f32 samples to `buf` as PCM16 little-endian. +fn append_pcm16(buf: &mut Vec, samples: &[f32]) { + buf.reserve(samples.len() * 2); + for &s in samples { + let v = (s.clamp(-1.0, 1.0) * i16::MAX as f32) as i16; + buf.extend_from_slice(&v.to_le_bytes()); + } +} + +fn b64(bytes: &[u8]) -> String { + base64::engine::general_purpose::STANDARD.encode(bytes) +} + +/// Send full `FRAME_BYTES` frames from `pending` as base64 `append` messages; when +/// `flush_all`, also send the trailing partial frame. +async fn flush_frames(write: &mut S, pending: &mut Vec, flush_all: bool) -> Result<(), WsError> +where + S: Sink + Unpin, +{ + while pending.len() >= FRAME_BYTES { + let frame: Vec = pending.drain(..FRAME_BYTES).collect(); + send_json_split(write, &ClientEvent::Append { audio: b64(&frame) }).await?; + } + if flush_all && !pending.is_empty() { + let frame = std::mem::take(pending); + send_json_split(write, &ClientEvent::Append { audio: b64(&frame) }).await?; + } + Ok(()) +} + +/// Serialize + send on a split sink half (used inside the worker). +async fn send_json_split(write: &mut S, ev: &ClientEvent) -> Result<(), WsError> +where + S: Sink + Unpin, +{ + let txt = serde_json::to_string(ev).unwrap_or_default(); + write.send(Message::Text(txt.into())).await +} + +/// Serialize + send on the whole stream (used before splitting, during handshake). +async fn send_json(stream: &mut WsStream, ev: &ClientEvent) -> Result<(), WsError> { + let txt = serde_json::to_string(ev).unwrap_or_default(); + stream.send(Message::Text(txt.into())).await +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::Arc; + + #[test] + fn client_event_json_shapes() { + assert_eq!( + serde_json::to_string(&ClientEvent::SessionUpdate { model: "m".into() }).unwrap(), + r#"{"type":"session.update","model":"m"}"# + ); + assert_eq!( + serde_json::to_string(&ClientEvent::Commit { final_flag: Some(true) }).unwrap(), + r#"{"type":"input_audio_buffer.commit","final":true}"# + ); + assert_eq!( + serde_json::to_string(&ClientEvent::Commit { final_flag: None }).unwrap(), + r#"{"type":"input_audio_buffer.commit"}"# + ); + assert_eq!( + serde_json::to_string(&ClientEvent::Append { audio: "AAA=".into() }).unwrap(), + r#"{"type":"input_audio_buffer.append","audio":"AAA="}"# + ); + } + + #[test] + fn pcm16_framing_is_little_endian() { + // +1.0 → i16::MAX (0x7FFF), -1.0 → -i16::MAX (0x8001), 0.0 → 0. + let mut buf = Vec::new(); + append_pcm16(&mut buf, &[0.0, 1.0, -1.0]); + assert_eq!(buf, vec![0x00, 0x00, 0xFF, 0x7F, 0x01, 0x80]); + let decoded = base64::engine::general_purpose::STANDARD + .decode(b64(&buf)) + .unwrap(); + assert_eq!(decoded, buf); + } + + #[test] + fn server_event_parsing() { + assert!(matches!( + serde_json::from_str::(r#"{"type":"transcription.delta","delta":"hi"}"#), + Ok(ServerEvent::Delta { delta }) if delta == "hi" + )); + assert!(matches!( + serde_json::from_str::(r#"{"type":"transcription.done","text":"done"}"#), + Ok(ServerEvent::Done { text }) if text == "done" + )); + assert!(matches!( + serde_json::from_str::(r#"{"type":"session.created","id":"s1"}"#), + Ok(ServerEvent::SessionCreated) + )); + assert!(matches!( + serde_json::from_str::(r#"{"type":"something.else"}"#), + Ok(ServerEvent::Other) + )); + } + + #[test] + fn url_normalization() { + // Host-only → default realtime path appended. + assert_eq!( + normalize_realtime_url("ws://localhost:8000"), + "ws://localhost:8000/v1/realtime" + ); + // Trailing slash treated as no path. + assert_eq!( + normalize_realtime_url("ws://localhost:8000/"), + "ws://localhost:8000/v1/realtime" + ); + // http(s) scheme mapped to ws(s). + assert_eq!( + normalize_realtime_url("https://asr.example.com"), + "wss://asr.example.com/v1/realtime" + ); + assert_eq!( + normalize_realtime_url("http://box:9000"), + "ws://box:9000/v1/realtime" + ); + // Explicit path preserved. + assert_eq!( + normalize_realtime_url("ws://host/custom/realtime"), + "ws://host/custom/realtime" + ); + } + + /// End-to-end against a mock `/v1/realtime` server: the provider must send + /// session.update + a leading commit + audio + a final commit, and turn the + /// server's delta/done into Partial → Final events. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn streams_partial_then_final() { + use tokio::net::TcpListener; + + let tcp = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = tcp.local_addr().unwrap(); + let server = tokio::spawn(async move { + let (sock, _) = tcp.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + ws.send(Message::Text(r#"{"type":"session.created","id":"s1"}"#.into())) + .await + .unwrap(); + let mut saw_final_commit = false; + while let Some(Ok(msg)) = ws.next().await { + if let Message::Text(t) = msg { + if t.as_str().contains(r#""final":true"#) { + saw_final_commit = true; + // Incremental deltas that accumulate to the full text, as a + // real server streams them. + ws.send(Message::Text( + r#"{"type":"transcription.delta","delta":"hallo "}"#.into(), + )) + .await + .unwrap(); + ws.send(Message::Text( + r#"{"type":"transcription.delta","delta":"welt"}"#.into(), + )) + .await + .unwrap(); + ws.send(Message::Text( + r#"{"type":"transcription.done","text":"hallo welt"}"#.into(), + )) + .await + .unwrap(); + break; + } + } + } + saw_final_commit + }); + + let provider = VoxtralRealtimeProvider::new(test_config(addr)); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let session = provider.start_session(None, tx).await.unwrap(); + session.audio_tx.send(vec![0.1f32; 1600]).unwrap(); + // Close the audio channel → triggers flush + final commit. + drop(session.audio_tx); + + let mut saw_partial = false; + let mut final_text = None; + while let Some(ev) = rx.recv().await { + match ev { + StreamTranscriptEvent::Partial { .. } => saw_partial = true, + StreamTranscriptEvent::Final { text, .. } => { + final_text = Some(text); + break; + } + StreamTranscriptEvent::Error { message, .. } => { + panic!("unexpected error: {message}") + } + } + } + + assert!(server.await.unwrap(), "server saw the final commit"); + assert!(saw_partial, "expected a streaming Partial"); + assert_eq!(final_text.as_deref(), Some("hallo welt")); + } + + fn test_config(addr: std::net::SocketAddr) -> CustomTranscriptionConfig { + CustomTranscriptionConfig { + endpoint: format!("ws://{addr}/v1/realtime"), + api_key: None, + model: "m".to_string(), + protocol: "voxtral-realtime".to_string(), + delay_ms: None, + // Far beyond anything these tests feed, so only the rollover test + // exercises the limit. + max_session_seconds: None, + } + } + + /// Read client frames until the leading `input_audio_buffer.commit` that ends + /// the handshake, so a mock server can't hang up before the client is ready. + async fn await_handshake(ws: &mut tokio_tungstenite::WebSocketStream) + where + S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, + { + while let Some(Ok(msg)) = ws.next().await { + if let Message::Text(t) = msg { + if t.as_str().contains("input_audio_buffer.commit") { + return; + } + } + } + } + + /// A server that drops the connection mid-recording must not kill transcription: + /// the provider reconnects and keeps transcribing on the new socket. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn reconnects_after_the_server_drops_mid_session() { + use tokio::net::TcpListener; + + let tcp = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = tcp.local_addr().unwrap(); + let server = tokio::spawn(async move { + // Connection 1: transcribe a little, then hang up mid-recording. + let (sock, _) = tcp.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + await_handshake(&mut ws).await; + ws.send(Message::Text( + r#"{"type":"transcription.delta","delta":"before drop"}"#.into(), + )) + .await + .unwrap(); + let _ = ws.close(None).await; + drop(ws); + + // Connection 2: the reconnect. Transcribe through to done. + let (sock, _) = tcp.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + while let Some(Ok(msg)) = ws.next().await { + if let Message::Text(t) = msg { + if t.as_str().contains(r#""final":true"#) { + ws.send(Message::Text( + r#"{"type":"transcription.done","text":"after reconnect"}"#.into(), + )) + .await + .unwrap(); + return true; + } + } + } + false + }); + + let provider = VoxtralRealtimeProvider::new(test_config(addr)); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let session = provider.start_session(None, tx).await.unwrap(); + + // Feed audio continuously (so the reconnected socket has something to + // flush) until told to stop — dropping the sender ends the recording. + let stop = Arc::new(AtomicBool::new(false)); + let pump_stop = stop.clone(); + let audio_tx = session.audio_tx; + let pump = tokio::spawn(async move { + while !pump_stop.load(Ordering::Relaxed) { + if audio_tx.send(vec![0.1f32; 1600]).is_err() { + break; + } + tokio::time::sleep(tokio::time::Duration::from_millis(20)).await; + } + drop(audio_tx); // end of recording → final commit on the live socket + }); + + let mut reconnect_notices = 0; + let mut finals = Vec::new(); + let mut fatal = None; + while let Some(ev) = rx.recv().await { + match ev { + StreamTranscriptEvent::Partial { .. } => {} + StreamTranscriptEvent::Final { text, .. } => { + finals.push(text); + if finals.len() == 2 { + break; + } + } + StreamTranscriptEvent::Error { fatal: true, message } => { + fatal = Some(message); + break; + } + StreamTranscriptEvent::Error { message, .. } => { + if message.contains("reconnecting (attempt") { + reconnect_notices += 1; + // Reconnect is under way; stop the audio so the new session + // reaches its final commit and we can assert on the result. + stop.store(true, Ordering::Relaxed); + } + } + } + } + let _ = pump.await; + + assert!(reconnect_notices >= 1, "a dropped connection must surface a warning"); + assert_eq!(fatal, None, "a recoverable drop must not be reported as fatal"); + assert!( + server.await.unwrap(), + "the reconnected session must reach the final commit" + ); + assert_eq!( + finals.first().map(String::as_str), + Some("before drop"), + "text from the dead session is closed out before reconnecting" + ); + assert_eq!( + finals.get(1).map(String::as_str), + Some("after reconnect"), + "transcription continues on the reconnected socket" + ); + } + + /// When the endpoint stays down, the provider must give up and say so loudly — + /// the silent-failure case this guards against is a whole meeting recorded with + /// an empty transcript and no indication anything went wrong. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn gives_up_with_a_fatal_error_when_the_endpoint_stays_down() { + use tokio::net::TcpListener; + + let tcp = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = tcp.local_addr().unwrap(); + // The first connection completes the handshake and is then dropped; every + // retry is refused outright, standing in for a server that stays down. + let server = tokio::spawn(async move { + let (sock, _) = tcp.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + await_handshake(&mut ws).await; + let _ = ws.close(None).await; + drop(ws); + while let Ok((sock, _)) = tcp.accept().await { + drop(sock); + } + }); + + let provider = VoxtralRealtimeProvider::new(test_config(addr)); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let session = provider.start_session(None, tx).await.unwrap(); + // Hold the recording open so teardown can't be mistaken for a clean stop. + let _audio_tx = session.audio_tx; + + let mut reconnect_notices = 0; + let mut fatal = None; + while let Some(ev) = rx.recv().await { + match ev { + StreamTranscriptEvent::Error { fatal: true, message } => { + fatal = Some(message); + break; + } + StreamTranscriptEvent::Error { message, .. } => { + if message.contains("reconnecting (attempt") { + reconnect_notices += 1; + } + } + _ => {} + } + } + + let fatal = fatal.expect("exhausted reconnects must emit a FATAL error, not silence"); + assert!( + fatal.contains("could not reconnect"), + "fatal error should explain what happened: {fatal}" + ); + assert!( + fatal.contains("audio is still being recorded"), + "fatal error should tell the user the recording itself is safe: {fatal}" + ); + assert_eq!( + reconnect_notices, MAX_RECONNECT_ATTEMPTS as usize, + "one notice per reconnect attempt before giving up" + ); + // The events channel closes once the worker exits — no zombie session. + assert!(rx.recv().await.is_none(), "worker should exit after giving up"); + server.abort(); + } + + /// A long meeting must outlive whatever one server session can hold: at the + /// configured audio budget the provider finalizes the session and continues on + /// a fresh one, without the user seeing an error. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn rolls_over_to_a_fresh_session_at_the_audio_limit() { + use tokio::net::TcpListener; + + let tcp = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = tcp.local_addr().unwrap(); + let server = tokio::spawn(async move { + // Session 1 ends when the client finalizes it at the audio limit. + let (sock, _) = tcp.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + await_handshake(&mut ws).await; + let mut sessions = 0; + while let Some(Ok(msg)) = ws.next().await { + if let Message::Text(t) = msg { + if t.as_str().contains(r#""final":true"#) { + sessions += 1; + ws.send(Message::Text( + r#"{"type":"transcription.done","text":"first session"}"#.into(), + )) + .await + .unwrap(); + break; + } + } + } + + // Session 2: the rollover. It runs to the end of the recording. + let (sock, _) = tcp.accept().await.unwrap(); + let mut ws = tokio_tungstenite::accept_async(sock).await.unwrap(); + await_handshake(&mut ws).await; + while let Some(Ok(msg)) = ws.next().await { + if let Message::Text(t) = msg { + if t.as_str().contains(r#""final":true"#) { + sessions += 1; + ws.send(Message::Text( + r#"{"type":"transcription.done","text":"second session"}"#.into(), + )) + .await + .unwrap(); + break; + } + } + } + sessions + }); + + // One second of audio per session, fed in 100 ms pushes. + let mut config = test_config(addr); + config.max_session_seconds = Some(1); + let provider = VoxtralRealtimeProvider::new(config); + let (tx, mut rx) = mpsc::unbounded_channel::(); + let session = provider.start_session(None, tx).await.unwrap(); + + let stop = Arc::new(AtomicBool::new(false)); + let pump_stop = stop.clone(); + let audio_tx = session.audio_tx; + let pump = tokio::spawn(async move { + while !pump_stop.load(Ordering::Relaxed) { + if audio_tx.send(vec![0.1f32; 1600]).is_err() { + break; + } + tokio::time::sleep(tokio::time::Duration::from_millis(5)).await; + } + drop(audio_tx); // end of recording → final commit on session 2 + }); + + let mut finals = Vec::new(); + let mut warnings = Vec::new(); + while let Some(ev) = rx.recv().await { + match ev { + StreamTranscriptEvent::Partial { .. } => {} + StreamTranscriptEvent::Final { text, .. } => { + finals.push(text); + // The first session closed; end the recording so the second one + // finalizes too and we can assert both landed. + stop.store(true, Ordering::Relaxed); + if finals.len() == 2 { + break; + } + } + StreamTranscriptEvent::Error { message, .. } => warnings.push(message), + } + } + let _ = pump.await; + + assert_eq!(server.await.unwrap(), 2, "both sessions must be finalized"); + assert_eq!( + finals, + vec!["first session".to_string(), "second session".to_string()], + "transcription continues across the rollover" + ); + assert!( + !warnings.iter().any(|w| w.contains("reconnecting")), + "a planned rollover is not a connection failure: {warnings:?}" + ); + } + + #[test] + fn session_limit_defaults_and_can_be_disabled() { + let mut config = CustomTranscriptionConfig { + endpoint: "ws://host".into(), + api_key: None, + model: "m".into(), + protocol: "voxtral-realtime".into(), + delay_ms: None, + max_session_seconds: None, + }; + // Unset → the conservative default, so long meetings survive out of the box. + assert_eq!( + config.session_limit_seconds(), + Some(crate::audio::transcription::DEFAULT_MAX_SESSION_SECONDS) + ); + config.max_session_seconds = Some(900); + assert_eq!(config.session_limit_seconds(), Some(900)); + // 0 is the escape hatch: one session for the whole recording. + config.max_session_seconds = Some(0); + assert_eq!(config.session_limit_seconds(), None); + } + + #[test] + fn config_without_the_limit_field_still_deserializes() { + // Configs saved before this setting existed must keep loading. + let stored = r#"{"endpoint":"ws://host","apiKey":null,"model":"m","protocol":"voxtral-realtime","delayMs":null}"#; + let config: CustomTranscriptionConfig = serde_json::from_str(stored).unwrap(); + assert_eq!(config.max_session_seconds, None); + assert_eq!( + config.session_limit_seconds(), + Some(crate::audio::transcription::DEFAULT_MAX_SESSION_SECONDS) + ); + } + + #[test] + fn models_url_derivation() { + assert_eq!( + models_url("wss://asr.example.com/v1/realtime"), + "https://asr.example.com/v1/models" + ); + assert_eq!( + models_url("ws://localhost:8000/v1/realtime"), + "http://localhost:8000/v1/models" + ); + // Host-only endpoints (the realtime path is implied) and trailing slashes. + assert_eq!(models_url("ws://localhost:8000/"), "http://localhost:8000/v1/models"); + assert_eq!( + models_url("https://asr.example.com"), + "https://asr.example.com/v1/models" + ); + // An unusual path layout falls back to the host root rather than guessing. + assert_eq!(models_url("ws://host/custom/socket"), "http://host/v1/models"); + } + + #[test] + fn context_length_prefers_the_requested_model() { + let body = serde_json::json!({ + "object": "list", + "data": [ + { "id": "other", "max_model_len": 32768 }, + { "id": "wanted", "max_model_len": 8192 }, + ] + }); + assert_eq!(context_length_for_model(&body, "wanted"), Some(8192)); + // Unknown id → fall back to whatever the server does announce. + assert_eq!(context_length_for_model(&body, "missing"), Some(32768)); + } + + #[test] + fn context_length_absent_is_not_an_error() { + let body = serde_json::json!({ "data": [{ "id": "m", "object": "model" }] }); + assert_eq!(context_length_for_model(&body, "m"), None); + // Alternate spellings used by non-vLLM servers. + let alt = serde_json::json!({ "data": [{ "id": "m", "context_length": 4096 }] }); + assert_eq!(context_length_for_model(&alt, "m"), Some(4096)); + } + + #[test] + fn seconds_for_context_leaves_headroom() { + // 8192 tokens: ~655s of audio alone, less once the transcript and headroom + // are accounted for — and safely under the ~8 min where such a server dies. + let secs = seconds_for_context(8192); + assert!((400..=500).contains(&secs), "unexpected: {secs}"); + assert!(seconds_for_context(32768) > seconds_for_context(8192)); + // Absurd values stay in a usable band. + assert_eq!(seconds_for_context(128), 60); + assert_eq!(seconds_for_context(10_000_000), 3600); + } +} diff --git a/frontend/src-tauri/src/database/models.rs b/frontend/src-tauri/src/database/models.rs index 9cc8f57338..04ee543780 100644 --- a/frontend/src-tauri/src/database/models.rs +++ b/frontend/src-tauri/src/database/models.rs @@ -127,4 +127,19 @@ pub struct TranscriptSetting { #[sqlx(rename = "openaiApiKey")] #[serde(rename = "openaiApiKey")] pub openai_api_key: Option, + /// Custom streaming (websocket) transcription endpoint configuration stored as JSON + #[sqlx(rename = "customTranscriptionConfig")] + #[serde(rename = "customTranscriptionConfig")] + pub custom_transcription_config: Option, +} + +impl TranscriptSetting { + /// Parse the custom streaming transcription config from JSON string + pub fn get_custom_transcription_config( + &self, + ) -> Option { + self.custom_transcription_config + .as_ref() + .and_then(|json| serde_json::from_str(json).ok()) + } } diff --git a/frontend/src-tauri/src/database/repositories/setting.rs b/frontend/src-tauri/src/database/repositories/setting.rs index 79583f91c2..d6a78f4069 100644 --- a/frontend/src-tauri/src/database/repositories/setting.rs +++ b/frontend/src-tauri/src/database/repositories/setting.rs @@ -1,3 +1,4 @@ +use crate::audio::transcription::{CustomTranscriptionConfig, CUSTOM_STREAMING_PROVIDER}; use crate::database::models::{Setting, TranscriptSetting}; use crate::summary::CustomOpenAIConfig; use sqlx::SqlitePool; @@ -180,6 +181,8 @@ impl SettingsRepository { let api_key_column = match provider { "localWhisper" => "whisperApiKey", "parakeet" => return Ok(()), // Parakeet doesn't need an API key, return early + // Custom streaming stores its key inside the JSON config, not a column + p if p == CUSTOM_STREAMING_PROVIDER => return Ok(()), "deepgram" => "deepgramApiKey", "elevenLabs" => "elevenLabsApiKey", "groq" => "groqApiKey", @@ -212,6 +215,8 @@ impl SettingsRepository { let api_key_column = match provider { "localWhisper" => "whisperApiKey", "parakeet" => return Ok(None), // Parakeet doesn't need an API key + // Custom streaming stores its key inside the JSON config, not a column + p if p == CUSTOM_STREAMING_PROVIDER => return Ok(None), "deepgram" => "deepgramApiKey", "elevenLabs" => "elevenLabsApiKey", "groq" => "groqApiKey", @@ -345,4 +350,82 @@ impl SettingsRepository { Ok(()) } + + // ===== CUSTOM STREAMING TRANSCRIPTION CONFIG METHODS ===== + + /// Gets the custom streaming transcription configuration from JSON + /// + /// # Returns + /// * `Ok(Some(CustomTranscriptionConfig))` - Config exists and is valid JSON + /// * `Ok(None)` - No config stored + /// * `Err(sqlx::Error)` - Database error + pub async fn get_custom_transcription_config( + pool: &SqlitePool, + ) -> std::result::Result, sqlx::Error> { + use sqlx::Row; + + let row = sqlx::query( + r#" + SELECT customTranscriptionConfig + FROM transcript_settings + WHERE id = '1' + LIMIT 1 + "#, + ) + .fetch_optional(pool) + .await?; + + match row { + Some(record) => { + let config_json: Option = record.get("customTranscriptionConfig"); + + if let Some(json) = config_json { + let config: CustomTranscriptionConfig = + serde_json::from_str(&json).map_err(|e| { + sqlx::Error::Protocol( + format!("Invalid JSON in customTranscriptionConfig: {}", e).into(), + ) + })?; + + Ok(Some(config)) + } else { + Ok(None) + } + } + None => Ok(None), + } + } + + /// Saves the custom streaming transcription configuration as JSON. + /// + /// Also sets `provider` to the custom-streaming identifier and mirrors the model, + /// so the active transcript config selects this provider. + pub async fn save_custom_transcription_config( + pool: &SqlitePool, + config: &CustomTranscriptionConfig, + ) -> std::result::Result<(), sqlx::Error> { + let config_json = serde_json::to_string(config).map_err(|e| { + sqlx::Error::Protocol( + format!("Failed to serialize transcription config to JSON: {}", e).into(), + ) + })?; + + sqlx::query( + r#" + INSERT INTO transcript_settings (id, provider, model, customTranscriptionConfig) + VALUES ('1', $1, $2, $3) + ON CONFLICT(id) DO UPDATE SET + provider = excluded.provider, + model = excluded.model, + customTranscriptionConfig = excluded.customTranscriptionConfig + "#, + ) + .bind(CUSTOM_STREAMING_PROVIDER) + .bind(&config.model) + .bind(config_json) + .execute(pool) + .await?; + + Ok(()) + } } diff --git a/frontend/src-tauri/src/lib.rs b/frontend/src-tauri/src/lib.rs index e757c98748..159558dc8b 100644 --- a/frontend/src-tauri/src/lib.rs +++ b/frontend/src-tauri/src/lib.rs @@ -643,6 +643,10 @@ pub fn run() { api::api_get_transcript_config, api::api_save_transcript_config, api::api_get_transcript_api_key, + api::api_get_custom_transcription_config, + api::api_save_custom_transcription_config, + api::api_test_custom_transcription_connection, + api::api_detect_custom_transcription_limits, api::api_delete_meeting, api::api_get_meeting, api::api_get_meeting_metadata, diff --git a/frontend/src/app/_components/TranscriptPanel.tsx b/frontend/src/app/_components/TranscriptPanel.tsx index 793f25505f..9cbb27339e 100644 --- a/frontend/src/app/_components/TranscriptPanel.tsx +++ b/frontend/src/app/_components/TranscriptPanel.tsx @@ -111,7 +111,12 @@ export function TranscriptPanel({ isPaused={isPaused} isProcessing={isProcessingStop} isStopping={isStopping} - enableStreaming={isRecording} + // The typewriter reveal animates each NEW (immutable, id-keyed) + // segment. A realtime provider instead grows one segment's text in + // place under a stable id, which the typewriter can't track (it + // snapshots on id change), so it would freeze on the first word. + // Disable it there and show the server's live text directly. + enableStreaming={isRecording && transcriptModelConfig.provider !== 'customStreaming'} showConfidence={true} /> diff --git a/frontend/src/components/LanguageSelection.tsx b/frontend/src/components/LanguageSelection.tsx index 1ce0b99517..ac98434972 100644 --- a/frontend/src/components/LanguageSelection.tsx +++ b/frontend/src/components/LanguageSelection.tsx @@ -118,7 +118,7 @@ interface LanguageSelectionProps { selectedLanguage: string; onLanguageChange: (language: string) => void; disabled?: boolean; - provider?: 'localWhisper' | 'parakeet' | 'deepgram' | 'elevenLabs' | 'groq' | 'openai'; + provider?: 'localWhisper' | 'parakeet' | 'deepgram' | 'elevenLabs' | 'groq' | 'openai' | 'customStreaming'; } export function LanguageSelection({ @@ -130,9 +130,11 @@ export function LanguageSelection({ const [saving, setSaving] = useState(false); const { setSelectedLanguage } = useConfig(); - // Parakeet only supports auto-detection (doesn't support manual language selection) + // Parakeet and Voxtral realtime only support auto-detection (no manual language selection) const isParakeet = provider === 'parakeet'; - const availableLanguages = isParakeet + const isCustomStreaming = provider === 'customStreaming'; + const autoDetectOnly = isParakeet || isCustomStreaming; + const availableLanguages = autoDetectOnly ? LANGUAGES.filter(lang => lang.code === 'auto' || lang.code === 'auto-translate') : LANGUAGES; @@ -205,6 +207,14 @@ export function LanguageSelection({ )} + {/* Realtime streaming language limitation warning */} + {isCustomStreaming && ( +
+

ℹ️ Realtime Language Support

+

Realtime models such as Voxtral detect the spoken language automatically; manual language selection is handled server-side and not available here.

+
+ )} + {/* Info text */}

diff --git a/frontend/src/components/TranscriptSettings.tsx b/frontend/src/components/TranscriptSettings.tsx index 146906703d..0ae2992634 100644 --- a/frontend/src/components/TranscriptSettings.tsx +++ b/frontend/src/components/TranscriptSettings.tsx @@ -1,16 +1,18 @@ import { useState, useEffect } from 'react'; import { invoke } from '@tauri-apps/api/core'; +import { toast } from 'sonner'; import { Select, SelectContent, SelectItem, SelectTrigger, SelectValue } from './ui/select'; import { Input } from './ui/input'; import { Button } from './ui/button'; import { Label } from './ui/label'; -import { Eye, EyeOff, Lock, Unlock } from 'lucide-react'; +import { Eye, EyeOff, Lock, Unlock, Loader2 } from 'lucide-react'; import { ModelManager } from './WhisperModelManager'; import { ParakeetModelManager } from './ParakeetModelManager'; +import { configService } from '@/services/configService'; export interface TranscriptModelProps { - provider: 'localWhisper' | 'parakeet' | 'deepgram' | 'elevenLabs' | 'groq' | 'openai'; + provider: 'localWhisper' | 'parakeet' | 'deepgram' | 'elevenLabs' | 'groq' | 'openai' | 'customStreaming'; model: string; apiKey?: string | null; } @@ -39,6 +41,37 @@ export function TranscriptSettings({ transcriptModelConfig, setTranscriptModelCo } }, [transcriptModelConfig.provider]); + // ── Custom streaming (websocket) transcription endpoint state ────────────── + const [streamingEndpoint, setStreamingEndpoint] = useState(''); + const [streamingModel, setStreamingModel] = useState(''); + const [streamingApiKey, setStreamingApiKey] = useState(''); + const [showStreamingApiKey, setShowStreamingApiKey] = useState(false); + const [isTestingStreaming, setIsTestingStreaming] = useState(false); + const [isSavingStreaming, setIsSavingStreaming] = useState(false); + // Seconds of audio per server session. Empty = use the backend default. + const [streamingMaxSeconds, setStreamingMaxSeconds] = useState(''); + const [isDetectingLimits, setIsDetectingLimits] = useState(false); + + // Load the saved streaming config when the streaming provider is selected. + useEffect(() => { + if (uiProvider !== 'customStreaming') return; + let cancelled = false; + configService.getCustomTranscriptionConfig() + .then((cfg) => { + if (cancelled || !cfg) return; + setStreamingEndpoint(cfg.endpoint || ''); + setStreamingModel(cfg.model || ''); + setStreamingApiKey(cfg.apiKey || ''); + setStreamingMaxSeconds( + cfg.maxSessionSeconds === null || cfg.maxSessionSeconds === undefined + ? '' + : String(cfg.maxSessionSeconds) + ); + }) + .catch((err) => console.error('Failed to load streaming transcription config:', err)); + return () => { cancelled = true; }; + }, [uiProvider]); + const fetchApiKey = async (provider: string) => { try { @@ -50,6 +83,105 @@ export function TranscriptSettings({ transcriptModelConfig, setTranscriptModelCo setApiKey(null); } }; + + const testStreamingConnection = async () => { + if (!streamingEndpoint.trim() || !streamingModel.trim()) { + toast.error('Please enter the endpoint URL and model name first'); + return; + } + setIsTestingStreaming(true); + try { + const result = await configService.testCustomTranscriptionConnection( + streamingEndpoint.trim(), + streamingModel.trim(), + streamingApiKey.trim() || null, + ); + toast.success(result.message || 'Connection successful!'); + } catch (err) { + toast.error(err instanceof Error ? err.message : String(err)); + } finally { + setIsTestingStreaming(false); + } + }; + + /** + * Ask the endpoint how big its context is and turn that into a session length. + * Servers that don't announce one aren't an error — the user fills it in. + */ + const detectEndpointLimits = async () => { + if (!streamingEndpoint.trim() || !streamingModel.trim()) { + toast.error('Please enter the endpoint URL and model name first'); + return; + } + setIsDetectingLimits(true); + try { + const limits = await configService.detectCustomTranscriptionLimits( + streamingEndpoint.trim(), + streamingModel.trim(), + streamingApiKey.trim() || null, + ); + if (limits.recommended_seconds !== null) { + setStreamingMaxSeconds(String(limits.recommended_seconds)); + toast.success(limits.detail); + } else { + toast.info(limits.detail); + } + } catch (err) { + toast.error(err instanceof Error ? err.message : String(err)); + } finally { + setIsDetectingLimits(false); + } + }; + + /** + * The session length as the backend wants it: `null` for "use the default", + * or `undefined` when what's in the box isn't a usable number. + */ + const parseMaxSessionSeconds = (): number | null | undefined => { + const raw = streamingMaxSeconds.trim(); + if (!raw) return null; + const seconds = Number(raw); + if (!Number.isInteger(seconds) || seconds < 0) return undefined; + // 0 means "never split"; anything shorter than half a minute isn't a session. + if (seconds > 0 && seconds < 30) return undefined; + return seconds; + }; + + const saveStreamingConfig = async () => { + if (!streamingEndpoint.trim() || !streamingModel.trim()) { + toast.error('Please enter the endpoint URL and model name first'); + return; + } + const maxSessionSeconds = parseMaxSessionSeconds(); + if (maxSessionSeconds === undefined) { + toast.error('Max session length must be a whole number of seconds — at least 30, or 0 to never split'); + return; + } + setIsSavingStreaming(true); + try { + await configService.saveCustomTranscriptionConfig({ + endpoint: streamingEndpoint.trim(), + model: streamingModel.trim(), + apiKey: streamingApiKey.trim() || null, + protocol: 'voxtral-realtime', + delayMs: null, + maxSessionSeconds, + }); + // Saving also activates the streaming provider on the backend; mirror + // that into the app config so the rest of the UI stays in sync. + setTranscriptModelConfig({ + ...transcriptModelConfig, + provider: 'customStreaming', + model: streamingModel.trim(), + }); + toast.success('Realtime transcription endpoint saved'); + if (onModelSelect) onModelSelect(); + } catch (err) { + toast.error(err instanceof Error ? err.message : String(err)); + } finally { + setIsSavingStreaming(false); + } + }; const modelOptions = { localWhisper: [], // Model selection handled by ModelManager component parakeet: [], // Model selection handled by ParakeetModelManager component @@ -123,6 +255,7 @@ export function TranscriptSettings({ transcriptModelConfig, setTranscriptModelCo ⚡ Parakeet (Recommended - Real-time / Accurate) 🏠 Local Whisper (High Accuracy) + 🌐 Custom Realtime (WebSocket) {/* ☁️ Deepgram (Backup) ☁️ ElevenLabs ☁️ Groq @@ -130,7 +263,7 @@ export function TranscriptSettings({ transcriptModelConfig, setTranscriptModelCo - {uiProvider !== 'localWhisper' && uiProvider !== 'parakeet' && ( + {uiProvider !== 'localWhisper' && uiProvider !== 'parakeet' && uiProvider !== 'customStreaming' && ( setStreamingEndpoint(e.target.value)} + placeholder="ws://localhost:8000/v1/realtime" + autoCapitalize="off" + autoCorrect="off" + spellCheck={false} + /> +

+ ws:// or wss:// (http/https is accepted and mapped). The default + path /v1/realtime is added when you give only a host. +

+
+ +
+ + setStreamingModel(e.target.value)} + placeholder="voxtral-mini-transcribe-realtime-2602" + autoCapitalize="off" + autoCorrect="off" + spellCheck={false} + /> +
+ +
+ +
+ setStreamingApiKey(e.target.value)} + placeholder="Leave empty if the server needs no auth" + autoCapitalize="off" + autoCorrect="off" + spellCheck={false} + /> +
+ +
+
+
+ +
+ +
+ setStreamingMaxSeconds(e.target.value)} + placeholder="300 (default)" + /> + +
+

+ A realtime server holds the whole session in one context window; + once it fills, transcription stops for the rest of the meeting. + Audio is therefore sent in batches of this length, each on a fresh + session, so long meetings keep transcribing. Detect reads the + context size from the endpoint if it publishes one — otherwise set + what your backend can take. Use 0 to never split. +

+
+ +
+ + +
+ + )} + {requiresApiKey && (
diff --git a/frontend/src/contexts/TranscriptContext.tsx b/frontend/src/contexts/TranscriptContext.tsx index df5083a437..48bd74e43e 100644 --- a/frontend/src/contexts/TranscriptContext.tsx +++ b/frontend/src/contexts/TranscriptContext.tsx @@ -7,6 +7,7 @@ import { useRecordingState } from './RecordingStateContext'; import { transcriptService } from '@/services/transcriptService'; import { recordingService } from '@/services/recordingService'; import { indexedDBService } from '@/services/indexedDBService'; +import { mergeTranscripts } from '@/lib/transcript-merge'; interface TranscriptContextType { transcripts: Transcript[]; @@ -243,33 +244,10 @@ export function TranscriptProvider({ children }: { children: ReactNode }) { const allNewTranscripts = [...sortedTranscripts, ...sortedRecentTranscripts, ...sortedStaleTranscripts, ...sortedForceFlushTranscripts]; if (allNewTranscripts.length > 0) { - setTranscripts(prev => { - // Create a set of existing sequence_ids for deduplication - const existingSequenceIds = new Set(prev.map(t => t.sequence_id).filter(id => id !== undefined)); - - // Filter out any new transcripts that already exist - const uniqueNewTranscripts = allNewTranscripts.filter(transcript => - transcript.sequence_id !== undefined && !existingSequenceIds.has(transcript.sequence_id) - ); - - // Only combine if we have unique new transcripts - if (uniqueNewTranscripts.length === 0) { - console.log('No unique transcripts to add - all were duplicates'); - return prev; // No new unique transcripts to add - } - - console.log(`Adding ${uniqueNewTranscripts.length} unique transcripts out of ${allNewTranscripts.length} received`); - - // Merge with existing transcripts, maintaining chronological order - const combined = [...prev, ...uniqueNewTranscripts]; - - // Sort by chunk_start_time first, then by sequence_id - return combined.sort((a, b) => { - const chunkTimeDiff = (a.chunk_start_time || 0) - (b.chunk_start_time || 0); - if (chunkTimeDiff !== 0) return chunkTimeDiff; - return (a.sequence_id || 0) - (b.sequence_id || 0); - }); - }); + // Upsert by sequence_id rather than discarding repeats: a streaming + // provider refines a segment in place under a stable id. See + // `mergeTranscripts` — for Whisper/Parakeet this only ever appends. + setTranscripts(prev => mergeTranscripts(prev, allNewTranscripts)); // Log the processing summary const logMessage = forceFlush @@ -296,11 +274,12 @@ export function TranscriptProvider({ children }: { children: ReactNode }) { buffer_size_before: transcriptBuffer.size }); - // Check for duplicate sequence_id before processing - if (transcriptBuffer.has(update.sequence_id)) { - console.log('🚫 MAIN LISTENER: Duplicate sequence_id, skipping buffer:', update.sequence_id); - return; - } + // NOTE: repeated sequence_ids are NOT dropped here. A streaming provider + // refines a segment in place (a live partial growing word-by-word, then + // finalizing) by re-sending the same id with newer text; the buffer + // entry below is overwritten so the latest text wins, and the state + // merge upserts by id. Whisper/Parakeet always use fresh ids, so this is + // a no-op for them. // Create transcript for buffer with NEW timestamp fields const newTranscript: Transcript = { @@ -321,19 +300,27 @@ export function TranscriptProvider({ children }: { children: ReactNode }) { transcriptBuffer.set(update.sequence_id, newTranscript); console.log(`✅ MAIN LISTENER: Buffered transcript with sequence_id ${update.sequence_id}. Buffer size: ${transcriptBuffer.size}, Last processed: ${lastProcessedSequence}`); - // Save to IndexedDB (non-blocking) - if (currentMeetingId) { + // Save settled segments to IndexedDB for reload-recovery (non-blocking). + // Skip live partials — only the finalized text needs to be recoverable, + // and streaming emits many partial updates per segment. + if (currentMeetingId && !update.is_partial) { indexedDBService.saveTranscript(currentMeetingId, update) .catch(err => console.warn('IndexedDB save failed:', err)); } - // Clear any existing timer and set a new one - if (processingTimer) { - clearTimeout(processingTimer); + // THROTTLE (not debounce): schedule a flush only if none is pending. + // A streaming provider emits partials in rapid bursts; a debounce that + // reset the timer on every update would never fire until speech paused, + // so the live segment would appear frozen on its first word and only + // update at sentence gaps. Throttling flushes the buffer at a steady + // ~10ms cadence while updates keep arriving. (Whisper's infrequent chunk + // updates behave the same either way.) + if (!processingTimer) { + processingTimer = setTimeout(() => { + processingTimer = undefined; + processBufferedTranscripts(); + }, 10); } - - // Process buffer with minimal delay for immediate UI updates (serial workers = sequential order) - processingTimer = setTimeout(processBufferedTranscripts, 10); }); console.log('✅ MAIN transcript listener setup complete'); } catch (error) { diff --git a/frontend/src/lib/transcript-merge.ts b/frontend/src/lib/transcript-merge.ts new file mode 100644 index 0000000000..c76de2f6c7 --- /dev/null +++ b/frontend/src/lib/transcript-merge.ts @@ -0,0 +1,69 @@ +import type { Transcript } from '@/types'; + +/** + * Merge incoming transcript updates into the current list, keyed by `sequence_id`. + * + * Chunk providers (Whisper, Parakeet) emit each segment once under a fresh id, so + * merging only ever appends. A streaming provider instead refines a segment **in + * place**: it re-sends the same id with longer text as the speaker talks, and once + * more when the segment finalizes. So a repeated id is an update, not a duplicate + * to discard. + * + * Returns the previous array unchanged when nothing actually differs, so React can + * skip the re-render — streaming providers re-send identical text often. + */ +export function mergeTranscripts( + prev: Transcript[], + incoming: Transcript[], +): Transcript[] { + if (incoming.length === 0) return prev; + + const bySeq = new Map(); + // Entries without a sequence_id can't be keyed, so they're carried through + // untouched and appended (they only appear in loaded/legacy history). + const seqless: Transcript[] = []; + for (const t of prev) { + if (t.sequence_id === undefined) seqless.push(t); + else bySeq.set(t.sequence_id, t); + } + + let changed = false; + for (const t of incoming) { + if (t.sequence_id === undefined) { + seqless.push(t); + changed = true; + continue; + } + const existing = bySeq.get(t.sequence_id); + if (!existing) { + bySeq.set(t.sequence_id, t); + changed = true; + continue; + } + if (existing.text === t.text && existing.is_partial === t.is_partial) { + continue; // identical resend — leave the object identity alone + } + // Keep the original id and start time: this is the same segment being + // refined, and replacing them would make React remount the row mid-sentence. + bySeq.set(t.sequence_id, { + ...existing, + text: t.text, + is_partial: t.is_partial, + confidence: t.confidence, + audio_end_time: t.audio_end_time, + duration: t.duration, + }); + changed = true; + } + + if (!changed) return prev; + + return [...bySeq.values(), ...seqless].sort(compareTranscripts); +} + +/** Chronological order: by chunk start time, then by sequence id. */ +export function compareTranscripts(a: Transcript, b: Transcript): number { + const chunkTimeDiff = (a.chunk_start_time || 0) - (b.chunk_start_time || 0); + if (chunkTimeDiff !== 0) return chunkTimeDiff; + return (a.sequence_id || 0) - (b.sequence_id || 0); +} diff --git a/frontend/src/services/configService.ts b/frontend/src/services/configService.ts index a8580f11ec..d7d05bbcbc 100644 --- a/frontend/src/services/configService.ts +++ b/frontend/src/services/configService.ts @@ -6,7 +6,7 @@ */ import { invoke } from '@tauri-apps/api/core'; -import { TranscriptModelProps } from '@/components/TranscriptSettings'; +import type { TranscriptModelProps } from '@/components/TranscriptSettings'; export interface ModelConfig { provider: 'ollama' | 'groq' | 'claude' | 'openrouter' | 'openai' | 'builtin-ai' | 'custom-openai'; @@ -41,6 +41,42 @@ export interface RecordingPreferences { preferred_system_device: string | null; } +/** + * Configuration for a self-hosted realtime (websocket) transcription endpoint, + * e.g. a vLLM server hosting Voxtral Realtime. Mirrors the backend + * `CustomTranscriptionConfig` (JSON in `transcript_settings.customTranscriptionConfig`). + */ +export interface CustomTranscriptionConfig { + /** Base URL of the endpoint (ws://, wss://, http:// or https://). */ + endpoint: string; + /** Optional bearer token (null/empty if the server needs none). */ + apiKey: string | null; + /** Model identifier requested from the server. */ + model: string; + /** Streaming protocol dialect. Defaults to "voxtral-realtime". */ + protocol: string; + /** Optional requested transcription delay (ms); protocol-specific, may be ignored. */ + delayMs: number | null; + /** + * Seconds of audio per server session before the client rolls over to a fresh + * one. `null` uses the backend default (300s), `0` disables rollover. + */ + maxSessionSeconds: number | null; +} + +/** + * What a realtime endpoint reports about how much audio one session can hold. + * Mirrors the backend `DetectedSessionLimit` (serialized snake_case). + */ +export interface DetectedSessionLimit { + /** Context window in tokens, or null when the endpoint announces none. */ + max_model_len: number | null; + /** Session length derived from that context, in seconds. */ + recommended_seconds: number | null; + /** Plain-language account of what was found. */ + detail: string; +} + /** * Configuration Service * Singleton service for managing app configuration @@ -112,6 +148,73 @@ export class ConfigService { model, }); } + + /** + * Get the custom streaming (websocket) transcription configuration. + * @returns Promise with CustomTranscriptionConfig or null if not configured + */ + async getCustomTranscriptionConfig(): Promise { + return invoke('api_get_custom_transcription_config'); + } + + /** + * Save the custom streaming transcription configuration. This also activates + * the streaming provider (sets the transcript provider to "customStreaming"). + * @param config - CustomTranscriptionConfig to save + * @returns Promise with result status + */ + async saveCustomTranscriptionConfig( + config: CustomTranscriptionConfig + ): Promise<{ status: string; message: string }> { + return invoke<{ status: string; message: string }>('api_save_custom_transcription_config', { + endpoint: config.endpoint, + model: config.model, + apiKey: config.apiKey, + protocol: config.protocol, + delayMs: config.delayMs, + maxSessionSeconds: config.maxSessionSeconds, + }); + } + + /** + * Ask the endpoint how much audio it can hold in one session (OpenAI-style + * `/v1/models` context size). Resolves with `recommended_seconds: null` when the + * server announces nothing — that is an answer, not a failure. + */ + async detectCustomTranscriptionLimits( + endpoint: string, + model: string, + apiKey: string | null, + protocol: string = 'voxtral-realtime' + ): Promise { + return invoke('api_detect_custom_transcription_limits', { + endpoint, + model, + apiKey, + protocol, + }); + } + + /** + * Test connectivity to a realtime transcription endpoint (connects, performs + * the protocol handshake — which validates the model — and disconnects). + * @returns Promise with test result; rejects with an error message on failure + */ + async testCustomTranscriptionConnection( + endpoint: string, + model: string, + apiKey: string | null, + protocol: string = 'voxtral-realtime', + delayMs: number | null = null + ): Promise<{ status: string; message: string }> { + return invoke<{ status: string; message: string }>('api_test_custom_transcription_connection', { + endpoint, + model, + apiKey, + protocol, + delayMs, + }); + } } // Export singleton instance diff --git a/frontend/tests/lib/transcript-merge.test.ts b/frontend/tests/lib/transcript-merge.test.ts new file mode 100644 index 0000000000..184aa5f788 --- /dev/null +++ b/frontend/tests/lib/transcript-merge.test.ts @@ -0,0 +1,113 @@ +import { describe, expect, test } from "bun:test"; +import { mergeTranscripts } from "../../src/lib/transcript-merge"; +import type { Transcript } from "../../src/types"; + +function seg( + sequence_id: number, + text: string, + overrides: Partial = {}, +): Transcript { + return { + id: `id-${sequence_id}`, + text, + timestamp: "14:30:05", + sequence_id, + chunk_start_time: sequence_id, + is_partial: false, + ...overrides, + }; +} + +describe("mergeTranscripts", () => { + describe("chunk providers (Whisper/Parakeet) — unchanged append-only behavior", () => { + test("appends segments with fresh sequence ids", () => { + const merged = mergeTranscripts([seg(0, "one")], [seg(1, "two")]); + expect(merged.map((t) => t.text)).toEqual(["one", "two"]); + }); + + test("orders by chunk_start_time, then sequence_id", () => { + const merged = mergeTranscripts( + [], + [ + seg(2, "third", { chunk_start_time: 9 }), + seg(0, "first", { chunk_start_time: 1 }), + seg(1, "second", { chunk_start_time: 5 }), + ], + ); + expect(merged.map((t) => t.text)).toEqual(["first", "second", "third"]); + }); + + test("an identical resend is a no-op and preserves array identity", () => { + const prev = [seg(0, "hello")]; + expect(mergeTranscripts(prev, [seg(0, "hello")])).toBe(prev); + }); + + test("empty input returns the previous array unchanged", () => { + const prev = [seg(0, "hello")]; + expect(mergeTranscripts(prev, [])).toBe(prev); + }); + }); + + describe("streaming providers — a segment refines in place", () => { + test("a repeated sequence_id updates text instead of appending", () => { + let state: Transcript[] = []; + state = mergeTranscripts(state, [seg(0, "Hello", { is_partial: true })]); + state = mergeTranscripts(state, [seg(0, "Hello world", { is_partial: true })]); + state = mergeTranscripts(state, [seg(0, "Hello world.", { is_partial: false })]); + + expect(state).toHaveLength(1); + expect(state[0].text).toBe("Hello world."); + expect(state[0].is_partial).toBe(false); + }); + + test("keeps the original id and start time while a segment grows", () => { + const prev = [seg(0, "Hel", { id: "original", chunk_start_time: 4, is_partial: true })]; + const merged = mergeTranscripts(prev, [ + seg(0, "Hello", { id: "later", chunk_start_time: 99, is_partial: true }), + ]); + + // Replacing these would remount the row mid-sentence and reorder the list. + expect(merged[0].id).toBe("original"); + expect(merged[0].chunk_start_time).toBe(4); + expect(merged[0].text).toBe("Hello"); + }); + + test("carries forward refreshed confidence and timing fields", () => { + const prev = [seg(0, "hi", { confidence: 0.1, audio_end_time: 1, duration: 1 })]; + const merged = mergeTranscripts(prev, [ + seg(0, "hi there", { confidence: 0.9, audio_end_time: 4, duration: 4 }), + ]); + + expect(merged[0].confidence).toBe(0.9); + expect(merged[0].audio_end_time).toBe(4); + expect(merged[0].duration).toBe(4); + }); + + test("a finalized segment stays put while the next one streams", () => { + let state = mergeTranscripts([], [seg(0, "One.", { is_partial: false })]); + state = mergeTranscripts(state, [seg(1, "Two", { is_partial: true })]); + state = mergeTranscripts(state, [seg(1, "Two words.", { is_partial: false })]); + + expect(state.map((t) => t.text)).toEqual(["One.", "Two words."]); + }); + + test("a partial-to-final flip with identical text still updates", () => { + const prev = [seg(0, "done.", { is_partial: true })]; + const merged = mergeTranscripts(prev, [seg(0, "done.", { is_partial: false })]); + expect(merged[0].is_partial).toBe(false); + }); + }); + + describe("transcripts without a sequence_id", () => { + test("are carried through and appended rather than dropped", () => { + const legacy: Transcript = { + id: "legacy", + text: "loaded from history", + timestamp: "14:00:00", + }; + const merged = mergeTranscripts([legacy], [seg(0, "live")]); + expect(merged.map((t) => t.text)).toContain("loaded from history"); + expect(merged).toHaveLength(2); + }); + }); +});