1. 1//! Channel voice notes end to end (docs/design/49-live-voice.md): a bridge
2. 2//! only attaches the audio; the gateway transcribes it after allowlist
3. 3//! admission, through the chat's voice route, and the model receives the
4. 4//! words as the user's message. A chat awaiting approval never spends a
5. 5//! provider call.
6. 6
7. 7#![cfg(unix)]
8. 8#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
9. 9
10. 10use std::path::{Path, PathBuf};
11. 11use std::sync::{Arc, Mutex};
12. 12
13. 13use tokio_util::sync::CancellationToken;
14. 14use vak_core::Core;
15. 15use vak_llm::stream;
16. 16use vak_llm::types::{AssistantMessage, ChatRequest, ContentBlock, StopReason, Usage};
17. 17use vak_llm::{EventStream, LlmError, Provider};
18. 18
19. 19/// Answers every turn and keeps the last request it was sent.
20. 20#[derive(Default)]
21. 21struct Recording {
22. 22 last_request: Mutex<String>,
23. 23}
24. 24
25. 25#[async_trait::async_trait]
26. 26impl Provider for Recording {
27. 27 fn name(&self) -> &str {
28. 28 "recording"
29. 29 }
30. 30
31. 31 async fn stream(
32. 32 &self,
33. 33 request: ChatRequest,
34. 34 _cancel: CancellationToken,
35. 35 ) -> Result<EventStream, LlmError> {
36. 36 *self.last_request.lock().unwrap() = format!("{:?}", request.messages.last());
37. 37 let message = AssistantMessage {
38. 38 content: vec![ContentBlock::text("Four.")],
39. 39 stop_reason: StopReason::EndTurn,
40. 40 usage: Usage::default(),
41. 41 model: "test-model".into(),
42. 42 response_id: None,
43. 43 };
44. 44 let (mut sink, rx) = stream::channel(64);
45. 45 sink.push(stream::StreamEvent::Start {
46. 46 partial: message.clone(),
47. 47 });
48. 48 sink.close_message(message).await;
49. 49 Ok(rx)
50. 50 }
51. 51}
52. 52
53. 53fn transcriber(dir: &Path) -> (PathBuf, PathBuf) {
54. 54 use std::os::unix::fs::PermissionsExt;
55. 55 let calls = dir.join("transcriber-calls");
56. 56 let path = dir.join("transcriber.sh");
57. 57 std::fs::write(
58. 58 &path,
59. 59 format!(
60. 60 "#!/bin/sh\n/bin/cat >/dev/null\necho call >> '{}'\nprintf 'what is two plus two'\n",
61. 61 calls.display()
62. 62 ),
63. 63 )
64. 64 .unwrap();
65. 65 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o700)).unwrap();
66. 66 (path, calls)
67. 67}
68. 68
69. 69fn calls(path: &Path) -> usize {
70. 70 std::fs::read_to_string(path)
71. 71 .map(|text| text.lines().count())
72. 72 .unwrap_or(0)
73. 73}
74. 74
75. 75async fn spawn_gateway(allowlist_open: bool, provider: Arc<Recording>) -> String {
76. 76 vak_config::paths::isolate_home_for_tests();
77. 77 let dir = tempfile::tempdir().unwrap();
78. 78 let cwd = dir.path().to_path_buf();
79. 79 std::fs::create_dir_all(cwd.join(".vak")).unwrap();
80. 80 std::fs::write(
81. 81 cwd.join(".vak/config.toml"),
82. 82 format!(
83. 83 "[memory]\nreflection = false\n[gateway]\nchat_allowlist_open = {allowlist_open}\n"
84. 84 ),
85. 85 )
86. 86 .unwrap();
87. 87 let core = Core::new_with_trust(cwd, true).unwrap();
88. 88 core.set_sessions_home(dir.path().join("home"));
89. 89 core.set_provider_instance(provider);
90. 90 core.apply_persisted_voice(vak_config::VoiceSettings {
91. 91 enabled: true,
92. 92 provider: Some("local".into()),
93. 93 ..Default::default()
94. 94 });
95. 95 std::mem::forget(dir);
96. 96 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
97. 97 let addr = listener.local_addr().unwrap();
98. 98 let app = vak_server::gateway_router(core);
99. 99 tokio::spawn(async move {
100. 100 axum::serve(listener, app).await.unwrap();
101. 101 });
102. 102 format!("http://{addr}")
103. 103}
104. 104
105. 105fn voice_note(chat: &str) -> serde_json::Value {
106. 106 use base64::Engine as _;
107. 107 serde_json::json!({
108. 108 "surface": "telegram",
109. 109 "chat": chat,
110. 110 "sender": "alice",
111. 111 "text": "",
112. 112 "wait": true,
113. 113 "attachments": [{
114. 114 "kind": "audio",
115. 115 "mime": "audio/ogg",
116. 116 "filename": "voice.ogg",
117. 117 "data": base64::engine::general_purpose::STANDARD.encode(b"OggS fake opus"),
118. 118 }],
119. 119 })
120. 120}
121. 121
122. 122/// One test, because the transcriber override is process-global.
123. 123#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
124. 124async fn channel_voice_notes_are_transcribed_only_after_admission() {
125. 125 let scratch = tempfile::tempdir().unwrap();
126. 126 let (script, calls_path) = transcriber(scratch.path());
127. 127 vak_config::set_override(vak_voice::TRANSCRIBER_VAR, script.display().to_string());
128. 128 let client = reqwest::Client::new();
129. 129
130. 130 // An unknown chat lands as a pending review; its voice note is never
131. 131 // transcribed, so a stranger cannot spend the operator's provider.
132. 132 let closed = spawn_gateway(false, Arc::new(Recording::default())).await;
133. 133 let pending = client
134. 134 .post(format!("{closed}/gateway/inbound"))
135. 135 .json(&voice_note("1001"))
136. 136 .send()
137. 137 .await
138. 138 .unwrap();
139. 139 assert_eq!(pending.status(), reqwest::StatusCode::FORBIDDEN);
140. 140 assert_eq!(
141. 141 calls(&calls_path),
142. 142 0,
143. 143 "a pending chat must not be transcribed"
144. 144 );
145. 145
146. 146 // An admitted chat's note becomes the user's words.
147. 147 let provider = Arc::new(Recording::default());
148. 148 let open = spawn_gateway(true, provider.clone()).await;
149. 149 let admitted = client
150. 150 .post(format!("{open}/gateway/inbound"))
151. 151 .json(&voice_note("2002"))
152. 152 .send()
153. 153 .await
154. 154 .unwrap();
155. 155 let status = admitted.status();
156. 156 let body: serde_json::Value = admitted.json().await.unwrap();
157. 157 assert!(status.is_success(), "{status}: {body}");
158. 158 assert_eq!(calls(&calls_path), 1);
159. 159 let request = provider.last_request.lock().unwrap().clone();
160. 160 assert!(
161. 161 request.contains("what is two plus two"),
162. 162 "the model must receive the transcript: {request}"
163. 163 );
164. 164 assert!(
165. 165 !request.contains("not configured"),
166. 166 "no false statement about the transcription provider: {request}"
167. 167 );
168. 168}
169. 169
170. 170/// A bot's voice tier is checked where it is written: a provider that does
171. 171/// not exist is refused, not stored to fail later on a voice note.
172. 172#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
173. 173async fn a_bot_voice_tier_with_an_unknown_provider_is_refused() {
174. 174 let base = spawn_gateway(true, Arc::new(Recording::default())).await;
175. 175 let client = reqwest::Client::new();
176. 176 let created = client
177. 177 .post(format!("{base}/gateway/bots"))
178. 178 .json(&serde_json::json!({"id": "support", "surface": "telegram", "label": "Support"}))
179. 179 .send()
180. 180 .await
181. 181 .unwrap();
182. 182 assert!(created.status().is_success(), "{}", created.status());
183. 183 let patch = |provider: &str| {
184. 184 client
185. 185 .patch(format!("{base}/gateway/bots/support"))
186. 186 .json(&serde_json::json!({"voice": {"provider": provider}}))
187. 187 .send()
188. 188 };
189. 189 assert_eq!(
190. 190 patch("google").await.unwrap().status(),
191. 191 reqwest::StatusCode::BAD_REQUEST
192. 192 );
193. 193 assert!(patch("openai").await.unwrap().status().is_success());
194. 194}
195. 195