Текстовый файл из учебного комплекта, не свидетельство установленного runtime. Вернуться к учебнику · Исходный файл
1 //! Synchronous UI handles for one helper-owning actor. Closing any handle or dropping the last 2 //! owner cancels pending I/O; bounded, ordered commands preserve every mute transition. 3 //! A shared process-lifetime runtime keeps child reapers alive after an actor stops. 4 5 use std::sync::Arc; 6 use std::sync::Mutex; 7 use std::sync::atomic::AtomicU16; 8 use std::sync::atomic::Ordering; 9 use std::sync::mpsc as blocking; 10 use std::time::Duration; 11 12 use anyhow::Result; 13 use futures::future::AbortHandle; 14 use futures::future::AbortRegistration; 15 use futures::future::Abortable; 16 use tokio::sync::mpsc; 17 18 use crate::AudioControls; 19 use crate::ConnectionError; 20 use crate::SessionDescription; 21 use crate::VoiceHost; 22 23 // Handshake (5s), runtime initialization (30s), offer gathering (20s), plus overhead. 24 const STARTUP_WAIT: Duration = Duration::from_secs(/*secs*/ 60); 25 // An in-flight request (5s), answer (20s), devices (5s), controls (5s), plus overhead. 26 const ANSWER_WAIT: Duration = Duration::from_secs(/*secs*/ 40); 27 static RUNTIME: Mutex<Option<Arc<tokio::runtime::Runtime>>> = Mutex::new(None); 28 29 enum Command { 30 Answer( 31 SessionDescription, 32 blocking::SyncSender<std::result::Result<(), crate::ConnectionError>>, 33 ), 34 Controls(AudioControls), 35 } 36 37 #[derive(Default)] 38 struct State { 39 microphone: AtomicU16, 40 speaker: AtomicU16, 41 error: Mutex<Option<String>>, 42 } 43 44 struct Owner { 45 sender: mpsc::Sender<Command>, 46 stop: AbortHandle, 47 state: Arc<State>, 48 controls: Arc<Mutex<AudioControls>>, 49 } 50 51 impl Drop for Owner { 52 fn drop(&mut self) { 53 self.stop.abort(); 54 } 55 } 56 57 /// A cloned handle shares ownership; native media remains entirely inside the child process. 58 #[derive(Clone)] 59 pub struct RealtimeWebrtcSessionHandle(Arc<Owner>); 60 61 impl std::fmt::Debug for RealtimeWebrtcSessionHandle { 62 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 63 f.write_str("RealtimeWebrtcSessionHandle") 64 } 65 } 66 67 /// An offer is not proof of connectivity or working devices. 68 pub struct StartedRealtimeWebrtcSession { 69 pub offer_sdp: String, 70 pub handle: RealtimeWebrtcSessionHandle, 71 } 72 73 impl std::fmt::Debug for StartedRealtimeWebrtcSession { 74 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 75 f.write_str("StartedRealtimeWebrtcSession { offer_sdp: [REDACTED] }") 76 } 77 } 78 79 pub struct RealtimeWebrtcSession; 80 81 impl RealtimeWebrtcSession { 82 /// Check package availability without loading native code or touching any audio device. 83 /// Linux packages may pair a musl application with a GNU helper; startup checks host loading. 84 pub fn is_supported() -> bool { 85 cfg!(any( 86 target_os = "macos", 87 target_os = "linux", 88 all(windows, target_env = "msvc") 89 )) && codex_install_context::InstallContext::current() 90 .package_layout 91 .as_ref() 92 .is_some_and(|package| package_has_runtime(package.package_dir.as_path())) 93 } 94 95 /// Called off the UI thread. Cancellation also owns startup before a handle is returned. 96 pub fn start(abort: AbortRegistration) -> Result<StartedRealtimeWebrtcSession> { 97 let package = codex_install_context::InstallContext::current() 98 .package_layout 99 .clone() 100 .ok_or_else(|| anyhow::anyhow!("voice package unavailable"))?; 101 let build_commit = codex_build_info::BuildInfo::get().build_commit().to_owned(); 102 let (sender, receiver) = mpsc::channel(/*buffer*/ 8); 103 let (offer, result) = blocking::sync_channel(/*bound*/ 1); 104 let (stop, stopped) = AbortHandle::new_pair(); 105 let state = Arc::new(State::default()); 106 let controls = Arc::new(Mutex::new(AudioControls { 107 microphone_muted: false, 108 speaker_suppressed: false, 109 })); 110 let handle = RealtimeWebrtcSessionHandle(Arc::new(Owner { 111 sender, 112 stop, 113 state: state.clone(), 114 controls: controls.clone(), 115 })); 116 let runtime = { 117 let mut runtime = RUNTIME 118 .lock() 119 .unwrap_or_else(std::sync::PoisonError::into_inner); 120 if let Some(runtime) = runtime.as_ref() { 121 runtime.clone() 122 } else { 123 let shared = Arc::new( 124 tokio::runtime::Builder::new_multi_thread() 125 .worker_threads(/*val*/ 1) 126 .enable_all() 127 .build()?, 128 ); 129 *runtime = Some(shared.clone()); 130 shared 131 } 132 }; 133 std::thread::Builder::new() 134 .name("voice-session".into()) 135 .spawn(move || { 136 let task = async { 137 let host = report_failure( 138 ConnectionError::HelperStartup, 139 VoiceHost::connect(&package, &build_commit).await, 140 )?; 141 let host = report_failure( 142 ConnectionError::RuntimeInitialization, 143 host.initialize_runtime().await, 144 )?; 145 let (host, sdp) = 146 report_failure(ConnectionError::Transport, host.start_transport().await)?; 147 offer 148 .send(Ok(sdp.into_sdp())) 149 .map_err(|_| anyhow::anyhow!("voice startup cancelled"))?; 150 run(host, receiver, &state, &controls).await 151 }; 152 let result = runtime.block_on(Abortable::new(Abortable::new(task, abort), stopped)); 153 if let Ok(Ok(Err(error))) = result { 154 let failure = error 155 .downcast_ref::<ConnectionError>() 156 .copied() 157 .unwrap_or(ConnectionError::Failed); 158 let _ = offer.try_send(Err(failure)); 159 *state 160 .error 161 .lock() 162 .unwrap_or_else(std::sync::PoisonError::into_inner) = 163 Some(failure.to_string()); 164 } 165 })?; 166 let offer_sdp = result 167 .recv_timeout(STARTUP_WAIT) 168 .map_err(|_| anyhow::anyhow!("voice startup failed"))??; 169 Ok(StartedRealtimeWebrtcSession { offer_sdp, handle }) 170 } 171 } 172 173 // Availability is not proof of runtime integrity, device access, or session connectivity. 174 fn package_has_runtime(package: &std::path::Path) -> bool { 175 let voice = package.join("codex-resources/voice"); 176 let (helper, runtime) = if cfg!(target_os = "macos") { 177 ("bin/codex-voice-host", "lib/libgstreamer-1.0.0.dylib") 178 } else if cfg!(windows) { 179 ("bin/codex-voice-host.exe", "bin/gstreamer-1.0-0.dll") 180 } else { 181 ("bin/codex-voice-host", "lib/libgstreamer-1.0.so.0") 182 }; 183 voice.join(helper).is_file() && voice.join(runtime).is_file() 184 } 185 186 impl RealtimeWebrtcSessionHandle { 187 pub fn close(&self) { 188 self.0.stop.abort(); 189 } 190 191 /// Queue an ordered privacy transition; saturation terminates the session instead of dropping it. 192 pub fn set_microphone_muted(&self, muted: bool) -> Result<()> { 193 let mut controls = self 194 .0 195 .controls 196 .lock() 197 .unwrap_or_else(std::sync::PoisonError::into_inner); 198 controls.microphone_muted = muted; 199 self.send(Command::Controls(*controls)) 200 } 201 202 pub fn set_speaker_suppressed(&self, suppressed: bool) { 203 let mut controls = self 204 .0 205 .controls 206 .lock() 207 .unwrap_or_else(std::sync::PoisonError::into_inner); 208 controls.speaker_suppressed = suppressed; 209 let _ = self.send(Command::Controls(*controls)); 210 } 211 212 /// Called off the UI thread; return only after negotiation, device startup and restored controls. 213 pub fn apply_answer_sdp( 214 &self, 215 answer: String, 216 ) -> std::result::Result<(), crate::ConnectionError> { 217 let sdp = 218 SessionDescription::try_from(answer).map_err(|_| crate::ConnectionError::Failed)?; 219 let (complete, result) = blocking::sync_channel(/*bound*/ 1); 220 self.send(Command::Answer(sdp, complete)) 221 .map_err(|_| crate::ConnectionError::Failed)?; 222 match result.recv_timeout(ANSWER_WAIT) { 223 Ok(result) => result, 224 Err(_) => { 225 self.close(); 226 Err(crate::ConnectionError::Failed) 227 } 228 } 229 } 230 231 pub fn take_error(&self) -> Option<String> { 232 self.0 233 .state 234 .error 235 .lock() 236 .unwrap_or_else(std::sync::PoisonError::into_inner) 237 .take() 238 } 239 240 pub fn take_microphone_peak(&self) -> u16 { 241 self.0.state.microphone.swap(/*val*/ 0, Ordering::AcqRel) 242 } 243 244 pub fn take_speaker_peak(&self) -> u16 { 245 self.0.state.speaker.swap(/*val*/ 0, Ordering::AcqRel) 246 } 247 248 fn send(&self, command: Command) -> Result<()> { 249 if self.0.sender.try_send(command).is_err() { 250 *self 251 .0 252 .state 253 .error 254 .lock() 255 .unwrap_or_else(std::sync::PoisonError::into_inner) = 256 Some("Voice control channel unavailable.".into()); 257 self.close(); 258 anyhow::bail!("voice control channel unavailable"); 259 } 260 Ok(()) 261 } 262 } 263 264 async fn run( 265 mut host: VoiceHost, 266 mut commands: mpsc::Receiver<Command>, 267 state: &State, 268 controls: &Mutex<AudioControls>, 269 ) -> Result<()> { 270 let mut connected = false; 271 let mut poll = tokio::time::interval(Duration::from_millis(/*millis*/ 50)); 272 poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); 273 loop { 274 tokio::select! { 275 biased; 276 command = commands.recv() => match command { 277 Some(Command::Answer(sdp, complete)) if !connected => { 278 let startup = async { 279 let mut host = report_failure(ConnectionError::Transport, host.apply_answer(sdp).await)?; 280 host = report_failure(ConnectionError::AudioDevices, host.open_devices().await)?; 281 let applied = startup_controls(&mut commands, controls, |initial| { 282 host.begin_audio_controls(initial) 283 }); 284 let applied = report_failure(ConnectionError::AudioControls, applied)?; 285 report_failure(ConnectionError::AudioControls, applied.await)?; 286 Ok::<_, anyhow::Error>(host) 287 }.await; 288 host = match startup { 289 Ok(host) => host, 290 Err(error) => { 291 let failure = error.downcast_ref::<ConnectionError>() 292 .copied().unwrap_or(ConnectionError::Failed); 293 let _ = complete.send(Err(failure)); 294 // This failure is delivered by the startup completion only. 295 return Ok(()); 296 } 297 }; 298 connected = true; 299 let _ = complete.send(Ok(())); 300 } 301 Some(Command::Controls(next)) => { 302 if connected { 303 report_failure(ConnectionError::AudioControls, host.set_audio_controls(next).await)?; 304 } 305 } 306 Some(Command::Answer(..)) => anyhow::bail!("voice answer already applied"), 307 None => return report_failure(ConnectionError::Shutdown, host.close().await), 308 }, 309 _ = poll.tick() => { 310 let audio = report_failure(ConnectionError::AudioSession, host.inspect_audio().await)?; 311 state.microphone.fetch_max(audio.microphone_peak, Ordering::Release); 312 state.speaker.fetch_max(audio.speaker_peak, Ordering::Release); 313 } 314 } 315 } 316 } 317 318 // Keep diagnostics bounded and independent of untyped native, SDP, or device error text. 319 fn report_failure<T>(stage: ConnectionError, result: Result<T>) -> Result<T> { 320 result.map_err(|error| { 321 let kind = if error.is::<tokio::time::error::Elapsed>() { 322 "timeout" 323 } else if let Some(error) = error.downcast_ref::<std::io::Error>() { 324 match error.kind() { 325 std::io::ErrorKind::UnexpectedEof 326 | std::io::ErrorKind::BrokenPipe 327 | std::io::ErrorKind::ConnectionReset => "closed", 328 std::io::ErrorKind::InvalidData => "protocol", 329 _ => "io", 330 } 331 } else { 332 "other" 333 }; 334 tracing::warn!(?stage, kind, "voice session operation failed"); 335 let failure = error 336 .downcast_ref::<ConnectionError>() 337 .copied() 338 .unwrap_or(stage); 339 // Replace the original error instead of retaining a potentially sensitive source chain. 340 anyhow::Error::new(failure) 341 }) 342 } 343 344 // Devices are still disabled. The snapshot and request enqueue share the setters' lock; 345 // later setters remain ordered after this request without waiting for its acknowledgement. 346 fn startup_controls<T>( 347 commands: &mut mpsc::Receiver<Command>, 348 controls: &Mutex<AudioControls>, 349 dispatch: impl FnOnce(AudioControls) -> Result<T>, 350 ) -> Result<T> { 351 let controls = controls 352 .lock() 353 .unwrap_or_else(std::sync::PoisonError::into_inner); 354 for _ in 0..commands.max_capacity() { 355 match commands.try_recv() { 356 Ok(Command::Controls(_)) => {} 357 Err(mpsc::error::TryRecvError::Empty) => return dispatch(*controls), 358 Ok(Command::Answer(..)) | Err(mpsc::error::TryRecvError::Disconnected) => { 359 anyhow::bail!("voice startup control sequence invalid"); 360 } 361 } 362 } 363 anyhow::ensure!(commands.is_empty(), "voice startup controls overloaded"); 364 dispatch(*controls) 365 } 366 367 #[cfg(test)] 368 #[path = "session_tests.rs"] 369 mod tests;