From 137507c6b9e2c7f936a91df2cfdebf5873cfba47 Mon Sep 17 00:00:00 2001 From: Mayu-infinite Date: Sun, 26 Jul 2026 20:28:29 +0530 Subject: [PATCH 1/2] feat(voice): end-to-end voice pipeline with IPC integration Client: microphone capture (cpal) to WAV (hound), push-to-talk (v), recording UI state, and local TTS via speech-dispatcher (spd-say) with interrupt. Runtime: VOICE_END carries the recorded path; the bridge transcribes it (Groq Whisper) and starts the normal chat pipeline via ChatPipelineStartEvent. Backend stays audio-device agnostic. Closes #18 --- client/Cargo.toml | 2 + client/src/app.rs | 72 +++++++++ client/src/audio.rs | 153 +++++++++++++++++++ client/src/ipc/client.rs | 16 ++ client/src/ipc/messages.rs | 34 ++++- client/src/main.rs | 13 +- client/src/widgets/conversation.rs | 10 ++ runtime/src/orion/transport/bridge.py | 52 ++++++- runtime/src/orion/transport/messages.py | 9 +- runtime/src/orion/transport/transcription.py | 40 +++++ runtime/tests/transport/test_bridge.py | 35 ++++- runtime/tests/transport/test_messages.py | 4 +- 12 files changed, 427 insertions(+), 13 deletions(-) create mode 100644 client/src/audio.rs create mode 100644 runtime/src/orion/transport/transcription.py diff --git a/client/Cargo.toml b/client/Cargo.toml index 1569e1f..283c2d2 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -14,3 +14,5 @@ serde_json = "1" thiserror = "1" futures-util = "0.3" tachyonfx = "=0.16.0" +cpal = "0.18.1" +hound = "3.5.1" diff --git a/client/src/app.rs b/client/src/app.rs index b0f42a9..8380062 100644 --- a/client/src/app.rs +++ b/client/src/app.rs @@ -1,6 +1,7 @@ use crossterm::event::{MouseButton, MouseEvent, MouseEventKind}; use ratatui::layout::{Position, Rect}; +use crate::audio::{Recorder, Speaker, recording_path}; use crate::effects::Effects; use crate::ipc::{client::OrionClient, events::RuntimeEvent}; use crate::theme; @@ -27,6 +28,10 @@ pub struct App { // Client pub client: Option, + // Audio — the client owns all audio hardware + pub recorder: Option, + pub speaker: Speaker, + // Store layout bounds for mouse click target checks pub prompt_area: Rect, pub events_area: Rect, @@ -51,6 +56,8 @@ impl App { cursor_position: 0, msg_counter: 0, client: None, + recorder: None, + speaker: Speaker::new(), prompt_area: Rect::default(), events_area: Rect::default(), events: EventLog::new(), @@ -156,6 +163,69 @@ impl App { self.mode = "THINKING".into(); } + // --- Voice (push-to-talk) ------------------------------------------ + + /// Toggle recording: press once to start, again to stop and send. + pub async fn toggle_recording(&mut self) { + if self.recorder.is_some() { + self.stop_recording().await; + } else { + self.start_recording().await; + } + } + + async fn start_recording(&mut self) { + self.speaker.stop(); // a new recording interrupts any speech + + match Recorder::start(recording_path()) { + Ok(recorder) => { + let sample_rate = recorder.sample_rate(); + let channels = recorder.channels() as u32; + self.recorder = Some(recorder); + self.mode = "RECORDING".into(); + self.effects.on_status_change(theme::DANGER); + self.events + .push("VOICE_START", EventStatus::Running, "recording"); + + if let Some(client) = &mut self.client { + let _ = client.send_voice_start(sample_rate, channels).await; + } + } + Err(err) => { + self.mode = format!("MIC ERROR: {}", err); + self.events.push("VOICE_START", EventStatus::Failed, err); + } + } + } + + async fn stop_recording(&mut self) { + let Some(recorder) = self.recorder.take() else { + return; + }; + + match recorder.finish() { + Ok(path) => { + let path = path.to_string_lossy().to_string(); + self.mode = "TRANSCRIBING".into(); + self.events + .push("VOICE_END", EventStatus::Completed, path.clone()); + + if let Some(client) = &mut self.client { + let _ = client.send_voice_end(path).await; + } + } + Err(err) => { + self.mode = format!("REC ERROR: {}", err); + self.events.push("VOICE_END", EventStatus::Failed, err); + } + } + } + + /// Interrupt any assistant speech currently playing. + pub fn interrupt_speech(&self) { + self.speaker.stop(); + } + pub fn handle_runtime_event(&mut self, event: RuntimeEvent) { match event { RuntimeEvent::Connected => self.on_connected(), @@ -191,6 +261,8 @@ impl App { self.conversation.finish_assistant_message(); self.events .bump_last("RESPONSE", EventStatus::Completed, "response complete"); + // Client-side TTS: speak the completed response. + self.speaker.speak(&self.conversation.last_assistant_text()); } RuntimeEvent::ToolStarted { name } => { diff --git a/client/src/audio.rs b/client/src/audio.rs new file mode 100644 index 0000000..c078380 --- /dev/null +++ b/client/src/audio.rs @@ -0,0 +1,153 @@ +//! Client-side audio: microphone capture and speech output. +//! +//! Per the architecture, the *client* owns all audio hardware. [`Recorder`] +//! captures the microphone to a 16-bit PCM WAV (via `cpal` + `hound`); the app +//! sends that file's path to the runtime, which transcribes it. [`Speaker`] +//! speaks the assistant's reply through the system speech engine by shelling +//! out to `spd-say` (speech-dispatcher) — no build-time audio library required, +//! and it degrades to silence if speech-dispatcher isn't installed. + +use std::path::PathBuf; +use std::process::Command; +use std::sync::{Arc, Mutex}; + +use cpal::traits::{DeviceTrait, HostTrait, StreamTrait}; + +/// Default location for the recorded clip (shared with the runtime, same host). +pub fn recording_path() -> PathBuf { + std::env::temp_dir().join("orion_client_recording.wav") +} + +/// An in-progress microphone recording. Dropping/finishing it stops capture. +pub struct Recorder { + stream: cpal::Stream, + samples: Arc>>, + sample_rate: u32, + channels: u16, + path: PathBuf, +} + +impl Recorder { + /// Open the default input device and start capturing into memory. + pub fn start(path: PathBuf) -> Result { + let host = cpal::default_host(); + let device = host + .default_input_device() + .ok_or_else(|| "no input device".to_string())?; + let supported = device.default_input_config().map_err(|e| e.to_string())?; + + let sample_rate = supported.sample_rate(); + let channels = supported.channels(); + let sample_format = supported.sample_format(); + let config: cpal::StreamConfig = supported.into(); + + let samples = Arc::new(Mutex::new(Vec::::new())); + let sink = samples.clone(); + let err_fn = |err| eprintln!("audio input error: {err}"); + + let stream = match sample_format { + cpal::SampleFormat::I16 => device.build_input_stream( + config, + move |data: &[i16], _: &cpal::InputCallbackInfo| { + if let Ok(mut buf) = sink.lock() { + buf.extend_from_slice(data); + } + }, + err_fn, + None, + ), + cpal::SampleFormat::F32 => device.build_input_stream( + config, + move |data: &[f32], _: &cpal::InputCallbackInfo| { + if let Ok(mut buf) = sink.lock() { + buf.extend(data.iter().map(|s| (s.clamp(-1.0, 1.0) * 32767.0) as i16)); + } + }, + err_fn, + None, + ), + other => return Err(format!("unsupported sample format: {other:?}")), + } + .map_err(|e| e.to_string())?; + + stream.play().map_err(|e| e.to_string())?; + + Ok(Self { + stream, + samples, + sample_rate, + channels, + path, + }) + } + + pub fn sample_rate(&self) -> u32 { + self.sample_rate + } + + pub fn channels(&self) -> u16 { + self.channels + } + + /// Stop capture, write the WAV, and return its path. + pub fn finish(self) -> Result { + let Recorder { + stream, + samples, + sample_rate, + channels, + path, + } = self; + + drop(stream); // stop the input stream + + let samples = samples + .lock() + .map_err(|_| "audio buffer poisoned".to_string())?; + + let spec = hound::WavSpec { + channels, + sample_rate, + bits_per_sample: 16, + sample_format: hound::SampleFormat::Int, + }; + + let mut writer = hound::WavWriter::create(&path, spec).map_err(|e| e.to_string())?; + for &sample in samples.iter() { + writer.write_sample(sample).map_err(|e| e.to_string())?; + } + writer.finalize().map_err(|e| e.to_string())?; + + Ok(path) + } +} + +/// Speaks assistant responses through speech-dispatcher's `spd-say` CLI. +pub struct Speaker; + +impl Speaker { + pub fn new() -> Self { + Self + } + + /// Speak `text`, cancelling anything currently being spoken first. + pub fn speak(&self, text: &str) { + let text = text.trim(); + if text.is_empty() { + return; + } + // `-C` cancels current speech, so a new reply interrupts an old one. + let _ = Command::new("spd-say").arg("-C").arg(text).spawn(); + } + + /// Stop any in-progress speech. + pub fn stop(&self) { + let _ = Command::new("spd-say").arg("--cancel").spawn(); + } +} + +impl Default for Speaker { + fn default() -> Self { + Self::new() + } +} diff --git a/client/src/ipc/client.rs b/client/src/ipc/client.rs index 96f24da..eb5fc36 100644 --- a/client/src/ipc/client.rs +++ b/client/src/ipc/client.rs @@ -31,6 +31,22 @@ impl OrionClient { self.session.send(&envelope).await } + /// Announce the start of a voice recording (metadata only). + pub async fn send_voice_start( + &mut self, + sample_rate: u32, + channels: u32, + ) -> Result<(), IpcError> { + self.session + .send(&Envelope::voice_start(sample_rate, channels)) + .await + } + + /// Finish a recording by sending the runtime the recorded file path. + pub async fn send_voice_end(&mut self, path: impl Into) -> Result<(), IpcError> { + self.session.send(&Envelope::voice_end(path)).await + } + /// Send a ping message. pub async fn ping(&mut self) -> Result<(), IpcError> { self.session.send(&Envelope::ping()).await diff --git a/client/src/ipc/messages.rs b/client/src/ipc/messages.rs index e1f17f7..f2b9f99 100644 --- a/client/src/ipc/messages.rs +++ b/client/src/ipc/messages.rs @@ -90,6 +90,33 @@ impl Envelope { } } + /// Announce the start of a voice recording session (metadata only). + pub fn voice_start(sample_rate: u32, channels: u32) -> Self { + Self { + version: 1, + id: Uuid::new_v4(), + correlation_id: Uuid::new_v4(), + message_type: MessageType::VoiceStart, + payload: serde_json::to_value(VoiceStartPayload { + sample_rate, + channels, + encoding: "pcm16".to_string(), + }) + .unwrap(), + } + } + + /// Finish a recording by handing the runtime the recorded file path. + pub fn voice_end(path: impl Into) -> Self { + Self { + version: 1, + id: Uuid::new_v4(), + correlation_id: Uuid::new_v4(), + message_type: MessageType::VoiceEnd, + payload: serde_json::to_value(VoiceEndPayload { path: path.into() }).unwrap(), + } + } + /// Deserialize the payload into a strongly typed struct. pub fn payload(&self) -> serde_json::Result where @@ -126,8 +153,11 @@ pub struct VoiceChunkPayload { pub data: Vec, } -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct VoiceEndPayload; +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct VoiceEndPayload { + /// Path to the recorded audio file the runtime should transcribe. + pub path: String, +} #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct AssistantStartPayload; diff --git a/client/src/main.rs b/client/src/main.rs index 474a43d..944918f 100644 --- a/client/src/main.rs +++ b/client/src/main.rs @@ -1,4 +1,5 @@ mod app; +mod audio; mod effects; mod ipc; mod theme; @@ -24,7 +25,9 @@ const SOCKET_PATH: &str = "/tmp/orion.sock"; // ~30 FPS so tachyonfx effects animate smoothly instead of stepping. const TICK_RATE: Duration = Duration::from_millis(33); -#[tokio::main] +// Single-threaded runtime: the audio input stream is `!Send`, so the app +// state (which owns it) must stay on one thread. +#[tokio::main(flavor = "current_thread")] async fn main() -> Result<(), Box> { // Setup terminal enable_raw_mode()?; @@ -87,6 +90,14 @@ async fn main() -> Result<(), Box> { KeyCode::Char('q') | KeyCode::Esc => { app.should_quit = true; } + // Push-to-talk: toggle voice recording + KeyCode::Char('v') => { + app.toggle_recording().await; + } + // Interrupt assistant speech + KeyCode::Char('s') => { + app.interrupt_speech(); + } // Conversation Scrolling (Vim keys & standard navigation) KeyCode::Up | KeyCode::Char('k') => { app.conversation.scroll_up(1); diff --git a/client/src/widgets/conversation.rs b/client/src/widgets/conversation.rs index a618437..c0bb999 100644 --- a/client/src/widgets/conversation.rs +++ b/client/src/widgets/conversation.rs @@ -161,6 +161,16 @@ impl ConversationWidget { self.scroll_offset = 0; } + /// Text of the most recent assistant message (for TTS), or empty. + pub fn last_assistant_text(&self) -> String { + self.messages + .iter() + .rev() + .find(|m| m.author == Author::Orion) + .map(|m| m.content.clone()) + .unwrap_or_default() + } + pub fn render(&mut self, frame: &mut Frame, area: Rect) { // Outer Panel Block let outer_block = Block::default() diff --git a/runtime/src/orion/transport/bridge.py b/runtime/src/orion/transport/bridge.py index aa5e36d..1485300 100644 --- a/runtime/src/orion/transport/bridge.py +++ b/runtime/src/orion/transport/bridge.py @@ -14,6 +14,8 @@ from uuid import UUID +from collections.abc import Awaitable, Callable + from orion.bus.event_bus import EventBus from orion.events.events import ChatPipelineStartEvent from orion.transport.messages import ( @@ -21,8 +23,13 @@ MessageType, PongPayload, SubmitPromptPayload, + VoiceEndPayload, ) from orion.transport.session import ClientSession +from orion.transport.transcription import transcribe_audio + +#: Async callable turning a recorded audio path into transcript text. +Transcriber = Callable[[str], Awaitable[str]] class IPCBridge: @@ -30,8 +37,13 @@ class IPCBridge: Bridges IPC protocol messages and Orion domain events. """ - def __init__(self, event_bus: EventBus) -> None: + def __init__( + self, + event_bus: EventBus, + transcriber: Transcriber = transcribe_audio, + ) -> None: self._event_bus = event_bus + self._transcriber = transcriber self._sessions: dict[UUID, ClientSession] = {} # ------------------------------------------------------------------ @@ -92,13 +104,15 @@ async def handle( await self._handle_submit_prompt(session, envelope) case MessageType.VOICE_START: - ... + # Session metadata only; the path-based flow acts on VOICE_END. + pass case MessageType.VOICE_CHUNK: - ... + # Streamed audio is a future enhancement. + pass case MessageType.VOICE_END: - ... + await self._handle_voice_end(session, envelope) case _: raise ValueError(f"Unsupported IPC message: {envelope.type}") @@ -141,6 +155,36 @@ async def _handle_submit_prompt( ) ) + async def _handle_voice_end( + self, + session: ClientSession, + envelope: Envelope, + ) -> None: + """ + Transcribe a recorded voice message and start the chat pipeline. + + The client sends the path to the audio it recorded; the runtime + transcribes it and drives the same pipeline as a typed prompt. + """ + + payload = VoiceEndPayload.model_validate(envelope.payload) + + transcript = (await self._transcriber(payload.path)).strip() + + # Ignore empty transcriptions (silence / failed capture). + if not transcript: + return + + await self._event_bus.publish( + ChatPipelineStartEvent( + correlation_id=envelope.correlation_id, + session_id=session.id, + source="ipc", + message="Voice prompt transcribed via IPC.", + text=transcript, + ) + ) + # ------------------------------------------------------------------ # Session Lifecycle # ------------------------------------------------------------------ diff --git a/runtime/src/orion/transport/messages.py b/runtime/src/orion/transport/messages.py index ee92aba..7d3570e 100644 --- a/runtime/src/orion/transport/messages.py +++ b/runtime/src/orion/transport/messages.py @@ -152,7 +152,14 @@ class VoiceChunkPayload(BaseModel): class VoiceEndPayload(BaseModel): - """Marks the end of a streamed voice recording.""" + """Marks the end of a voice recording. + + Carries the path to the file the client recorded; the runtime reads and + transcribes it (the preferred, non-streamed flow). + """ + + #: Filesystem path to the recorded audio, as seen by the runtime. + path: str # ====================================================================== diff --git a/runtime/src/orion/transport/transcription.py b/runtime/src/orion/transport/transcription.py new file mode 100644 index 0000000..f942dd6 --- /dev/null +++ b/runtime/src/orion/transport/transcription.py @@ -0,0 +1,40 @@ +""" +Audio transcription for the voice pipeline. + +The client records audio and sends the runtime a file path over IPC; the +runtime transcribes it here (Groq Whisper) and feeds the text into the normal +chat pipeline. The backend never touches a microphone — it only reads the file +the client produced. +""" + +from __future__ import annotations + +import asyncio +import os + +from groq import Groq + +#: Groq speech-to-text model used for transcription. +_MODEL = "whisper-large-v3-turbo" + + +async def transcribe_audio(path: str) -> str: + """ + Transcribe the audio file at ``path`` to text. + + Runs the blocking Groq call in a worker thread so the event loop stays + responsive. + """ + return await asyncio.to_thread(_transcribe_sync, path) + + +def _transcribe_sync(path: str) -> str: + client = Groq(api_key=os.getenv("GROQ_API_KEY")) + + with open(path, "rb") as audio_file: + result = client.audio.transcriptions.create( + file=audio_file, + model=_MODEL, + ) + + return result.text diff --git a/runtime/tests/transport/test_bridge.py b/runtime/tests/transport/test_bridge.py index 559789b..6af5335 100644 --- a/runtime/tests/transport/test_bridge.py +++ b/runtime/tests/transport/test_bridge.py @@ -190,14 +190,43 @@ async def test_voice_chunk_not_implemented() -> None: @pytest.mark.asyncio -async def test_voice_end_not_implemented() -> None: +async def test_voice_end_transcribes_and_starts_pipeline() -> None: bus = FakeEventBus() - bridge = IPCBridge(bus) + + async def fake_transcriber(path: str) -> str: + assert path == "/tmp/orion/input.wav" + return " what is the system status " + + bridge = IPCBridge(bus, transcriber=fake_transcriber) session = FakeSession() message = Envelope( type=MessageType.VOICE_END, - payload={}, + payload={"path": "/tmp/orion/input.wav"}, + ) + + await bridge.handle(session, message) + + assert len(bus.events) == 1 + event = bus.events[0] + assert isinstance(event, ChatPipelineStartEvent) + assert event.text == "what is the system status" + assert event.correlation_id == message.correlation_id + + +@pytest.mark.asyncio +async def test_voice_end_ignores_empty_transcript() -> None: + bus = FakeEventBus() + + async def fake_transcriber(_path: str) -> str: + return " " + + bridge = IPCBridge(bus, transcriber=fake_transcriber) + session = FakeSession() + + message = Envelope( + type=MessageType.VOICE_END, + payload={"path": "/tmp/orion/input.wav"}, ) await bridge.handle(session, message) diff --git a/runtime/tests/transport/test_messages.py b/runtime/tests/transport/test_messages.py index 3017cf0..4df0b0e 100644 --- a/runtime/tests/transport/test_messages.py +++ b/runtime/tests/transport/test_messages.py @@ -57,9 +57,9 @@ def test_voice_chunk_payload() -> None: def test_voice_end_payload() -> None: - payload = VoiceEndPayload() + payload = VoiceEndPayload(path="/tmp/orion/input.wav") - assert payload.model_dump() == {} + assert payload.model_dump() == {"path": "/tmp/orion/input.wav"} def test_assistant_chunk_payload() -> None: From 9d5a7bd426ea01ff4336aa48be2dadfc87d10ef7 Mon Sep 17 00:00:00 2001 From: Mayu-infinite Date: Sun, 26 Jul 2026 20:32:08 +0530 Subject: [PATCH 2/2] feat(voice): end-to-end voice pipeline with IPC integration Client: microphone capture (cpal) to WAV (hound), push-to-talk (v), recording UI state, and local TTS via speech-dispatcher (spd-say) with interrupt. Runtime: VOICE_END carries the recorded path; the bridge transcribes it (Groq Whisper) and starts the normal chat pipeline via ChatPipelineStartEvent. Backend stays audio-device agnostic. Closes #18 --- .github/workflows/pytest.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml index 0f2529b..9ac913c 100644 --- a/.github/workflows/pytest.yml +++ b/.github/workflows/pytest.yml @@ -77,5 +77,11 @@ jobs: - name: Check out repository uses: actions/checkout@v4 + - name: Install audio build dependencies + run: sudo apt-get update && sudo apt-get install -y libasound2-dev + - name: Build client run: cargo build --verbose + + - name: Run client tests + run: cargo test --verbose