diff --git a/Cargo.lock b/Cargo.lock index 7d6912b..79c4d59 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -633,6 +633,22 @@ version = "2.8.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + +[[package]] +name = "mime_guess" +version = "2.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f7c44f8e672c00fe5308fa235f821cb4198414e1c77935c1ab6948d3fd78550e" +dependencies = [ + "mime", + "unicase", +] + [[package]] name = "mio" version = "1.2.1" @@ -849,6 +865,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "http", "http-body", "http-body-util", @@ -857,6 +874,7 @@ dependencies = [ "hyper-util", "js-sys", "log", + "mime_guess", "percent-encoding", "pin-project-lite", "quinn", @@ -1396,6 +1414,12 @@ version = "1.20.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" +[[package]] +name = "unicase" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dbc4bc3a9f746d862c45cb89d705aa10f187bb96c76001afab07a0d35ce60142" + [[package]] name = "unicode-ident" version = "1.0.24" diff --git a/Cargo.toml b/Cargo.toml index e57e366..56c3334 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -12,7 +12,7 @@ path = "src/main.rs" [dependencies] tokio = { version = "1", features = ["rt-multi-thread", "macros", "process", "time", "sync", "signal"] } -reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls"] } +reqwest = { version = "0.12", default-features = false, features = ["json", "multipart", "rustls-tls"] } rusqlite = { version = "0.32", features = ["bundled"] } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/README.md b/README.md index 0f2848e..a5be99d 100644 --- a/README.md +++ b/README.md @@ -75,6 +75,7 @@ and the rest of your day. ## What works today - iMessage on macOS and Telegram private chats on macOS or Linux +- Telegram voice notes with OpenAI transcription and spoken replies - Claude Code, Codex, and Pi backends, selectable by channel or conversation - One Git-versioned assistant repository containing `SOUL.md`, `context/`, and approved `jobs/` diff --git a/docs/architecture.md b/docs/architecture.md index e6be597..67f7520 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -223,6 +223,26 @@ One provider can fail or rate-limit without cancelling the other. A shared shutdown signal cancels pending polls and lets every channel drain its workers. Replies remain bound to the originating loop and exact target. +## Voice Boundary + +Voice has two independent adapters. A channel adapter exposes opaque inbound +voice metadata, downloads bytes only after the normal allowlist accepts the +message, and uploads a generated audio clip. The provider-neutral voice layer +transcribes and synthesizes in-memory audio. Its output is plain text or a +generic audio clip, so it has no Telegram identifiers or Bot API behavior. +Download and transcription run in the accepted message's per-thread worker, +so slow audio cannot pause polling or work in another conversation. + +The first provider uses `OPENAI_API_KEY` for both `gpt-4o-transcribe` and +`gpt-4o-mini-tts`. The first channel implementation is Telegram. A future +channel needs only voice download and upload support. It does not need to +change gateway routing, agent requests, or the OpenAI client. + +Voice is an optional enhancement. Missing credentials stop voice processing +for that message with a text explanation, while ordinary text traffic remains +available. Speech synthesis and voice delivery happen after durable text +delivery, so a voice-side failure cannot discard an agent answer. + Optional `primary_delivery` selects one enabled, allowlisted channel target for proactive output. Resolution is lazy: an absent or invalid primary returns a scoped error to the proactive caller without affecting reply polling. diff --git a/docs/configuration.md b/docs/configuration.md index 9847036..1ca6c73 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -91,6 +91,11 @@ private. `push init` creates new config files with mode `0600` on Unix. The `bot_token_env` setting remains available when an environment variable is a better fit. See the [Telegram guide](telegram.md). +Telegram voice notes are optional and need no TOML settings. Set +`OPENAI_API_KEY` in the gateway process environment to enable both +transcription and speech replies. Without it, text remains fully available and +voice notes get a helpful fallback. See [Voice Messages](telegram.md#voice-messages). + ### Run both providers Use `channels` instead of `channel`: diff --git a/docs/services.md b/docs/services.md index 2b66bad..3befc86 100644 --- a/docs/services.md +++ b/docs/services.md @@ -31,6 +31,8 @@ Use absolute paths in service files. The service user needs: - for iMessage on macOS, Full Disk Access and `osascript` - for Telegram, a token in the private config and network access to `api.telegram.org` +- for optional voice messages, `OPENAI_API_KEY` in the service environment and + network access to `api.openai.com` `state_path` stores independent cursors for each channel and backend session ids. `database_path` stores the canonical conversation journal. Chat agents run @@ -76,6 +78,8 @@ and replace `YOU` with your macOS user name: PATH /Users/YOU/.local/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin + OPENAI_API_KEY + replace-with-your-openai-api-key RunAtLoad @@ -109,6 +113,10 @@ launchctl bootstrap gui/$(id -u) ~/Library/LaunchAgents/com.owainlewis.push.plis launchctl kickstart -k gui/$(id -u)/com.owainlewis.push ``` +The plist contains the optional OpenAI key, so protect it with +`chmod 600 ~/Library/LaunchAgents/com.owainlewis.push.plist`. Omit the key when +you do not want voice support. + ## Linux systemd Use this for Telegram-only deployments. The iMessage channel still requires @@ -136,6 +144,7 @@ WorkingDirectory=%h/.push Restart=on-failure RestartSec=10 Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +EnvironmentFile=-%h/.config/push/env [Install] WantedBy=default.target @@ -150,6 +159,14 @@ systemctl --user status push.service journalctl --user -u push.service -f ``` +For voice support, create the optional private environment file: + +```sh +printf 'OPENAI_API_KEY=replace-with-your-openai-api-key\n' > ~/.config/push/env +chmod 600 ~/.config/push/env +systemctl --user restart push.service +``` + Keep `~/.push/config.toml` at mode `0600` because it contains the Telegram bot token. Do not commit this file or print it in service logs. diff --git a/docs/telegram.md b/docs/telegram.md index 09c6467..2254c0b 100644 --- a/docs/telegram.md +++ b/docs/telegram.md @@ -50,14 +50,41 @@ Doctor validates that a token is available without displaying its value. A Telegram-only preflight does not open `chat.db` and does not require macOS or `osascript`. +## Voice Messages + +Set one optional environment variable to enable both incoming voice +transcription and spoken replies: + +```sh +export OPENAI_API_KEY="your-api-key" +push doctor +push +``` + +Send the bot a Telegram voice note. Push downloads it only after the sender +passes the normal allowlist, transcribes it with `gpt-4o-transcribe`, sends the +transcript through the selected coding agent, then returns the answer as both +text and an Opus voice note generated by `gpt-4o-mini-tts`. + +There is no local Whisper, FFmpeg, or other audio dependency. Audio stays in +memory and is limited to 20 MB. If `OPENAI_API_KEY` is absent, normal text +messages keep working and voice notes receive an actionable text reply. API or +speech generation errors also fall back to text without stopping the gateway. +The spoken reply is AI-generated. Voice-note audio is sent to OpenAI for +processing, so review OpenAI's data controls before enabling this feature. + +Voice processing is separate from the Telegram adapter. Telegram is the first +channel to provide voice attachment download and upload, so another messaging +channel can add those transport operations without changing the OpenAI layer. + ## Allowlisting and Routing An incoming Telegram update reaches the agent only when all of these are true: -- it is a normal text message in a private chat +- it is a normal text message or voice note in a private chat - its numeric sender id is in `telegram.allow_user_ids`, or its numeric chat id is in `telegram.allow_chat_ids` -- the text is not empty +- the message contains non-empty text or a voice attachment Group chats, channels, group forum topics, edited messages, and other update types are out of scope and ignored. The private-chat thread key is diff --git a/examples/launchd/com.owainlewis.push.plist b/examples/launchd/com.owainlewis.push.plist index 121abfe..8bb0919 100644 --- a/examples/launchd/com.owainlewis.push.plist +++ b/examples/launchd/com.owainlewis.push.plist @@ -20,6 +20,9 @@ PATH /Users/YOU/.local/bin:/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin + + + RunAtLoad diff --git a/examples/systemd/push.service b/examples/systemd/push.service index c7c2a1a..bf3d2e3 100644 --- a/examples/systemd/push.service +++ b/examples/systemd/push.service @@ -10,6 +10,7 @@ WorkingDirectory=%h/.push Restart=on-failure RestartSec=10 Environment=PATH=%h/.local/bin:/usr/local/bin:/usr/bin:/bin +EnvironmentFile=-%h/.config/push/env [Install] WantedBy=default.target diff --git a/src/audit.rs b/src/audit.rs index 396b099..323f842 100644 --- a/src/audit.rs +++ b/src/audit.rs @@ -296,6 +296,7 @@ mod tests { chat_identifier: "+15551234567".to_string(), is_group: false, text: "secret request".to_string(), + voice: None, is_from_me: false, is_supported: true, thread_id: None, diff --git a/src/channel.rs b/src/channel.rs index b676f3f..46c2080 100644 --- a/src/channel.rs +++ b/src/channel.rs @@ -7,6 +7,18 @@ use anyhow::{bail, Context, Result}; use crate::config::{ChannelKind, Config}; use crate::imessage::{Poller as IMessagePoller, Sender as IMessageSender}; use crate::telegram::Telegram; +use crate::voice::AudioClip; + +#[derive(Debug, Clone)] +pub struct InboundVoice { + /// Channel-owned file identifier. The voice layer treats this as opaque. + pub locator: String, + pub file_size: Option, + pub mime_type: String, + pub filename: String, + /// Channels that already have the bytes may provide them directly. + pub data: Option>, +} #[derive(Debug, Clone)] pub struct RawMessage { @@ -16,6 +28,7 @@ pub struct RawMessage { pub chat_identifier: String, pub is_group: bool, pub text: String, + pub voice: Option, pub is_from_me: bool, pub is_supported: bool, /// Channel-specific thread/topic id (Telegram `message_thread_id`). @@ -150,6 +163,7 @@ impl Channel { chat_identifier: message.chat_identifier, is_group: message.is_group, text: message.text, + voice: None, is_from_me: message.is_from_me, is_supported: true, thread_id: None, @@ -172,7 +186,10 @@ impl Channel { /// Returns `(thread_key, reply_target)` for an accepted message. pub fn accept(&self, message: &RawMessage) -> Option<(String, String)> { - if !message.is_supported || message.is_group || message.text.trim().is_empty() { + if !message.is_supported + || message.is_group + || (message.text.trim().is_empty() && message.voice.is_none()) + { return None; } match self { @@ -222,8 +239,8 @@ impl Channel { "unsupported_update" } else if message.is_group { "group_chat" - } else if message.text.trim().is_empty() { - "empty_text" + } else if message.text.trim().is_empty() && message.voice.is_none() { + "empty_message" } else { match self { Self::IMessage { reply_marker, .. } @@ -283,6 +300,28 @@ impl Channel { Self::Telegram(telegram) => telegram.send_typing(target).await, } } + + pub async fn download_voice(&self, voice: &InboundVoice) -> Result { + if let Some(bytes) = &voice.data { + return Ok(AudioClip { + bytes: bytes.clone(), + filename: voice.filename.clone(), + mime_type: voice.mime_type.clone(), + }); + } + match self { + Self::Telegram(telegram) => telegram.download_voice(voice).await, + Self::IMessage { .. } => bail!("iMessage voice messages are not supported yet"), + } + } + + #[cfg_attr(test, allow(dead_code))] + pub async fn send_voice(&self, target: &str, clip: &AudioClip) -> Result<()> { + match self { + Self::Telegram(telegram) => telegram.send_voice(target, clip).await, + Self::IMessage { .. } => bail!("iMessage voice replies are not supported yet"), + } + } } pub(crate) fn normalize_handle(value: &str) -> String { @@ -331,6 +370,7 @@ mod tests { chat_identifier: chat.to_string(), is_group, text: "hello".to_string(), + voice: None, is_from_me: false, is_supported: true, thread_id: None, diff --git a/src/doctor.rs b/src/doctor.rs index d9048ed..fe1544a 100644 --- a/src/doctor.rs +++ b/src/doctor.rs @@ -75,9 +75,24 @@ fn run_checks(cfg: &config::Config) -> CheckReport { Err(e) => checks.push(Check::fail("channels", e.to_string())), } check_bins(cfg, &mut checks); + check_voice(&mut checks); CheckReport { checks } } +fn check_voice(checks: &mut Vec) { + if std::env::var(crate::voice::OPENAI_API_KEY_ENV).is_ok_and(|value| !value.trim().is_empty()) { + checks.push(Check::pass( + "voice messages", + "OPENAI_API_KEY is set; transcription and speech replies are enabled", + )); + } else { + checks.push(Check::pass( + "voice messages", + "OPENAI_API_KEY is not set; text messaging works and voice requests get a helpful fallback", + )); + } +} + fn check_config(cfg: &config::Config, checks: &mut Vec) { checks.push(Check::pass( "config", diff --git a/src/gateway/mod.rs b/src/gateway/mod.rs index bc8073d..da2154c 100644 --- a/src/gateway/mod.rs +++ b/src/gateway/mod.rs @@ -19,18 +19,22 @@ use tracing::{error, info, warn}; use crate::agent::Runner; use crate::approval::{AnswerOrigin, AnswerOutcome}; use crate::audit::{AuditEvent, AuditLog}; -use crate::channel::{Channel, RawMessage}; +use crate::channel::{Channel, InboundVoice, RawMessage}; use crate::config::{AgentBackend, ChannelKind, Config, PrimaryDeliveryConfig}; use crate::history::{History, OutboundOrigin}; use crate::jobs; use crate::store::Store; use crate::util::now_ms; +use crate::voice::Voice; const QUEUE_DEPTH: usize = 32; #[cfg(not(test))] const SEND_TIMEOUT: Duration = Duration::from_secs(30); const SHUTDOWN_GRACE: Duration = Duration::from_secs(30); +#[cfg(test)] +type SentVoiceReplies = Arc)>>>; + #[derive(Clone)] struct Job { row_id: i64, @@ -40,6 +44,8 @@ struct Job { backend: AgentBackend, approval_origin: AnswerOrigin, text: String, + reply_with_voice: bool, + voice_attachment: Option, } /// Shared, cheaply cloneable context handed to each worker task. @@ -55,11 +61,14 @@ struct Ctx { reply_marker: String, assistant_dir: String, audit: Arc, + voice: Option, #[cfg(test)] setup_failure_replies: Arc>>, #[cfg(test)] sent_replies: Arc>>, #[cfg(test)] + sent_voice_replies: SentVoiceReplies, + #[cfg(test)] send_failures_remaining: Arc>, } @@ -341,11 +350,17 @@ impl Gateway { reply_marker: cfg.reply_marker.clone(), assistant_dir: cfg.assistant_dir.clone(), audit, + #[cfg(not(test))] + voice: Voice::from_env(), + #[cfg(test)] + voice: None, #[cfg(test)] setup_failure_replies: Arc::new(Mutex::new(Vec::new())), #[cfg(test)] sent_replies: Arc::new(Mutex::new(Vec::new())), #[cfg(test)] + sent_voice_replies: Arc::new(Mutex::new(Vec::new())), + #[cfg(test)] send_failures_remaining: Arc::new(Mutex::new(0)), }; let poll_interval = cfg.poll_interval_dur()?; @@ -493,12 +508,22 @@ impl Gateway { continue; } if let Some((thread, target)) = self.channel.accept(m) { + let reply_with_voice = m.voice.is_some(); + let message_text = if reply_with_voice { + "[Voice message]".to_string() + } else { + m.text.trim().to_string() + }; let approval_origin = approval_origin(m, &thread); - let approval = self.ctx.history.lock().unwrap().answer_question( - &approval_origin, - m.text.trim(), - now_ms(), - ); + let approval = if reply_with_voice { + Ok(AnswerOutcome::NotAnAnswer) + } else { + self.ctx.history.lock().unwrap().answer_question( + &approval_origin, + message_text.trim(), + now_ms(), + ) + }; match approval { Ok(AnswerOutcome::NotAnAnswer) => {} Ok(outcome @ (AnswerOutcome::Selected(_) | AnswerOutcome::Duplicate(_))) => { @@ -546,7 +571,7 @@ impl Gateway { m.channel, &thread, &m.event_id(), - m.text.trim(), + message_text.trim(), ) { Ok(id) => id, Err(error) => { @@ -593,7 +618,9 @@ impl Gateway { target, backend, approval_origin, - text: m.text.trim().to_string(), + text: message_text, + reply_with_voice, + voice_attachment: m.voice.clone(), }; if !self.route(job).await { return; diff --git a/src/gateway/tests.rs b/src/gateway/tests.rs index 40324ae..ef5e23c 100644 --- a/src/gateway/tests.rs +++ b/src/gateway/tests.rs @@ -5,13 +5,56 @@ use super::worker::{ use super::*; use crate::agent::{FakeRunCall, FakeRunner}; use crate::approval::Question; -use crate::channel::{normalize_handle, thread_handle}; +use crate::channel::{normalize_handle, thread_handle, InboundVoice}; use crate::history::DeliveryStatus; use crate::imessage::{Poller, Sender}; +use crate::voice::{AudioClip, VoiceFuture, VoiceProvider}; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use uuid::Uuid; +struct FakeVoice; + +impl VoiceProvider for FakeVoice { + fn transcribe<'a>(&'a self, _clip: AudioClip) -> VoiceFuture<'a, String> { + Box::pin(async { Ok("voice request".to_string()) }) + } + + fn synthesize<'a>(&'a self, _text: &'a str) -> VoiceFuture<'a, AudioClip> { + Box::pin(async { + Ok(AudioClip { + bytes: vec![4, 5, 6], + filename: "reply.opus".to_string(), + mime_type: "audio/ogg".to_string(), + }) + }) + } +} + +struct BlockingVoice { + release: tokio::sync::Mutex>>, +} + +impl VoiceProvider for BlockingVoice { + fn transcribe<'a>(&'a self, _clip: AudioClip) -> VoiceFuture<'a, String> { + Box::pin(async move { + let release = self.release.lock().await.take().unwrap(); + release.await.unwrap(); + Ok("slow voice request".to_string()) + }) + } + + fn synthesize<'a>(&'a self, _text: &'a str) -> VoiceFuture<'a, AudioClip> { + Box::pin(async { + Ok(AudioClip { + bytes: vec![7], + filename: "reply.opus".to_string(), + mime_type: "audio/ogg".to_string(), + }) + }) + } +} + #[tokio::test] async fn periodic_activity_refreshes_until_operation_completes() { let (complete, completed) = tokio::sync::oneshot::channel(); @@ -118,6 +161,7 @@ fn msg(chat: &str, handle: &str, from_me: bool, text: &str) -> RawMessage { handle: handle.to_string(), chat_identifier: chat.to_string(), text: text.to_string(), + voice: None, is_from_me: from_me, is_group: false, is_supported: true, @@ -182,8 +226,10 @@ fn setup_failure_ctx( false, "imessage", )), + voice: None, setup_failure_replies: Arc::new(Mutex::new(Vec::new())), sent_replies: Arc::new(Mutex::new(Vec::new())), + sent_voice_replies: Arc::new(Mutex::new(Vec::new())), send_failures_remaining: Arc::new(Mutex::new(0)), } } @@ -202,6 +248,8 @@ fn setup_failure_job(row_id: i64) -> Job { chat_key: "me@icloud.com".to_string(), }, text: "hello".to_string(), + reply_with_voice: false, + voice_attachment: None, } } @@ -1495,6 +1543,157 @@ async fn enabled_channels_process_concurrently_with_isolated_state_and_origin_re let _ = std::fs::remove_dir_all(assistant_dir); } +#[tokio::test(flavor = "current_thread")] +async fn telegram_voice_is_transcribed_and_gets_text_and_voice_replies() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("voice-sessions"); + let assistant_dir = temp_path("voice-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + gateway.ctx.voice = Some(Voice::with_provider(Arc::new(FakeVoice))); + + run_messages(&mut gateway, vec![telegram_voice_message(1, 7, 7)]).await; + + assert_eq!(calls.lock().unwrap()[0].prompt, "voice request"); + assert_eq!( + gateway.ctx.sent_replies.lock().unwrap().as_slice(), + [("7".to_string(), "fake reply: voice request".to_string())] + ); + assert_eq!( + gateway.ctx.sent_voice_replies.lock().unwrap().as_slice(), + [("7".to_string(), vec![4, 5, 6])] + ); + assert_eq!(gateway.store.lock().unwrap().cursor("telegram"), 1); + let history = gateway + .ctx + .history + .lock() + .unwrap() + .recent_messages_before("telegram", "telegram:dm:7", i64::MAX, 10) + .unwrap(); + assert!(history + .iter() + .any(|message| message.content == "voice request")); + assert!(history + .iter() + .all(|message| message.content != "[Voice message]")); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.db")); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn telegram_voice_without_openai_key_falls_back_without_running_agent() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("voice-missing-key-sessions"); + let assistant_dir = temp_path("voice-missing-key-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7]; + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + gateway.ctx.voice = None; + + run_messages(&mut gateway, vec![telegram_voice_message(1, 7, 7)]).await; + + assert!(calls.lock().unwrap().is_empty()); + assert!(gateway.ctx.sent_replies.lock().unwrap()[0] + .1 + .contains("OPENAI_API_KEY")); + assert_eq!(gateway.store.lock().unwrap().cursor("telegram"), 1); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.db")); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + +#[tokio::test(flavor = "current_thread")] +async fn slow_voice_transcription_does_not_block_another_telegram_thread() { + let state_path = temp_state_path(); + let sessions_dir = temp_path("voice-concurrency-sessions"); + let assistant_dir = temp_path("voice-concurrency-assistant"); + std::fs::create_dir_all(&assistant_dir).unwrap(); + let calls = Arc::new(Mutex::new(Vec::new())); + let mut cfg = test_config( + &state_path, + sessions_dir.to_str().unwrap(), + assistant_dir.to_str().unwrap(), + ); + cfg.channel = "telegram".to_string(); + cfg.telegram_bot_token = Some("secret".to_string()); + cfg.telegram_allow_user_ids = vec![7, 8]; + let mut gateway = Gateway::new(cfg).unwrap(); + gateway.ctx.runners = Arc::new(fake_runners(calls.clone())); + let (release, blocked) = tokio::sync::oneshot::channel(); + gateway.ctx.voice = Some(Voice::with_provider(Arc::new(BlockingVoice { + release: tokio::sync::Mutex::new(Some(blocked)), + }))); + + gateway + .tick_fake(vec![ + telegram_voice_message(1, 7, 7), + telegram_message(2, 8, 8, false, "fast text request"), + ]) + .await; + tokio::time::timeout(Duration::from_secs(1), async { + loop { + if calls + .lock() + .unwrap() + .iter() + .any(|call| call.prompt == "fast text request") + { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("text thread should run while voice transcription is blocked"); + assert!(!calls + .lock() + .unwrap() + .iter() + .any(|call| call.prompt == "slow voice request")); + + release.send(()).unwrap(); + gateway.queues.clear(); + gateway.drain_workers().await; + assert!(calls + .lock() + .unwrap() + .iter() + .any(|call| call.prompt == "slow voice request")); + + let _ = std::fs::remove_file(&state_path); + let _ = std::fs::remove_file(format!("{state_path}.db")); + let _ = std::fs::remove_file(format!("{state_path}.audit.jsonl")); + let _ = std::fs::remove_dir_all(sessions_dir); + let _ = std::fs::remove_dir_all(assistant_dir); +} + #[tokio::test(flavor = "current_thread")] async fn primary_delivery_is_scoped_validated_and_non_fatal_when_missing_or_invalid() { let state_path = temp_state_path(); @@ -2002,6 +2201,8 @@ fn draft_test_job(row_id: i64, target: &str, origin: AnswerOrigin) -> Job { backend: AgentBackend::Codex, approval_origin: origin, text: "draft".to_string(), + reply_with_voice: false, + voice_attachment: None, } } @@ -2126,6 +2327,7 @@ fn message(row_id: i64, chat: &str, handle: &str, is_from_me: bool, text: &str) handle: handle.to_string(), chat_identifier: chat.to_string(), text: text.to_string(), + voice: None, is_from_me, is_group: false, is_supported: true, @@ -2146,9 +2348,22 @@ fn telegram_message( handle: user_id.to_string(), chat_identifier: chat_id.to_string(), text: text.to_string(), + voice: None, is_from_me: false, is_group, is_supported: true, thread_id: None, } } + +fn telegram_voice_message(row_id: i64, user_id: i64, chat_id: i64) -> RawMessage { + let mut message = telegram_message(row_id, user_id, chat_id, false, ""); + message.voice = Some(InboundVoice { + locator: "voice-file".to_string(), + file_size: Some(3), + mime_type: "audio/ogg".to_string(), + filename: "voice.ogg".to_string(), + data: Some(vec![1, 2, 3]), + }); + message +} diff --git a/src/gateway/worker.rs b/src/gateway/worker.rs index 6dc659e..fb388bc 100644 --- a/src/gateway/worker.rs +++ b/src/gateway/worker.rs @@ -15,6 +15,7 @@ use crate::history::{DeliveryStatus, OutboundMessage, OutboundOrigin}; use crate::rehydration::{self, RehydrationPrompt}; use crate::soul; use crate::util::now_ms; +use crate::voice::MAX_AUDIO_BYTES; use super::{audit, complete_row, reply_to, Ctx, Job}; @@ -32,7 +33,7 @@ pub(super) async fn run(ctx: Ctx, mut rx: mpsc::Receiver) { } } -pub(super) async fn handle(ctx: &Ctx, job: Job) { +pub(super) async fn handle(ctx: &Ctx, mut job: Job) { let existing_outbound = match ctx.history.lock().unwrap().outbound_for(job.inbound_id) { Ok(outbound) => outbound, Err(error) => { @@ -82,6 +83,31 @@ pub(super) async fn handle(ctx: &Ctx, job: Job) { return; } + if job.voice_attachment.is_some() { + match prepare_voice(ctx, &job).await { + Ok(transcript) => job.text = transcript, + Err(VoicePreparationError::History(error)) => { + history_error(ctx, &job, "store voice transcript", error); + return; + } + Err(VoicePreparationError::User { + event, + reply, + detail, + }) => { + warn!("[{}] {detail}", job.thread); + audit( + ctx, + ctx.audit + .failed(event, job.row_id, &job.thread, Some(job.backend), detail), + ); + let delivery = record_and_deliver(ctx, &job, OutboundOrigin::Gateway, reply).await; + report_delivery(ctx, &job, delivery, reply, event, "deliver voice fallback"); + return; + } + } + } + if let Some(reply) = command(ctx, &job) { let delivery = record_and_deliver(ctx, &job, OutboundOrigin::Gateway, &reply).await; if delivery.is_ok() { @@ -487,6 +513,62 @@ pub(super) async fn handle(ctx: &Ctx, job: Job) { } } +enum VoicePreparationError { + User { + event: &'static str, + reply: &'static str, + detail: String, + }, + History(anyhow::Error), +} + +async fn prepare_voice(ctx: &Ctx, job: &Job) -> std::result::Result { + let attachment = job + .voice_attachment + .as_ref() + .expect("voice preparation requires an attachment"); + if attachment + .file_size + .is_some_and(|size| size > MAX_AUDIO_BYTES) + { + return Err(VoicePreparationError::User { + event: "voice_too_large", + reply: "That voice message is too large. The limit is 20 MB.", + detail: "voice message exceeds the 20 MB limit".to_string(), + }); + } + let Some(voice) = &ctx.voice else { + return Err(VoicePreparationError::User { + event: "voice_not_configured", + reply: "Voice messages are unavailable. Set OPENAI_API_KEY and restart Push, or send text instead.", + detail: "OPENAI_API_KEY is not configured".to_string(), + }); + }; + let clip = ctx + .channel + .download_voice(attachment) + .await + .map_err(|error| VoicePreparationError::User { + event: "voice_download_failed", + reply: "I could not download that voice message. Please try again or send text.", + detail: format!("voice download failed: {error:#}"), + })?; + let transcript = voice + .transcribe(clip) + .await + .map_err(|error| VoicePreparationError::User { + event: "voice_transcription_failed", + reply: "I could not transcribe that voice message. Please try again or send text.", + detail: format!("voice transcription failed: {error:#}"), + })?; + ctx.history + .lock() + .unwrap() + .replace_inbound_content(job.inbound_id, &transcript) + .map_err(VoicePreparationError::History)?; + Ok(transcript) +} + /// Error and completion labels for one gateway-authored reply flow. struct ReplyLabels { record: &'static str, @@ -778,6 +860,7 @@ async fn deliver_stored( .unwrap() .mark_delivery(outbound.id, status)?; if delivered { + deliver_voice_reply(ctx, job, &outbound.content).await; return Ok(DeliveryOutcome::Delivered); } audit( @@ -807,6 +890,39 @@ async fn deliver_stored( } } +async fn deliver_voice_reply(ctx: &Ctx, job: &Job, text: &str) { + if !job.reply_with_voice { + return; + } + let Some(voice) = &ctx.voice else { + return; + }; + let clip = match voice.synthesize(text).await { + Ok(clip) => clip, + Err(error) => { + warn!( + "[{}] voice reply synthesis failed; text reply was delivered: {error:#}", + job.thread + ); + return; + } + }; + #[cfg(test)] + { + ctx.sent_voice_replies + .lock() + .unwrap() + .push((job.target.clone(), clip.bytes)); + } + #[cfg(not(test))] + if let Err(error) = ctx.channel.send_voice(&job.target, &clip).await { + warn!( + "[{}] voice reply delivery failed; text reply was delivered: {error:#}", + job.thread + ); + } +} + fn history_error(ctx: &Ctx, job: &Job, action: &str, error: anyhow::Error) { error!( "[{}] canonical history {action} failed: {error}; refusing unrecorded delivery", diff --git a/src/history.rs b/src/history.rs index 3241ebe..bd4ef47 100644 --- a/src/history.rs +++ b/src/history.rs @@ -186,6 +186,24 @@ impl History { Ok(message) } + pub fn replace_inbound_content(&mut self, inbound_id: i64, content: &str) -> Result<()> { + let changed = self + .conn + .execute( + "UPDATE messages + SET content = ?2, updated_at = strftime('%Y-%m-%dT%H:%M:%fZ', 'now') + WHERE id = ?1 AND direction = 'inbound'", + params![inbound_id, content], + ) + .with_context(|| { + format!("update inbound message content in {}", self.path.display()) + })?; + if changed != 1 { + bail!("inbound message {inbound_id} does not exist"); + } + Ok(()) + } + pub fn outbound_for(&self, inbound_id: i64) -> Result> { outbound_for_query(&self.conn, inbound_id).with_context(|| { format!( diff --git a/src/main.rs b/src/main.rs index a5b7584..a5fadb8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -24,6 +24,7 @@ mod telegram; #[cfg(test)] mod test_support; mod util; +mod voice; use anyhow::{bail, Context, Result}; diff --git a/src/telegram.rs b/src/telegram.rs index 8e095a9..6da7844 100644 --- a/src/telegram.rs +++ b/src/telegram.rs @@ -6,11 +6,12 @@ use std::pin::Pin; use std::sync::Arc; use std::time::Duration; -use anyhow::{bail, Result}; +use anyhow::{bail, Context, Result}; use serde::Deserialize; use serde_json::{json, Value}; -use crate::channel::RawMessage; +use crate::channel::{InboundVoice, RawMessage}; +use crate::voice::{AudioClip, MAX_AUDIO_BYTES}; pub const TEXT_LIMIT: usize = 4096; pub const RICH_TEXT_LIMIT: usize = 32768; @@ -23,10 +24,25 @@ struct TransportResponse { } type TransportFuture<'a> = Pin> + Send + 'a>>; +type BytesFuture<'a> = Pin>> + Send + 'a>>; trait Transport: Send + Sync { fn post<'a>(&'a self, token: &'a str, method: &'static str, body: Value) -> TransportFuture<'a>; + + fn download<'a>(&'a self, _token: &'a str, _file_path: &'a str) -> BytesFuture<'a> { + Box::pin(async { bail!("Telegram file download is not available") }) + } + + #[cfg_attr(test, allow(dead_code))] + fn post_voice<'a>( + &'a self, + _token: &'a str, + _payload: Value, + _clip: &'a AudioClip, + ) -> TransportFuture<'a> { + Box::pin(async { bail!("Telegram voice upload is not available") }) + } } struct ReqwestTransport { @@ -57,6 +73,85 @@ impl Transport for ReqwestTransport { Ok(TransportResponse { status, body }) }) } + + fn download<'a>(&'a self, token: &'a str, file_path: &'a str) -> BytesFuture<'a> { + Box::pin(async move { + let url = format!("https://api.telegram.org/file/bot{token}/{file_path}"); + let mut response = self + .client + .get(url) + .timeout(Duration::from_secs(HTTP_TIMEOUT_SECONDS)) + .send() + .await + .map_err(|_| anyhow::anyhow!("Telegram voice download failed"))?; + let status = response.status(); + if !status.is_success() { + bail!("Telegram voice download returned HTTP {}", status.as_u16()); + } + if response + .content_length() + .is_some_and(|size| size > MAX_AUDIO_BYTES as u64) + { + bail!("Telegram voice message exceeds the 20 MB limit"); + } + let mut bytes = Vec::with_capacity( + response + .content_length() + .unwrap_or_default() + .min(MAX_AUDIO_BYTES as u64) as usize, + ); + while let Some(chunk) = response + .chunk() + .await + .context("read Telegram voice message")? + { + if bytes.len().saturating_add(chunk.len()) > MAX_AUDIO_BYTES { + bail!("Telegram voice message exceeds the 20 MB limit"); + } + bytes.extend_from_slice(&chunk); + } + Ok(bytes) + }) + } + + fn post_voice<'a>( + &'a self, + token: &'a str, + payload: Value, + clip: &'a AudioClip, + ) -> TransportFuture<'a> { + Box::pin(async move { + let url = format!("https://api.telegram.org/bot{token}/sendVoice"); + let mut form = reqwest::multipart::Form::new(); + let object = payload + .as_object() + .context("Telegram voice payload must be an object")?; + for (key, value) in object { + let value = value + .as_str() + .map(str::to_string) + .unwrap_or_else(|| value.to_string()); + form = form.text(key.clone(), value); + } + let part = reqwest::multipart::Part::bytes(clip.bytes.clone()) + .file_name(clip.filename.clone()) + .mime_str(&clip.mime_type) + .context("prepare Telegram voice reply")?; + let response = self + .client + .post(url) + .timeout(Duration::from_secs(HTTP_TIMEOUT_SECONDS)) + .multipart(form.part("voice", part)) + .send() + .await + .map_err(|_| anyhow::anyhow!("Telegram sendVoice request failed"))?; + let status = response.status().as_u16(); + let body = response.json().await.with_context(|| { + format!("Telegram sendVoice returned HTTP {status} with invalid JSON") + })?; + Ok(TransportResponse { status, body }) + }) + } } #[derive(Clone)] @@ -218,6 +313,75 @@ impl Telegram { Ok(()) } + pub async fn download_voice(&self, voice: &InboundVoice) -> Result { + if voice.file_size.is_some_and(|size| size > MAX_AUDIO_BYTES) { + bail!("Telegram voice message exceeds the 20 MB limit"); + } + let transport_response = self + .transport + .post(&self.token, "getFile", json!({"file_id": voice.locator})) + .await?; + let response: ApiResponse = + serde_json::from_value(transport_response.body) + .map_err(|_| anyhow::anyhow!("Telegram getFile returned an invalid response"))?; + if !response.ok { + bail!( + "Telegram getFile returned HTTP {}", + transport_response.status + ); + } + let file_path = response + .result + .context("Telegram getFile omitted the file path")? + .file_path; + let bytes = self.transport.download(&self.token, &file_path).await?; + Ok(AudioClip { + bytes, + filename: voice.filename.clone(), + mime_type: voice.mime_type.clone(), + }) + } + + #[cfg_attr(test, allow(dead_code))] + pub async fn send_voice(&self, target: &str, clip: &AudioClip) -> Result<()> { + let payload = target_payload(target); + let transport_response = self.post_voice_with_topic_fallback(payload, clip).await?; + let response: ApiResponse = serde_json::from_value(transport_response.body) + .map_err(|_| anyhow::anyhow!("Telegram sendVoice returned an invalid response"))?; + if !response.ok { + bail!( + "Telegram sendVoice returned HTTP {}", + transport_response.status + ); + } + Ok(()) + } + + #[cfg_attr(test, allow(dead_code))] + async fn post_voice_with_topic_fallback( + &self, + mut payload: Value, + clip: &AudioClip, + ) -> Result { + let response = self + .transport + .post_voice(&self.token, payload.clone(), clip) + .await?; + let thread_missing = response.status == 400 + && response + .body + .get("description") + .and_then(Value::as_str) + .is_some_and(|description| description.contains("message thread not found")); + if !thread_missing || payload.get("message_thread_id").is_none() { + return Ok(response); + } + if let Some(object) = payload.as_object_mut() { + object.remove("message_thread_id"); + } + self.transport.post_voice(&self.token, payload, clip).await + } + /// Posts the payload, retrying once without `message_thread_id` when /// Telegram rejects a private-chat topic send with "message thread not /// found", for example when a topic is stale or unavailable. The retry @@ -300,6 +464,7 @@ impl Update { chat_identifier: String::new(), is_group: false, text: String::new(), + voice: None, is_from_me: false, is_supported: false, thread_id: None, @@ -315,6 +480,13 @@ impl Update { chat_identifier: message.chat.id.to_string(), is_group: message.chat.kind != "private", text: message.text.unwrap_or_default(), + voice: message.voice.map(|voice| InboundVoice { + locator: voice.file_id, + file_size: voice.file_size, + mime_type: voice.mime_type.unwrap_or_else(|| "audio/ogg".to_string()), + filename: "voice.ogg".to_string(), + data: None, + }), is_from_me: false, is_supported: true, thread_id: message.message_thread_id, @@ -330,9 +502,25 @@ struct TelegramMessage { #[serde(default)] text: Option, #[serde(default)] + voice: Option, + #[serde(default)] message_thread_id: Option, } +#[derive(Deserialize)] +struct TelegramVoice { + file_id: String, + #[serde(default)] + file_size: Option, + #[serde(default)] + mime_type: Option, +} + +#[derive(Default, Deserialize)] +struct TelegramFile { + file_path: String, +} + #[derive(Deserialize)] struct User { id: i64, @@ -377,6 +565,9 @@ mod tests { struct FakeTransport { calls: Mutex>, responses: Mutex>, + downloads: Mutex>>, + download_paths: Mutex>, + voice_calls: Mutex>, } impl FakeTransport { @@ -389,6 +580,9 @@ mod tests { .map(|body| TransportResponse { status: 200, body }) .collect(), ), + downloads: Mutex::new(VecDeque::new()), + download_paths: Mutex::new(Vec::new()), + voice_calls: Mutex::new(Vec::new()), } } @@ -401,6 +595,9 @@ mod tests { .map(|(status, body)| TransportResponse { status, body }) .collect(), ), + downloads: Mutex::new(VecDeque::new()), + download_paths: Mutex::new(Vec::new()), + voice_calls: Mutex::new(Vec::new()), } } } @@ -421,6 +618,39 @@ mod tests { .ok_or_else(|| anyhow::anyhow!("missing fake response")) }) } + + fn download<'a>(&'a self, _token: &'a str, file_path: &'a str) -> BytesFuture<'a> { + Box::pin(async move { + self.download_paths + .lock() + .unwrap() + .push(file_path.to_string()); + self.downloads + .lock() + .unwrap() + .pop_front() + .ok_or_else(|| anyhow::anyhow!("missing fake download")) + }) + } + + fn post_voice<'a>( + &'a self, + _token: &'a str, + payload: Value, + clip: &'a AudioClip, + ) -> TransportFuture<'a> { + Box::pin(async move { + self.voice_calls + .lock() + .unwrap() + .push((payload, clip.clone())); + self.responses + .lock() + .unwrap() + .pop_front() + .ok_or_else(|| anyhow::anyhow!("missing fake response")) + }) + } } fn updates() -> Value { @@ -520,6 +750,7 @@ mod tests { kind: "private".to_string(), }, text: Some("hi".to_string()), + voice: None, message_thread_id: None, }), } @@ -532,6 +763,84 @@ mod tests { assert!(!telegram.is_allowed(&message)); } + #[test] + fn parses_voice_attachment_without_downloading_it() { + let message = Update { + update_id: 2, + message: Some(TelegramMessage { + from: Some(User { id: 7 }), + chat: Chat { + id: 7, + kind: "private".to_string(), + }, + text: None, + voice: Some(TelegramVoice { + file_id: "file-123".to_string(), + file_size: Some(42), + mime_type: Some("audio/ogg".to_string()), + }), + message_thread_id: None, + }), + } + .into_raw(); + + assert!(message.text.is_empty()); + let voice = message.voice.unwrap(); + assert_eq!(voice.locator, "file-123"); + assert_eq!(voice.file_size, Some(42)); + assert!(voice.data.is_none()); + } + + #[tokio::test] + async fn downloads_voice_by_file_id_and_returns_generic_audio() { + let fake = Arc::new(FakeTransport::with_responses(vec![json!({ + "ok": true, + "result": {"file_path": "voice/file.oga"} + })])); + fake.downloads.lock().unwrap().push_back(vec![1, 2, 3]); + let telegram = + Telegram::with_transport("secret".to_string(), vec![7], vec![], fake.clone()); + let voice = InboundVoice { + locator: "file-123".to_string(), + file_size: Some(3), + mime_type: "audio/ogg".to_string(), + filename: "voice.ogg".to_string(), + data: None, + }; + + let clip = telegram.download_voice(&voice).await.unwrap(); + + assert_eq!(clip.bytes, vec![1, 2, 3]); + assert_eq!(fake.calls.lock().unwrap()[0].0, "getFile"); + assert_eq!(fake.calls.lock().unwrap()[0].1["file_id"], "file-123"); + assert_eq!( + fake.download_paths.lock().unwrap().as_slice(), + ["voice/file.oga"] + ); + } + + #[tokio::test] + async fn uploads_opus_voice_to_the_exact_topic() { + let fake = Arc::new(FakeTransport::with_responses(vec![json!({ + "ok": true, + "result": {} + })])); + let telegram = + Telegram::with_transport("secret".to_string(), vec![7], vec![], fake.clone()); + let clip = AudioClip { + bytes: vec![4, 5, 6], + filename: "reply.opus".to_string(), + mime_type: "audio/ogg".to_string(), + }; + + telegram.send_voice("7:99", &clip).await.unwrap(); + + let calls = fake.voice_calls.lock().unwrap(); + assert_eq!(calls[0].0["chat_id"], "7"); + assert_eq!(calls[0].0["message_thread_id"], 99); + assert_eq!(calls[0].1, clip); + } + #[test] fn splits_exact_over_limit_multi_chunk_and_unicode_text() { assert_eq!(split_text(&"a".repeat(TEXT_LIMIT)).len(), 1); diff --git a/src/voice.rs b/src/voice.rs new file mode 100644 index 0000000..e8608dc --- /dev/null +++ b/src/voice.rs @@ -0,0 +1,391 @@ +//! Provider-neutral voice transcription and speech synthesis. +#![cfg_attr(test, allow(dead_code))] + +use std::future::Future; +use std::pin::Pin; +use std::sync::Arc; +use std::time::Duration; + +use anyhow::{bail, Context, Result}; +use reqwest::multipart::{Form, Part}; +use serde::Deserialize; +use serde_json::json; + +pub const OPENAI_API_KEY_ENV: &str = "OPENAI_API_KEY"; +pub const TRANSCRIPTION_MODEL: &str = "gpt-4o-transcribe"; +pub const SPEECH_MODEL: &str = "gpt-4o-mini-tts"; +pub const SPEECH_VOICE: &str = "marin"; +pub const MAX_AUDIO_BYTES: usize = 20 * 1024 * 1024; +const MAX_SPEECH_CHARS: usize = 4096; +const REQUEST_TIMEOUT: Duration = Duration::from_secs(90); + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct AudioClip { + pub bytes: Vec, + pub filename: String, + pub mime_type: String, +} + +pub(crate) type VoiceFuture<'a, T> = Pin> + Send + 'a>>; + +pub trait VoiceProvider: Send + Sync { + fn transcribe<'a>(&'a self, clip: AudioClip) -> VoiceFuture<'a, String>; + fn synthesize<'a>(&'a self, text: &'a str) -> VoiceFuture<'a, AudioClip>; +} + +#[derive(Clone)] +pub struct Voice { + provider: Arc, +} + +impl Voice { + pub fn from_env() -> Option { + let key = std::env::var(OPENAI_API_KEY_ENV).ok()?; + let key = key.trim(); + if key.is_empty() { + return None; + } + Some(Self { + provider: Arc::new(OpenAiVoice::new(key.to_string())), + }) + } + + #[cfg(test)] + pub fn with_provider(provider: Arc) -> Self { + Self { provider } + } + + pub async fn transcribe(&self, clip: AudioClip) -> Result { + if clip.bytes.is_empty() { + bail!("voice message is empty"); + } + if clip.bytes.len() > MAX_AUDIO_BYTES { + bail!("voice message exceeds the 20 MB limit"); + } + let text = self.provider.transcribe(clip).await?; + let text = text.trim(); + if text.is_empty() { + bail!("voice message did not contain recognizable speech"); + } + Ok(text.to_string()) + } + + pub async fn synthesize(&self, text: &str) -> Result { + if text.trim().is_empty() { + bail!("speech reply is empty"); + } + if text.chars().count() > MAX_SPEECH_CHARS { + bail!("speech reply exceeds the 4096 character limit"); + } + let clip = self.provider.synthesize(text).await?; + if clip.bytes.is_empty() { + bail!("speech reply is empty"); + } + if clip.bytes.len() > MAX_AUDIO_BYTES { + bail!("speech reply exceeds the 20 MB limit"); + } + Ok(clip) + } +} + +struct OpenAiVoice { + api_key: String, + client: reqwest::Client, + base_url: String, +} + +impl OpenAiVoice { + fn new(api_key: String) -> Self { + Self { + api_key, + client: reqwest::Client::new(), + base_url: "https://api.openai.com/v1".to_string(), + } + } + + #[cfg(test)] + fn with_base_url(api_key: String, base_url: String) -> Self { + Self { + api_key, + client: reqwest::Client::new(), + base_url, + } + } +} + +impl VoiceProvider for OpenAiVoice { + fn transcribe<'a>(&'a self, clip: AudioClip) -> VoiceFuture<'a, String> { + Box::pin(async move { + let part = Part::bytes(clip.bytes) + .file_name(clip.filename) + .mime_str(&clip.mime_type) + .context("prepare voice message")?; + let response = self + .client + .post(format!("{}/audio/transcriptions", self.base_url)) + .bearer_auth(&self.api_key) + .timeout(REQUEST_TIMEOUT) + .multipart( + Form::new() + .text("model", TRANSCRIPTION_MODEL) + .part("file", part), + ) + .send() + .await + .context("OpenAI transcription request failed")?; + let status = response.status(); + if !status.is_success() { + bail!("OpenAI transcription returned HTTP {}", status.as_u16()); + } + let result: Transcription = response + .json() + .await + .context("OpenAI transcription returned invalid JSON")?; + Ok(result.text) + }) + } + + fn synthesize<'a>(&'a self, text: &'a str) -> VoiceFuture<'a, AudioClip> { + Box::pin(async move { + let mut response = self + .client + .post(format!("{}/audio/speech", self.base_url)) + .bearer_auth(&self.api_key) + .timeout(REQUEST_TIMEOUT) + .json(&json!({ + "model": SPEECH_MODEL, + "voice": SPEECH_VOICE, + "input": text, + "instructions": "Speak naturally, clearly, and concisely.", + "response_format": "opus" + })) + .send() + .await + .context("OpenAI speech request failed")?; + let status = response.status(); + if !status.is_success() { + bail!("OpenAI speech returned HTTP {}", status.as_u16()); + } + if response + .content_length() + .is_some_and(|size| size > MAX_AUDIO_BYTES as u64) + { + bail!("OpenAI speech response exceeds the 20 MB limit"); + } + let mut bytes = Vec::with_capacity( + response + .content_length() + .unwrap_or_default() + .min(MAX_AUDIO_BYTES as u64) as usize, + ); + while let Some(chunk) = response + .chunk() + .await + .context("read OpenAI speech response")? + { + if bytes.len().saturating_add(chunk.len()) > MAX_AUDIO_BYTES { + bail!("OpenAI speech response exceeds the 20 MB limit"); + } + bytes.extend_from_slice(&chunk); + } + Ok(AudioClip { + bytes, + filename: "reply.opus".to_string(), + mime_type: "audio/ogg".to_string(), + }) + }) + } +} + +#[derive(Deserialize)] +struct Transcription { + text: String, +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::Value; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::thread::JoinHandle; + + struct FakeProvider; + + impl VoiceProvider for FakeProvider { + fn transcribe<'a>(&'a self, _clip: AudioClip) -> VoiceFuture<'a, String> { + Box::pin(async { Ok(" remind me tomorrow ".to_string()) }) + } + + fn synthesize<'a>(&'a self, _text: &'a str) -> VoiceFuture<'a, AudioClip> { + Box::pin(async { + Ok(AudioClip { + bytes: vec![1, 2, 3], + filename: "reply.opus".to_string(), + mime_type: "audio/ogg".to_string(), + }) + }) + } + } + + fn clip(bytes: Vec) -> AudioClip { + AudioClip { + bytes, + filename: "voice.ogg".to_string(), + mime_type: "audio/ogg".to_string(), + } + } + + fn serve_once( + status: &str, + content_type: &str, + body: Vec, + ) -> (String, JoinHandle>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + let status = status.to_string(); + let content_type = content_type.to_string(); + let handle = std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(5))) + .unwrap(); + let mut request = Vec::new(); + let mut buffer = [0_u8; 4096]; + loop { + let count = stream.read(&mut buffer).unwrap(); + if count == 0 { + break; + } + request.extend_from_slice(&buffer[..count]); + let Some(headers_end) = request.windows(4).position(|window| window == b"\r\n\r\n") + else { + continue; + }; + let headers = String::from_utf8_lossy(&request[..headers_end]); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().unwrap()) + }) + .unwrap_or_default(); + if request.len() >= headers_end + 4 + content_length { + break; + } + } + write!( + stream, + "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", + body.len() + ) + .unwrap(); + stream.write_all(&body).unwrap(); + request + }); + (format!("http://{address}"), handle) + } + + #[tokio::test] + async fn provider_boundary_trims_transcripts_and_returns_opus() { + let voice = Voice::with_provider(Arc::new(FakeProvider)); + assert_eq!( + voice.transcribe(clip(vec![1])).await.unwrap(), + "remind me tomorrow" + ); + assert_eq!( + voice.synthesize("done").await.unwrap().mime_type, + "audio/ogg" + ); + } + + #[tokio::test] + async fn provider_boundary_rejects_empty_and_oversized_input() { + let voice = Voice::with_provider(Arc::new(FakeProvider)); + assert!(voice.transcribe(clip(Vec::new())).await.is_err()); + assert!(voice + .transcribe(clip(vec![0; MAX_AUDIO_BYTES + 1])) + .await + .is_err()); + assert!(voice + .synthesize(&"x".repeat(MAX_SPEECH_CHARS + 1)) + .await + .is_err()); + } + + #[test] + fn defaults_use_current_openai_audio_models() { + assert_eq!(TRANSCRIPTION_MODEL, "gpt-4o-transcribe"); + assert_eq!(SPEECH_MODEL, "gpt-4o-mini-tts"); + } + + #[tokio::test] + async fn openai_transcription_http_contract_uses_multipart_and_current_model() { + let (base_url, request) = serve_once( + "200 OK", + "application/json", + br#"{"text":"contract transcript"}"#.to_vec(), + ); + let provider = OpenAiVoice::with_base_url("test-key".to_string(), base_url); + + let transcript = provider + .transcribe(clip(b"audio-bytes".to_vec())) + .await + .unwrap(); + let request = request.join().unwrap(); + let request_text = String::from_utf8_lossy(&request); + + assert_eq!(transcript, "contract transcript"); + assert!(request_text.starts_with("POST /audio/transcriptions HTTP/1.1")); + assert!(request_text + .to_ascii_lowercase() + .contains("authorization: bearer test-key")); + assert!(request_text.contains("name=\"model\"")); + assert!(request_text.contains(TRANSCRIPTION_MODEL)); + assert!(request_text.contains("filename=\"voice.ogg\"")); + assert!(request_text.contains("audio-bytes")); + } + + #[tokio::test] + async fn openai_speech_http_contract_requests_opus_and_returns_audio() { + let (base_url, request) = serve_once("200 OK", "application/octet-stream", vec![9, 8, 7]); + let provider = OpenAiVoice::with_base_url("test-key".to_string(), base_url); + + let output = provider.synthesize("hello").await.unwrap(); + let request = request.join().unwrap(); + let body_start = request + .windows(4) + .position(|window| window == b"\r\n\r\n") + .unwrap() + + 4; + let payload: Value = serde_json::from_slice(&request[body_start..]).unwrap(); + + assert_eq!(output.bytes, vec![9, 8, 7]); + assert_eq!(payload["model"], SPEECH_MODEL); + assert_eq!(payload["voice"], SPEECH_VOICE); + assert_eq!(payload["response_format"], "opus"); + assert_eq!(payload["input"], "hello"); + } + + #[tokio::test] + async fn openai_http_errors_are_actionable_without_returning_body_content() { + let (base_url, request) = serve_once( + "401 Unauthorized", + "application/json", + br#"{"error":"secret upstream detail"}"#.to_vec(), + ); + let provider = OpenAiVoice::with_base_url("bad-key".to_string(), base_url); + + let error = provider + .transcribe(clip(b"audio".to_vec())) + .await + .unwrap_err() + .to_string(); + request.join().unwrap(); + + assert!(error.contains("HTTP 401")); + assert!(!error.contains("secret upstream detail")); + assert!(!error.contains("bad-key")); + } +} diff --git a/tests/manual_job_crash.rs b/tests/manual_job_crash.rs index 074310e..7b9ec0a 100644 --- a/tests/manual_job_crash.rs +++ b/tests/manual_job_crash.rs @@ -2,7 +2,7 @@ use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; -use std::process::{Command, Stdio}; +use std::process::{Child, Command, ExitStatus, Stdio}; use std::time::{Duration, Instant}; use rusqlite::{Connection, OptionalExtension}; @@ -100,9 +100,7 @@ fn concurrent_first_runs_on_a_fresh_database_skip_without_sqlite_errors() { let mut second = spawn_run(binary, &config); wait_for_counts(&database, 1, 1); - let first_status = first.try_wait().unwrap(); - let second_status = second.try_wait().unwrap(); - assert!(first_status.is_none() ^ second_status.is_none()); + let (first_status, second_status) = wait_for_one_exit(&mut first, &mut second); let skipped_status = first_status.or(second_status).unwrap(); assert!(skipped_status.success()); if first_status.is_none() { @@ -196,6 +194,26 @@ fn spawn_run(binary: &str, config: &Path) -> std::process::Child { .unwrap() } +fn wait_for_one_exit( + first: &mut Child, + second: &mut Child, +) -> (Option, Option) { + let deadline = Instant::now() + Duration::from_secs(5); + while Instant::now() < deadline { + let first_status = first.try_wait().unwrap(); + let second_status = second.try_wait().unwrap(); + if first_status.is_some() ^ second_status.is_some() { + return (first_status, second_status); + } + assert!( + first_status.is_none() && second_status.is_none(), + "both concurrent runs exited" + ); + std::thread::sleep(Duration::from_millis(10)); + } + panic!("timed out waiting for the skipped run to exit"); +} + fn run_cli(binary: &str, config: &Path, args: &[&str]) -> std::process::Output { Command::new(binary) .args(args)