feat(pi-natives): added LiveWebRtcPeer and deviceCheckGenerateToken bindings
- Replaced puppeteer-based WebRTC with native LiveWebRtcPeer for cross-platform live audio delivery. - Added cross-platform microphone capture via miniaudio and Opus codec integration for live encoding/decoding. - Added Apple DeviceCheck attestation token generation via raw Objective-C FFI for macOS. - Updated live session model to "gpt-live-1-codex" and default voice to "sol" across protocol and controller. - Added LiveWebRtcPeer and deviceCheckGenerateToken to the public native bindings API.
This commit is contained in:
can1357
@@ -15,9 +15,11 @@ workspace = true
|
||||
|
||||
[dependencies]
|
||||
anyhow.workspace = true
|
||||
audiopus_sys.workspace = true
|
||||
arboard.workspace = true
|
||||
ast-grep-core.workspace = true
|
||||
base64.workspace = true
|
||||
bytes.workspace = true
|
||||
clap.workspace = true
|
||||
globset.workspace = true
|
||||
fontdue.workspace = true
|
||||
@@ -30,8 +32,10 @@ icy_sixel.workspace = true
|
||||
ignore.workspace = true
|
||||
image = { workspace = true, features = ["bmp"] }
|
||||
inferno.workspace = true
|
||||
miniaudio.workspace = true
|
||||
napi.workspace = true
|
||||
napi-derive.workspace = true
|
||||
opus.workspace = true
|
||||
parking_lot.workspace = true
|
||||
phf.workspace = true
|
||||
flume.workspace = true
|
||||
@@ -54,6 +58,7 @@ tokio-util.workspace = true
|
||||
toml.workspace = true
|
||||
unicode-segmentation.workspace = true
|
||||
unicode-width.workspace = true
|
||||
webrtc.workspace = true
|
||||
xxhash-rust.workspace = true
|
||||
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
|
||||
@@ -0,0 +1,427 @@
|
||||
//! Cross-platform microphone capture and streaming speaker playback.
|
||||
//!
|
||||
//! miniaudio owns platform device discovery, format conversion, channel mixing,
|
||||
//! and resampling. The N-API classes expose one stable mono `f32` contract to
|
||||
//! TypeScript while the internal playback stream is shared with native WebRTC.
|
||||
|
||||
use std::sync::{
|
||||
Arc,
|
||||
atomic::{AtomicBool, AtomicU32, Ordering},
|
||||
};
|
||||
|
||||
use flume::TryRecvError;
|
||||
use miniaudio::{Device, DeviceConfig, DeviceType, Format, PerformanceProfile};
|
||||
use napi::{
|
||||
bindgen_prelude::{Float32Array, Result},
|
||||
threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode, UnknownReturnValue},
|
||||
};
|
||||
use napi_derive::napi;
|
||||
use parking_lot::Mutex;
|
||||
use tokio::sync::Notify;
|
||||
|
||||
const AUDIO_CHANNELS: u32 = 1;
|
||||
const AUDIO_PERIOD_MS: u32 = 20;
|
||||
const PLAYBACK_DRAIN_CALLBACKS: usize = 2;
|
||||
|
||||
type CaptureCallback = ThreadsafeFunction<Float32Array, UnknownReturnValue>;
|
||||
type NativeResult<T> = std::result::Result<T, String>;
|
||||
|
||||
struct PlaybackState {
|
||||
gain_bits: AtomicU32,
|
||||
drained: AtomicBool,
|
||||
stopped: AtomicBool,
|
||||
notify: Notify,
|
||||
}
|
||||
|
||||
impl PlaybackState {
|
||||
fn new() -> Self {
|
||||
Self {
|
||||
gain_bits: AtomicU32::new(1.0f32.to_bits()),
|
||||
drained: AtomicBool::new(false),
|
||||
stopped: AtomicBool::new(false),
|
||||
notify: Notify::new(),
|
||||
}
|
||||
}
|
||||
|
||||
fn gain(&self) -> f32 {
|
||||
f32::from_bits(self.gain_bits.load(Ordering::Acquire))
|
||||
}
|
||||
|
||||
fn set_gain(&self, gain: f32) {
|
||||
self.gain_bits.store(gain.to_bits(), Ordering::Release);
|
||||
}
|
||||
|
||||
fn mark_drained(&self) {
|
||||
if !self.drained.swap(true, Ordering::AcqRel) {
|
||||
self.notify.notify_waiters();
|
||||
}
|
||||
}
|
||||
|
||||
fn mark_stopped(&self) {
|
||||
self.stopped.store(true, Ordering::Release);
|
||||
self.notify.notify_waiters();
|
||||
}
|
||||
|
||||
async fn wait_for_drain(&self) {
|
||||
loop {
|
||||
let notified = self.notify.notified();
|
||||
if self.drained.load(Ordering::Acquire) || self.stopped.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
notified.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Producer endpoint for one native playback device.
|
||||
#[derive(Clone)]
|
||||
pub(crate) struct PlaybackWriter {
|
||||
tx: flume::Sender<Vec<f32>>,
|
||||
state: Arc<PlaybackState>,
|
||||
}
|
||||
|
||||
impl PlaybackWriter {
|
||||
/// Queue mono floating-point samples without blocking the caller.
|
||||
pub(crate) fn write(&self, samples: &[f32]) -> NativeResult<()> {
|
||||
if samples.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
if self.state.stopped.load(Ordering::Acquire) || self.state.drained.load(Ordering::Acquire) {
|
||||
return Err("Native audio playback is closed".to_owned());
|
||||
}
|
||||
self.tx
|
||||
.send(samples.to_vec())
|
||||
.map_err(|_| "Native audio playback is closed".to_owned())
|
||||
}
|
||||
}
|
||||
|
||||
/// Running mono playback stream shared by N-API playback and native WebRTC.
|
||||
pub(crate) struct PlaybackStream {
|
||||
device: Option<Device>,
|
||||
writer: Option<PlaybackWriter>,
|
||||
state: Arc<PlaybackState>,
|
||||
}
|
||||
|
||||
impl PlaybackStream {
|
||||
/// Open and start the default speaker at the requested logical sample rate.
|
||||
pub(crate) fn start(sample_rate: u32) -> NativeResult<Self> {
|
||||
validate_sample_rate(sample_rate)?;
|
||||
let state = Arc::new(PlaybackState::new());
|
||||
let (tx, rx) = flume::unbounded::<Vec<f32>>();
|
||||
let mut config = audio_config(DeviceType::Playback, sample_rate);
|
||||
config.playback_mut().set_format(Format::F32);
|
||||
config.playback_mut().set_channels(AUDIO_CHANNELS);
|
||||
let mut device = Device::new(None, &config)
|
||||
.map_err(|error| format!("Failed to open the default speaker: {error}"))?;
|
||||
|
||||
let callback_state = Arc::clone(&state);
|
||||
let mut current = Vec::new();
|
||||
let mut cursor = 0;
|
||||
let mut empty_callbacks = 0;
|
||||
device.set_data_callback(move |_device, output, _input| {
|
||||
fill_playback(
|
||||
&rx,
|
||||
&mut current,
|
||||
&mut cursor,
|
||||
output.as_samples_mut::<f32>(),
|
||||
&callback_state,
|
||||
&mut empty_callbacks,
|
||||
);
|
||||
});
|
||||
let stop_state = Arc::clone(&state);
|
||||
device.set_stop_callback(move |_device| stop_state.mark_stopped());
|
||||
device
|
||||
.start()
|
||||
.map_err(|error| format!("Failed to start speaker playback: {error}"))?;
|
||||
|
||||
Ok(Self {
|
||||
device: Some(device),
|
||||
writer: Some(PlaybackWriter { tx, state: Arc::clone(&state) }),
|
||||
state,
|
||||
})
|
||||
}
|
||||
|
||||
/// Clone the producer endpoint used by the remote-audio decoder.
|
||||
pub(crate) fn writer(&self) -> NativeResult<PlaybackWriter> {
|
||||
self.writer
|
||||
.clone()
|
||||
.ok_or_else(|| "Native audio playback is closed".to_owned())
|
||||
}
|
||||
|
||||
fn state(&self) -> Arc<PlaybackState> {
|
||||
Arc::clone(&self.state)
|
||||
}
|
||||
|
||||
fn finish_input(&mut self) {
|
||||
self.writer.take();
|
||||
}
|
||||
|
||||
fn set_gain(&self, gain: f32) -> NativeResult<()> {
|
||||
if !gain.is_finite() {
|
||||
return Err("Audio playback gain must be finite".to_owned());
|
||||
}
|
||||
self.state.set_gain(gain.max(0.0));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Stop playback immediately and release the default speaker.
|
||||
pub(crate) fn stop(&mut self) -> NativeResult<()> {
|
||||
self.writer.take();
|
||||
self.state.mark_stopped();
|
||||
let Some(device) = self.device.take() else {
|
||||
return Ok(());
|
||||
};
|
||||
device
|
||||
.stop()
|
||||
.map_err(|error| format!("Failed to stop speaker playback: {error}"))
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for PlaybackStream {
|
||||
fn drop(&mut self) {
|
||||
let _ = self.stop();
|
||||
}
|
||||
}
|
||||
|
||||
fn audio_config(device_type: DeviceType, sample_rate: u32) -> DeviceConfig {
|
||||
let mut config = DeviceConfig::new(device_type);
|
||||
config.set_sample_rate(sample_rate);
|
||||
config.set_period_size_in_milliseconds(AUDIO_PERIOD_MS);
|
||||
config.set_performance_profile(PerformanceProfile::LowLatency);
|
||||
config
|
||||
}
|
||||
|
||||
fn validate_sample_rate(sample_rate: u32) -> NativeResult<()> {
|
||||
if sample_rate == 0 {
|
||||
return Err("Audio sample rate must be greater than zero".to_owned());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn fill_playback(
|
||||
rx: &flume::Receiver<Vec<f32>>,
|
||||
current: &mut Vec<f32>,
|
||||
cursor: &mut usize,
|
||||
output: &mut [f32],
|
||||
state: &PlaybackState,
|
||||
empty_callbacks: &mut usize,
|
||||
) {
|
||||
output.fill(0.0);
|
||||
if state.stopped.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
|
||||
let gain = state.gain();
|
||||
let mut output_offset = 0;
|
||||
while output_offset < output.len() {
|
||||
if *cursor == current.len() {
|
||||
match rx.try_recv() {
|
||||
Ok(next) => {
|
||||
*current = next;
|
||||
*cursor = 0;
|
||||
*empty_callbacks = 0;
|
||||
},
|
||||
Err(TryRecvError::Empty) => {
|
||||
*empty_callbacks = 0;
|
||||
break;
|
||||
},
|
||||
Err(TryRecvError::Disconnected) => {
|
||||
*empty_callbacks += 1;
|
||||
if *empty_callbacks >= PLAYBACK_DRAIN_CALLBACKS {
|
||||
state.mark_drained();
|
||||
}
|
||||
break;
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
let count = (current.len() - *cursor).min(output.len() - output_offset);
|
||||
let source = ¤t[*cursor..*cursor + count];
|
||||
let destination = &mut output[output_offset..output_offset + count];
|
||||
if gain == 1.0 {
|
||||
destination.copy_from_slice(source);
|
||||
} else {
|
||||
for (destination, source) in destination.iter_mut().zip(source) {
|
||||
*destination = *source * gain;
|
||||
}
|
||||
}
|
||||
*cursor += count;
|
||||
output_offset += count;
|
||||
}
|
||||
}
|
||||
|
||||
/// Default-microphone capture converted to mono `f32` at the requested sample rate.
|
||||
#[napi]
|
||||
pub struct AudioCapture {
|
||||
device: Mutex<Option<Device>>,
|
||||
}
|
||||
|
||||
#[napi]
|
||||
impl AudioCapture {
|
||||
/// Open the default microphone and deliver low-latency mono PCM chunks.
|
||||
#[napi(constructor)]
|
||||
pub fn new(
|
||||
sample_rate: u32,
|
||||
#[napi(ts_arg_type = "(error: Error | null, samples: Float32Array) => void")]
|
||||
on_audio: CaptureCallback,
|
||||
) -> Result<Self> {
|
||||
validate_sample_rate(sample_rate).map_err(napi::Error::from_reason)?;
|
||||
let mut config = audio_config(DeviceType::Capture, sample_rate);
|
||||
config.capture_mut().set_format(Format::F32);
|
||||
config.capture_mut().set_channels(AUDIO_CHANNELS);
|
||||
let mut device = Device::new(None, &config)
|
||||
.map_err(|error| napi::Error::from_reason(format!("Failed to open the default microphone: {error}")))?;
|
||||
device.set_data_callback(move |_device, _output, input| {
|
||||
if input.sample_count() == 0 {
|
||||
return;
|
||||
}
|
||||
on_audio.call(
|
||||
Ok(Float32Array::new(input.as_samples::<f32>().to_vec())),
|
||||
ThreadsafeFunctionCallMode::NonBlocking,
|
||||
);
|
||||
});
|
||||
device.start().map_err(|error| {
|
||||
napi::Error::from_reason(format!("Failed to start microphone capture: {error}"))
|
||||
})?;
|
||||
Ok(Self { device: Mutex::new(Some(device)) })
|
||||
}
|
||||
|
||||
/// Stop capture immediately and release the microphone.
|
||||
#[napi]
|
||||
pub fn stop(&self) -> Result<()> {
|
||||
let device = self.device.lock().take();
|
||||
let Some(device) = device else {
|
||||
return Ok(());
|
||||
};
|
||||
device
|
||||
.stop()
|
||||
.map_err(|error| napi::Error::from_reason(format!("Failed to stop microphone capture: {error}")))
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AudioCapture {
|
||||
fn drop(&mut self) {
|
||||
if let Some(device) = self.device.get_mut().take() {
|
||||
let _ = device.stop();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Gapless mono `f32` playback through the default speaker.
|
||||
#[napi]
|
||||
pub struct AudioPlayback {
|
||||
stream: Mutex<Option<PlaybackStream>>,
|
||||
state: Arc<PlaybackState>,
|
||||
}
|
||||
|
||||
#[napi]
|
||||
impl AudioPlayback {
|
||||
/// Open the default speaker at the requested logical sample rate.
|
||||
#[napi(constructor)]
|
||||
pub fn new(sample_rate: u32) -> Result<Self> {
|
||||
let stream = PlaybackStream::start(sample_rate).map_err(napi::Error::from_reason)?;
|
||||
let state = stream.state();
|
||||
Ok(Self { stream: Mutex::new(Some(stream)), state })
|
||||
}
|
||||
|
||||
/// Queue mono floating-point PCM in playback order.
|
||||
#[napi]
|
||||
pub fn write(&self, samples: Float32Array) -> Result<()> {
|
||||
let stream = self.stream.lock();
|
||||
let stream = stream
|
||||
.as_ref()
|
||||
.ok_or_else(|| napi::Error::from_reason("Native audio playback is closed"))?;
|
||||
stream
|
||||
.writer()
|
||||
.and_then(|writer| writer.write(&samples))
|
||||
.map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Scale audio at render time so gain changes affect already queued samples.
|
||||
#[napi]
|
||||
pub fn set_gain(&self, gain: f64) -> Result<()> {
|
||||
let stream = self.stream.lock();
|
||||
let stream = stream
|
||||
.as_ref()
|
||||
.ok_or_else(|| napi::Error::from_reason("Native audio playback is closed"))?;
|
||||
stream.set_gain(gain as f32).map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Close input, wait until queued samples reach the speaker, then release it.
|
||||
#[napi]
|
||||
pub async fn end(&self) -> Result<()> {
|
||||
{
|
||||
let mut stream = self.stream.lock();
|
||||
let Some(stream) = stream.as_mut() else {
|
||||
return Ok(());
|
||||
};
|
||||
stream.finish_input();
|
||||
}
|
||||
self.state.wait_for_drain().await;
|
||||
let stream = self.stream.lock().take();
|
||||
if let Some(mut stream) = stream {
|
||||
stream.stop().map_err(napi::Error::from_reason)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Stop immediately and discard all queued samples.
|
||||
#[napi]
|
||||
pub fn stop(&self) -> Result<()> {
|
||||
let stream = self.stream.lock().take();
|
||||
if let Some(mut stream) = stream {
|
||||
stream.stop().map_err(napi::Error::from_reason)?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for AudioPlayback {
|
||||
fn drop(&mut self) {
|
||||
if let Some(mut stream) = self.stream.get_mut().take() {
|
||||
let _ = stream.stop();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn playback_preserves_chunk_order_and_applies_render_gain() {
|
||||
let state = PlaybackState::new();
|
||||
state.set_gain(0.5);
|
||||
let (tx, rx) = flume::unbounded();
|
||||
tx.send(vec![1.0, -1.0]).expect("receiver is live");
|
||||
tx.send(vec![0.5, -0.5]).expect("receiver is live");
|
||||
drop(tx);
|
||||
let mut current = Vec::new();
|
||||
let mut cursor = 0;
|
||||
let mut empty_callbacks = 0;
|
||||
let mut output = [9.0; 5];
|
||||
|
||||
fill_playback(
|
||||
&rx,
|
||||
&mut current,
|
||||
&mut cursor,
|
||||
&mut output,
|
||||
&state,
|
||||
&mut empty_callbacks,
|
||||
);
|
||||
|
||||
assert_eq!(output, [0.5, -0.5, 0.25, -0.25, 0.0]);
|
||||
assert!(!state.drained.load(Ordering::Acquire));
|
||||
let mut silence = [1.0; 2];
|
||||
fill_playback(
|
||||
&rx,
|
||||
&mut current,
|
||||
&mut cursor,
|
||||
&mut silence,
|
||||
&state,
|
||||
&mut empty_callbacks,
|
||||
);
|
||||
assert_eq!(silence, [0.0, 0.0]);
|
||||
assert!(state.drained.load(Ordering::Acquire));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,348 @@
|
||||
//! Apple DeviceCheck token generation (`DCDevice.generateToken`).
|
||||
//!
|
||||
//! Reimplements the flow the ChatGPT desktop app's `devicecheck.node` addon
|
||||
//! uses to mint attestation tokens: resolve `DCDevice.currentDevice`, check
|
||||
//! `isSupported`, then call `generateTokenWithCompletionHandler:` and wait up
|
||||
//! to one second for the completion block, reporting the base64-encoded token
|
||||
//! or the failure reason.
|
||||
//!
|
||||
//! Uses raw Objective-C runtime FFI and a hand-built block literal — no
|
||||
//! `objc2`/`block2` dependency.
|
||||
//!
|
||||
//! # Platform
|
||||
//! - **macOS**: Full implementation via `DeviceCheck.framework`.
|
||||
//! - **Other**: Returns `supported: false` without touching the network.
|
||||
|
||||
use napi_derive::napi;
|
||||
|
||||
use crate::task;
|
||||
|
||||
/// Outcome of a single `DCDevice.generateToken` request.
|
||||
#[napi(object)]
|
||||
pub struct DeviceCheckTokenResult {
|
||||
/// Whether `DCDevice.isSupported` reported attestation support.
|
||||
pub supported: bool,
|
||||
/// Base64-encoded DeviceCheck token; present only when generation succeeded.
|
||||
pub token_base64: Option<String>,
|
||||
/// Human-readable failure reason when no token was produced.
|
||||
pub error: Option<String>,
|
||||
/// Wall-clock time spent in the native call, in milliseconds.
|
||||
pub latency_ms: f64,
|
||||
}
|
||||
|
||||
/// Generate an Apple DeviceCheck attestation token.
|
||||
///
|
||||
/// Resolves with the token (or the error reason) after at most a 1-second
|
||||
/// wait, matching the upstream `devicecheck.node` addon contract.
|
||||
#[napi]
|
||||
pub fn device_check_generate_token() -> task::Promise<DeviceCheckTokenResult> {
|
||||
task::blocking("devicecheck.generate_token", (), move |_| Ok(platform::generate_token()))
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// macOS implementation
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[cfg(target_os = "macos")]
|
||||
mod platform {
|
||||
use std::{
|
||||
ffi::{CStr, c_char, c_void},
|
||||
panic::{AssertUnwindSafe, catch_unwind},
|
||||
ptr,
|
||||
sync::mpsc::{self, SyncSender},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use super::DeviceCheckTokenResult;
|
||||
|
||||
/// How long to wait for the DeviceCheck completion handler before giving
|
||||
/// up, matching the timeout in the upstream `devicecheck.node` addon.
|
||||
const TOKEN_TIMEOUT: Duration = Duration::from_secs(1);
|
||||
|
||||
type Id = *mut c_void;
|
||||
type Sel = *mut c_void;
|
||||
|
||||
// `objc_msgSend` is typed per call signature via `#[link_name]` aliases,
|
||||
// the standard idiom for raw ObjC messaging without an objc crate.
|
||||
#[allow(
|
||||
clashing_extern_declarations,
|
||||
reason = "objc_msgSend is an assembly trampoline that forwards to the method IMP; each alias types the same symbol for a distinct call signature"
|
||||
)]
|
||||
#[link(name = "objc")]
|
||||
unsafe extern "C" {
|
||||
fn objc_getClass(name: *const c_char) -> Id;
|
||||
fn sel_registerName(name: *const c_char) -> Sel;
|
||||
fn objc_retain(obj: Id) -> Id;
|
||||
fn objc_release(obj: Id);
|
||||
fn objc_autoreleasePoolPush() -> *mut c_void;
|
||||
fn objc_autoreleasePoolPop(pool: *mut c_void);
|
||||
|
||||
#[link_name = "objc_msgSend"]
|
||||
fn msg_send_noarg(receiver: Id, selector: Sel) -> Id;
|
||||
#[link_name = "objc_msgSend"]
|
||||
fn msg_send_bool(receiver: Id, selector: Sel) -> u8;
|
||||
#[link_name = "objc_msgSend"]
|
||||
fn msg_send_u64(receiver: Id, selector: Sel, options: u64) -> Id;
|
||||
#[link_name = "objc_msgSend"]
|
||||
fn msg_send_block(receiver: Id, selector: Sel, block: *const c_void);
|
||||
}
|
||||
|
||||
// Linking DeviceCheck.framework registers `DCDevice` with the ObjC
|
||||
// runtime when the addon image loads.
|
||||
#[link(name = "DeviceCheck", kind = "framework")]
|
||||
unsafe extern "C" {}
|
||||
|
||||
unsafe extern "C" {
|
||||
/// Stack-block class from libsystem_blocks; used as the literal's isa.
|
||||
static _NSConcreteStackBlock: *const c_void;
|
||||
}
|
||||
|
||||
/// Outcome delivered once from the completion block to the waiting worker.
|
||||
enum Completion {
|
||||
Token(String),
|
||||
Error(String),
|
||||
}
|
||||
|
||||
/// Objective-C block ABI: the 32-byte literal header followed by the
|
||||
/// captured context (a raw pointer to the channel sender).
|
||||
#[repr(C)]
|
||||
struct CompletionBlock {
|
||||
isa: *const c_void,
|
||||
flags: i32,
|
||||
reserved: i32,
|
||||
invoke: unsafe extern "C" fn(*mut CompletionBlock, Id, Id),
|
||||
descriptor: *const CompletionBlockDescriptor,
|
||||
sender: *const SyncSender<Completion>,
|
||||
}
|
||||
|
||||
/// `Block_descriptor_1` followed immediately by `Block_descriptor_3`.
|
||||
/// No `Block_descriptor_2` (copy/dispose helpers) is emitted because the
|
||||
/// captured sender pointer is plain-old-data and needs no retain/release.
|
||||
#[repr(C)]
|
||||
struct CompletionBlockDescriptor {
|
||||
reserved: usize,
|
||||
size: usize,
|
||||
signature: *const c_char,
|
||||
}
|
||||
|
||||
/// `BLOCK_HAS_SIGNATURE` — the only flag needed for a POD stack block.
|
||||
const BLOCK_HAS_SIGNATURE: i32 = 1 << 30;
|
||||
|
||||
/// Type encoding for `void (^)(NSData *token, NSError *error)`:
|
||||
/// void return, 24 bytes of arguments (block at 0, token at 8, error at 16).
|
||||
const BLOCK_SIGNATURE: &CStr = c"v24@?0@8@16";
|
||||
|
||||
/// Immutable, process-lifetime data; the raw signature pointer is never
|
||||
/// mutated, so shared access from the ObjC runtime is race-free.
|
||||
unsafe impl Sync for CompletionBlockDescriptor {}
|
||||
|
||||
static COMPLETION_DESCRIPTOR: CompletionBlockDescriptor = CompletionBlockDescriptor {
|
||||
reserved: 0,
|
||||
size: size_of::<CompletionBlock>(),
|
||||
signature: BLOCK_SIGNATURE.as_ptr(),
|
||||
};
|
||||
|
||||
/// Resolve a selector by name; `sel_registerName` is idempotent and cheap.
|
||||
///
|
||||
/// # Safety
|
||||
/// The returned selector is valid for the lifetime of the process.
|
||||
unsafe fn selector(name: &CStr) -> Sel {
|
||||
// SAFETY: `name` is a valid null-terminated C string.
|
||||
unsafe { sel_registerName(name.as_ptr()) }
|
||||
}
|
||||
|
||||
/// Copy a C string owned by an autoreleased `NSString` into a Rust `String`.
|
||||
///
|
||||
/// # Safety
|
||||
/// `ptr` must be null or point to a valid null-terminated UTF-8 string that
|
||||
/// outlives the call.
|
||||
unsafe fn copy_c_string(ptr: *const c_char) -> String {
|
||||
if ptr.is_null() {
|
||||
return String::new();
|
||||
}
|
||||
// SAFETY: upheld by the caller; `CStr::from_ptr` only reads.
|
||||
unsafe { CStr::from_ptr(ptr) }
|
||||
.to_string_lossy()
|
||||
.into_owned()
|
||||
}
|
||||
|
||||
/// Read the UTF-8 payload of an `NSString` into a Rust `String`.
|
||||
///
|
||||
/// # Safety
|
||||
/// `string` must be a live `NSString` for the duration of the call.
|
||||
unsafe fn ns_string(string: Id) -> String {
|
||||
// SAFETY: `string` is a live NSString; the returned pointer stays valid
|
||||
// until the enclosing autorelease pool drains.
|
||||
unsafe { copy_c_string(msg_send_noarg(string, selector(c"UTF8String")).cast()) }
|
||||
}
|
||||
|
||||
/// Completion block body. Runs on DeviceCheck's XPC reply queue, which is
|
||||
/// why the result travels over a channel instead of a return value.
|
||||
///
|
||||
/// # Safety
|
||||
/// Called by the Objective-C runtime with a valid block literal; `token`
|
||||
/// and `error` are live `NSData`/`NSError` objects (or null) for the
|
||||
/// duration of the call.
|
||||
unsafe extern "C" fn completion_invoke(block: *mut CompletionBlock, token: Id, error: Id) {
|
||||
let completion = catch_unwind(AssertUnwindSafe(|| {
|
||||
if !token.is_null() {
|
||||
// SAFETY: `token` is a live NSData for the duration of the callback.
|
||||
let encoded =
|
||||
unsafe { msg_send_u64(token, selector(c"base64EncodedStringWithOptions:"), 0) };
|
||||
if encoded.is_null() {
|
||||
return Completion::Error("DeviceCheck returned no token".to_owned());
|
||||
}
|
||||
// SAFETY: `encoded` is a live NSString.
|
||||
return Completion::Token(unsafe { ns_string(encoded) });
|
||||
}
|
||||
if !error.is_null() {
|
||||
// SAFETY: `error` is a live NSError for the duration of the callback.
|
||||
let description = unsafe { msg_send_noarg(error, selector(c"localizedDescription")) };
|
||||
if description.is_null() {
|
||||
return Completion::Error("DeviceCheck token request failed".to_owned());
|
||||
}
|
||||
// SAFETY: `description` is a live NSString.
|
||||
return Completion::Error(unsafe { ns_string(description) });
|
||||
}
|
||||
Completion::Error("DeviceCheck returned no token".to_owned())
|
||||
}));
|
||||
let completion = match completion {
|
||||
Ok(completion) => completion,
|
||||
Err(payload) => {
|
||||
// Never let a panic escape into the ObjC runtime; mirror the
|
||||
// bounded-leak disposal used by `task::Blocking` instead of
|
||||
// dropping a potentially panicking payload type here.
|
||||
std::mem::forget(payload);
|
||||
Completion::Error("DeviceCheck completion panicked".to_owned())
|
||||
},
|
||||
};
|
||||
// SAFETY: the owner keeps the sender alive until the block has fired
|
||||
// (and leaks it on timeout), so the captured pointer is always valid.
|
||||
// `try_send` never blocks the XPC queue, even if the runtime were to
|
||||
// invoke the block more than once.
|
||||
unsafe {
|
||||
_ = (*(*block).sender).try_send(completion);
|
||||
}
|
||||
}
|
||||
|
||||
/// Build the result for a supported device by driving
|
||||
/// `generateTokenWithCompletionHandler:` and waiting on the channel.
|
||||
///
|
||||
/// # Safety
|
||||
/// `device` must be a live, retained `DCDevice` instance.
|
||||
unsafe fn run_token_request(device: Id) -> DeviceCheckTokenResult {
|
||||
let (sender, receiver) = mpsc::sync_channel::<Completion>(1);
|
||||
let sender = Box::into_raw(Box::new(sender));
|
||||
let block = CompletionBlock {
|
||||
isa: ptr::addr_of!(_NSConcreteStackBlock).cast::<c_void>(),
|
||||
flags: BLOCK_HAS_SIGNATURE,
|
||||
reserved: 0,
|
||||
invoke: completion_invoke,
|
||||
descriptor: &raw const COMPLETION_DESCRIPTOR,
|
||||
sender,
|
||||
};
|
||||
// SAFETY: `device` is a live DCDevice and `block` follows the block ABI;
|
||||
// the runtime copies the literal, so the stack frame may die after the call.
|
||||
unsafe {
|
||||
msg_send_block(
|
||||
device,
|
||||
selector(c"generateTokenWithCompletionHandler:"),
|
||||
(&raw const block).cast(),
|
||||
)
|
||||
};
|
||||
|
||||
let mut result = DeviceCheckTokenResult {
|
||||
supported: true,
|
||||
token_base64: None,
|
||||
error: None,
|
||||
latency_ms: 0.0,
|
||||
};
|
||||
match receiver.recv_timeout(TOKEN_TIMEOUT) {
|
||||
Ok(Completion::Token(token)) => {
|
||||
result.token_base64 = Some(token);
|
||||
// SAFETY: the block has fired and will not fire again, so the
|
||||
// sender is unreachable from the runtime and can be reclaimed.
|
||||
drop(unsafe { Box::from_raw(sender) });
|
||||
},
|
||||
Ok(Completion::Error(message)) => {
|
||||
result.error = Some(message);
|
||||
// SAFETY: same as above — the single-shot block already fired.
|
||||
drop(unsafe { Box::from_raw(sender) });
|
||||
},
|
||||
Err(_) => {
|
||||
// Timeout (or a vanished sender): the block may still fire on
|
||||
// the XPC queue, so deliberately leak the sender to keep the
|
||||
// captured pointer valid. Bounded to one leak per timeout.
|
||||
result.error = Some("timed out waiting for DeviceCheck token".to_owned());
|
||||
},
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
fn generate_token_inner() -> DeviceCheckTokenResult {
|
||||
let mut result = DeviceCheckTokenResult {
|
||||
supported: false,
|
||||
token_base64: None,
|
||||
error: None,
|
||||
latency_ms: 0.0,
|
||||
};
|
||||
// SAFETY: `c"DCDevice"` is a valid null-terminated class name.
|
||||
let class = unsafe { objc_getClass(c"DCDevice".as_ptr()) };
|
||||
if class.is_null() {
|
||||
result.error = Some("DeviceCheck framework unavailable".to_owned());
|
||||
return result;
|
||||
}
|
||||
// SAFETY: `class` is a registered ObjC class; `currentDevice` is a
|
||||
// documented DCDevice class method returning an autoreleased instance.
|
||||
let device = unsafe { msg_send_noarg(class, selector(c"currentDevice")) };
|
||||
if device.is_null() {
|
||||
result.error = Some("DeviceCheck currentDevice unavailable".to_owned());
|
||||
return result;
|
||||
}
|
||||
// SAFETY: `device` is a live object; retain balances the release below.
|
||||
let device = unsafe { objc_retain(device) };
|
||||
// SAFETY: `device` is a live DCDevice; `isSupported` returns BOOL.
|
||||
let supported = unsafe { msg_send_bool(device, selector(c"isSupported")) } != 0;
|
||||
if supported {
|
||||
// SAFETY: `device` is live and retained for the duration of the call.
|
||||
return unsafe {
|
||||
let mut token_result = run_token_request(device);
|
||||
objc_release(device);
|
||||
token_result.supported = true;
|
||||
token_result
|
||||
};
|
||||
}
|
||||
// SAFETY: balances the retain above.
|
||||
unsafe { objc_release(device) };
|
||||
result
|
||||
}
|
||||
|
||||
pub fn generate_token() -> DeviceCheckTokenResult {
|
||||
let start = Instant::now();
|
||||
// SAFETY: pool push/pop are balanced within this scope.
|
||||
let pool = unsafe { objc_autoreleasePoolPush() };
|
||||
let mut result = generate_token_inner();
|
||||
result.latency_ms = start.elapsed().as_secs_f64() * 1000.0;
|
||||
// SAFETY: balances the push above.
|
||||
unsafe { objc_autoreleasePoolPop(pool) };
|
||||
result
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Non-macOS stub
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[cfg(not(target_os = "macos"))]
|
||||
mod platform {
|
||||
use super::DeviceCheckTokenResult;
|
||||
|
||||
pub fn generate_token() -> DeviceCheckTokenResult {
|
||||
DeviceCheckTokenResult {
|
||||
supported: false,
|
||||
token_base64: None,
|
||||
error: None,
|
||||
latency_ms: 0.0,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -23,6 +23,7 @@
|
||||
#![feature(alloc_error_hook)]
|
||||
|
||||
pub mod appearance;
|
||||
pub mod audio;
|
||||
pub mod ast;
|
||||
pub mod block;
|
||||
pub mod clipboard;
|
||||
@@ -34,6 +35,7 @@ pub mod desktop;
|
||||
/// pure conversion helpers stay unit-testable without a live X server.
|
||||
#[cfg(any(target_os = "linux", test))]
|
||||
pub mod desktop_x11;
|
||||
pub mod devicecheck;
|
||||
pub mod diff;
|
||||
pub mod fd;
|
||||
pub mod glob;
|
||||
@@ -43,6 +45,7 @@ pub mod highlight;
|
||||
pub mod html;
|
||||
pub mod iofs;
|
||||
pub mod keys;
|
||||
pub mod live;
|
||||
pub mod sixel;
|
||||
pub mod snapcompact;
|
||||
pub use pi_ast::language;
|
||||
|
||||
@@ -0,0 +1,769 @@
|
||||
//! Native WebRTC media transport for Codex live conversations.
|
||||
//!
|
||||
//! The TypeScript host owns authenticated signaling and the sideband protocol;
|
||||
//! this module owns the realtime WebRTC peer, Opus media, and speaker playback.
|
||||
|
||||
use std::{
|
||||
sync::{
|
||||
Arc, Weak,
|
||||
atomic::{AtomicBool, AtomicUsize, Ordering},
|
||||
},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use bytes::Bytes;
|
||||
use napi::{
|
||||
bindgen_prelude::{Float32Array, Result},
|
||||
threadsafe_function::{ThreadsafeFunction, ThreadsafeFunctionCallMode, UnknownReturnValue},
|
||||
};
|
||||
use napi_derive::napi;
|
||||
use opus::{Application, Channels, Decoder, Encoder};
|
||||
use parking_lot::Mutex;
|
||||
use tokio::{sync::watch, task::JoinHandle};
|
||||
use crate::audio::{PlaybackStream, PlaybackWriter};
|
||||
use webrtc::{
|
||||
api::{
|
||||
APIBuilder,
|
||||
interceptor_registry::register_default_interceptors,
|
||||
media_engine::{MIME_TYPE_OPUS, MediaEngine},
|
||||
},
|
||||
data_channel::{RTCDataChannel, data_channel_message::DataChannelMessage},
|
||||
interceptor::registry::Registry,
|
||||
media::Sample,
|
||||
peer_connection::{
|
||||
RTCPeerConnection,
|
||||
configuration::RTCConfiguration,
|
||||
peer_connection_state::RTCPeerConnectionState,
|
||||
sdp::session_description::RTCSessionDescription,
|
||||
},
|
||||
rtp_transceiver::{
|
||||
rtp_codec::{RTCRtpCodecCapability, RTCRtpCodecParameters, RTPCodecType},
|
||||
rtp_sender::RTCRtpSender,
|
||||
},
|
||||
track::{
|
||||
track_local::{TrackLocal, track_local_static_sample::TrackLocalStaticSample},
|
||||
track_remote::TrackRemote,
|
||||
},
|
||||
};
|
||||
|
||||
const DATA_CHANNEL_LABEL: &str = "oai-events";
|
||||
const INPUT_SAMPLE_RATE: u32 = 16_000;
|
||||
const INPUT_FRAME_SAMPLES: usize = 320;
|
||||
const INPUT_FRAME_DURATION: Duration = Duration::from_millis(20);
|
||||
const MAX_ENCODED_OPUS_BYTES: usize = 1_275;
|
||||
const MAX_QUEUED_INPUT_SAMPLES: usize = 32_000;
|
||||
const OUTPUT_SAMPLE_RATE: u32 = 48_000;
|
||||
const MAX_DECODED_OPUS_SAMPLES: usize = 5_760;
|
||||
const OUTPUT_LEVEL_SAMPLES: usize = 2_400;
|
||||
const OUTPUT_FRAME_SAMPLES: usize = 960;
|
||||
const DEFAULT_OPEN_TIMEOUT_MS: u32 = 20_000;
|
||||
const DISCONNECT_GRACE: Duration = Duration::from_secs(2);
|
||||
const CLOSE_TASK_TIMEOUT: Duration = Duration::from_secs(1);
|
||||
|
||||
const OPUS_CAPABILITY: RTCRtpCodecCapability = RTCRtpCodecCapability {
|
||||
mime_type: String::new(),
|
||||
clock_rate: OUTPUT_SAMPLE_RATE,
|
||||
channels: 2,
|
||||
sdp_fmtp_line: String::new(),
|
||||
rtcp_feedback: Vec::new(),
|
||||
};
|
||||
|
||||
type StringCallback = ThreadsafeFunction<String, UnknownReturnValue>;
|
||||
type LevelCallback = ThreadsafeFunction<f64, UnknownReturnValue>;
|
||||
type NativeResult<T> = std::result::Result<T, String>;
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
enum PeerSignal {
|
||||
Connecting,
|
||||
Open,
|
||||
Failed(String),
|
||||
Closed,
|
||||
}
|
||||
|
||||
enum InputCommand {
|
||||
Audio(Vec<f32>),
|
||||
Muted(bool),
|
||||
Close,
|
||||
}
|
||||
|
||||
struct LiveCallbacks {
|
||||
event: StringCallback,
|
||||
level: LevelCallback,
|
||||
failure: StringCallback,
|
||||
}
|
||||
|
||||
struct LiveResources {
|
||||
peer: Arc<RTCPeerConnection>,
|
||||
data_channel: Arc<RTCDataChannel>,
|
||||
input_tx: flume::Sender<InputCommand>,
|
||||
input_task: JoinHandle<()>,
|
||||
rtcp_task: JoinHandle<()>,
|
||||
playback: PlaybackStream,
|
||||
}
|
||||
|
||||
struct LivePeerCore {
|
||||
callbacks: LiveCallbacks,
|
||||
resources: Mutex<Option<LiveResources>>,
|
||||
signal_tx: watch::Sender<PeerSignal>,
|
||||
started: AtomicBool,
|
||||
closing: AtomicBool,
|
||||
muted: AtomicBool,
|
||||
failure_reported: AtomicBool,
|
||||
queued_samples: AtomicUsize,
|
||||
}
|
||||
|
||||
impl LivePeerCore {
|
||||
fn new(callbacks: LiveCallbacks) -> Self {
|
||||
let (signal_tx, _) = watch::channel(PeerSignal::Connecting);
|
||||
Self {
|
||||
callbacks,
|
||||
resources: Mutex::new(None),
|
||||
signal_tx,
|
||||
started: AtomicBool::new(false),
|
||||
closing: AtomicBool::new(false),
|
||||
muted: AtomicBool::new(false),
|
||||
failure_reported: AtomicBool::new(false),
|
||||
queued_samples: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
async fn create_offer(self: &Arc<Self>) -> NativeResult<String> {
|
||||
if self.started.swap(true, Ordering::AcqRel) {
|
||||
return Err("Native live WebRTC peer has already started".to_owned());
|
||||
}
|
||||
if self.closing.load(Ordering::Acquire) {
|
||||
return Err("Native live WebRTC peer is closed".to_owned());
|
||||
}
|
||||
|
||||
let playback = PlaybackStream::start(OUTPUT_SAMPLE_RATE)?;
|
||||
let playback_tx = playback.writer()?;
|
||||
let mut media_engine = MediaEngine::default();
|
||||
let capability = opus_capability();
|
||||
media_engine
|
||||
.register_codec(
|
||||
RTCRtpCodecParameters {
|
||||
capability: capability.clone(),
|
||||
payload_type: 111,
|
||||
..Default::default()
|
||||
},
|
||||
RTPCodecType::Audio,
|
||||
)
|
||||
.map_err(|error| format!("Failed to register the live Opus codec: {error}"))?;
|
||||
let registry = register_default_interceptors(Registry::new(), &mut media_engine)
|
||||
.map_err(|error| format!("Failed to configure live WebRTC interceptors: {error}"))?;
|
||||
let api = APIBuilder::new()
|
||||
.with_media_engine(media_engine)
|
||||
.with_interceptor_registry(registry)
|
||||
.build();
|
||||
let peer = Arc::new(
|
||||
api.new_peer_connection(RTCConfiguration::default())
|
||||
.await
|
||||
.map_err(|error| format!("Failed to create the live WebRTC peer: {error}"))?,
|
||||
);
|
||||
|
||||
let track = Arc::new(TrackLocalStaticSample::new(
|
||||
capability,
|
||||
"audio".to_owned(),
|
||||
"omp-live".to_owned(),
|
||||
));
|
||||
let sender = match peer
|
||||
.add_track(Arc::clone(&track) as Arc<dyn TrackLocal + Send + Sync>)
|
||||
.await
|
||||
{
|
||||
Ok(sender) => sender,
|
||||
Err(error) => {
|
||||
let _ = peer.close().await;
|
||||
return Err(format!("Failed to add the live audio track: {error}"));
|
||||
},
|
||||
};
|
||||
|
||||
install_peer_callbacks(&peer, Arc::downgrade(self), playback_tx);
|
||||
let data_channel = match peer.create_data_channel(DATA_CHANNEL_LABEL, None).await {
|
||||
Ok(channel) => channel,
|
||||
Err(error) => {
|
||||
let _ = peer.close().await;
|
||||
return Err(format!("Failed to create the live data channel: {error}"));
|
||||
},
|
||||
};
|
||||
install_data_channel_callbacks(&data_channel, Arc::downgrade(self));
|
||||
|
||||
let offer = match peer.create_offer(None).await {
|
||||
Ok(offer) => offer,
|
||||
Err(error) => {
|
||||
let _ = peer.close().await;
|
||||
return Err(format!("Failed to create the live SDP offer: {error}"));
|
||||
},
|
||||
};
|
||||
if let Err(error) = peer.set_local_description(offer.clone()).await {
|
||||
let _ = peer.close().await;
|
||||
return Err(format!("Failed to install the live SDP offer: {error}"));
|
||||
}
|
||||
if self.closing.load(Ordering::Acquire) {
|
||||
let _ = peer.close().await;
|
||||
return Err("Native live WebRTC peer was closed while starting".to_owned());
|
||||
}
|
||||
|
||||
let (input_tx, input_rx) = flume::unbounded();
|
||||
let input_task = tokio::spawn(run_input_audio(
|
||||
track,
|
||||
input_rx,
|
||||
Arc::downgrade(self),
|
||||
));
|
||||
let rtcp_task = tokio::spawn(drain_rtcp(sender));
|
||||
let resources = LiveResources {
|
||||
peer,
|
||||
data_channel,
|
||||
input_tx,
|
||||
input_task,
|
||||
rtcp_task,
|
||||
playback,
|
||||
};
|
||||
*self.resources.lock() = Some(resources);
|
||||
Ok(offer.sdp)
|
||||
}
|
||||
|
||||
async fn accept_answer(&self, sdp: String) -> NativeResult<()> {
|
||||
let peer = self
|
||||
.resources
|
||||
.lock()
|
||||
.as_ref()
|
||||
.map(|resources| Arc::clone(&resources.peer))
|
||||
.ok_or_else(|| "Native live WebRTC peer has not started".to_owned())?;
|
||||
let answer = RTCSessionDescription::answer(sdp)
|
||||
.map_err(|error| format!("Codex returned an invalid live SDP answer: {error}"))?;
|
||||
peer.set_remote_description(answer)
|
||||
.await
|
||||
.map_err(|error| format!("Failed to install the live SDP answer: {error}"))
|
||||
}
|
||||
|
||||
async fn wait_for_open(&self, timeout_ms: u32) -> NativeResult<()> {
|
||||
let mut signal_rx = self.signal_tx.subscribe();
|
||||
let wait = async {
|
||||
loop {
|
||||
match signal_rx.borrow().clone() {
|
||||
PeerSignal::Open => return Ok(()),
|
||||
PeerSignal::Failed(message) => return Err(message),
|
||||
PeerSignal::Closed => return Err("Native live WebRTC peer closed before opening".to_owned()),
|
||||
PeerSignal::Connecting => {},
|
||||
}
|
||||
signal_rx
|
||||
.changed()
|
||||
.await
|
||||
.map_err(|_| "Native live WebRTC peer stopped before opening".to_owned())?;
|
||||
}
|
||||
};
|
||||
tokio::time::timeout(Duration::from_millis(u64::from(timeout_ms)), wait)
|
||||
.await
|
||||
.map_err(|_| "Timed out waiting for the live data channel to open".to_owned())?
|
||||
}
|
||||
|
||||
fn push_audio(&self, samples: &[f32]) -> NativeResult<()> {
|
||||
if samples.is_empty() || self.muted.load(Ordering::Acquire) {
|
||||
return Ok(());
|
||||
}
|
||||
let input_tx = self
|
||||
.resources
|
||||
.lock()
|
||||
.as_ref()
|
||||
.map(|resources| resources.input_tx.clone())
|
||||
.ok_or_else(|| "Native live WebRTC peer has not started".to_owned())?;
|
||||
let sample_count = samples.len().min(MAX_QUEUED_INPUT_SAMPLES);
|
||||
let retained = &samples[samples.len() - sample_count..];
|
||||
let queued = self.queued_samples.fetch_add(sample_count, Ordering::AcqRel);
|
||||
if queued.saturating_add(sample_count) > MAX_QUEUED_INPUT_SAMPLES {
|
||||
self.queued_samples.fetch_sub(sample_count, Ordering::AcqRel);
|
||||
return Ok(());
|
||||
}
|
||||
if input_tx.send(InputCommand::Audio(retained.to_vec())).is_err() {
|
||||
self.queued_samples.fetch_sub(sample_count, Ordering::AcqRel);
|
||||
return Err("Native live audio input is closed".to_owned());
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn set_muted(&self, muted: bool) -> NativeResult<()> {
|
||||
self.muted.store(muted, Ordering::Release);
|
||||
let input_tx = self
|
||||
.resources
|
||||
.lock()
|
||||
.as_ref()
|
||||
.map(|resources| resources.input_tx.clone());
|
||||
if let Some(input_tx) = input_tx {
|
||||
input_tx
|
||||
.send(InputCommand::Muted(muted))
|
||||
.map_err(|_| "Native live audio input is closed".to_owned())?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn report_event(&self, payload: String) {
|
||||
self.callbacks
|
||||
.event
|
||||
.call(Ok(payload), ThreadsafeFunctionCallMode::NonBlocking);
|
||||
}
|
||||
|
||||
fn report_level(&self, level: f64) {
|
||||
self.callbacks
|
||||
.level
|
||||
.call(Ok(level.clamp(0.0, 1.0)), ThreadsafeFunctionCallMode::NonBlocking);
|
||||
}
|
||||
|
||||
fn mark_open(&self) {
|
||||
if !self.closing.load(Ordering::Acquire) {
|
||||
self.signal_tx.send_replace(PeerSignal::Open);
|
||||
}
|
||||
}
|
||||
|
||||
fn report_failure(&self, message: String) {
|
||||
if self.closing.load(Ordering::Acquire) || self.failure_reported.swap(true, Ordering::AcqRel) {
|
||||
return;
|
||||
}
|
||||
self.signal_tx.send_replace(PeerSignal::Failed(message.clone()));
|
||||
self.callbacks
|
||||
.failure
|
||||
.call(Ok(message), ThreadsafeFunctionCallMode::NonBlocking);
|
||||
}
|
||||
|
||||
async fn close(&self) {
|
||||
if self.closing.swap(true, Ordering::AcqRel) {
|
||||
let mut signal_rx = self.signal_tx.subscribe();
|
||||
while !matches!(*signal_rx.borrow(), PeerSignal::Closed) {
|
||||
if signal_rx.changed().await.is_err() {
|
||||
break;
|
||||
}
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
let resources = self.resources.lock().take();
|
||||
if let Some(mut resources) = resources {
|
||||
let _ = resources.input_tx.send(InputCommand::Close);
|
||||
let _ = resources.peer.close().await;
|
||||
let _ = resources.playback.stop();
|
||||
let _ = tokio::time::timeout(CLOSE_TASK_TIMEOUT, resources.input_task).await;
|
||||
resources.rtcp_task.abort();
|
||||
let _ = resources.rtcp_task.await;
|
||||
drop(resources.data_channel);
|
||||
}
|
||||
self.queued_samples.store(0, Ordering::Release);
|
||||
self.signal_tx.send_replace(PeerSignal::Closed);
|
||||
}
|
||||
}
|
||||
|
||||
/// WebRTC peer that accepts 16 kHz mono PCM and renders remote Opus audio.
|
||||
#[napi]
|
||||
pub struct LiveWebRtcPeer {
|
||||
inner: Arc<LivePeerCore>,
|
||||
}
|
||||
|
||||
#[napi]
|
||||
impl LiveWebRtcPeer {
|
||||
/// Create an idle peer and register its event, output-level, and failure callbacks.
|
||||
#[napi(constructor)]
|
||||
pub fn new(
|
||||
#[napi(ts_arg_type = "(error: Error | null, payload: string) => void")]
|
||||
on_event: StringCallback,
|
||||
#[napi(ts_arg_type = "(error: Error | null, level: number) => void")]
|
||||
on_level: LevelCallback,
|
||||
#[napi(ts_arg_type = "(error: Error | null, message: string) => void")]
|
||||
on_failure: StringCallback,
|
||||
) -> Self {
|
||||
Self {
|
||||
inner: Arc::new(LivePeerCore::new(LiveCallbacks {
|
||||
event: on_event,
|
||||
level: on_level,
|
||||
failure: on_failure,
|
||||
})),
|
||||
}
|
||||
}
|
||||
|
||||
/// Start the native media peer and return its SDP offer.
|
||||
#[napi]
|
||||
pub async fn create_offer(&self) -> Result<String> {
|
||||
self.inner.create_offer().await.map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Apply the remote SDP answer returned by Codex signaling.
|
||||
#[napi]
|
||||
pub async fn accept_answer(&self, sdp: String) -> Result<()> {
|
||||
self.inner.accept_answer(sdp).await.map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Wait until the `oai-events` data channel is open.
|
||||
#[napi]
|
||||
pub async fn wait_for_open(&self, timeout_ms: Option<u32>) -> Result<()> {
|
||||
self.inner
|
||||
.wait_for_open(timeout_ms.unwrap_or(DEFAULT_OPEN_TIMEOUT_MS))
|
||||
.await
|
||||
.map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Queue 16 kHz mono floating-point PCM for Opus transmission.
|
||||
#[napi]
|
||||
pub fn push_audio(&self, samples: Float32Array) -> Result<()> {
|
||||
self.inner.push_audio(&samples).map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Enable or disable microphone transmission, discarding partial muted frames.
|
||||
#[napi]
|
||||
pub fn set_muted(&self, muted: bool) -> Result<()> {
|
||||
self.inner.set_muted(muted).map_err(napi::Error::from_reason)
|
||||
}
|
||||
|
||||
/// Close media, the data channel, the peer connection, and speaker playback.
|
||||
#[napi]
|
||||
pub async fn close(&self) {
|
||||
self.inner.close().await;
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for LiveWebRtcPeer {
|
||||
fn drop(&mut self) {
|
||||
if self.inner.closing.load(Ordering::Acquire) {
|
||||
return;
|
||||
}
|
||||
let inner = Arc::clone(&self.inner);
|
||||
if let Ok(runtime) = tokio::runtime::Handle::try_current() {
|
||||
runtime.spawn(async move {
|
||||
inner.close().await;
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn opus_capability() -> RTCRtpCodecCapability {
|
||||
RTCRtpCodecCapability {
|
||||
mime_type: MIME_TYPE_OPUS.to_owned(),
|
||||
clock_rate: OPUS_CAPABILITY.clock_rate,
|
||||
channels: OPUS_CAPABILITY.channels,
|
||||
sdp_fmtp_line: "minptime=10;useinbandfec=1".to_owned(),
|
||||
rtcp_feedback: Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fn install_peer_callbacks(
|
||||
peer: &Arc<RTCPeerConnection>,
|
||||
core: Weak<LivePeerCore>,
|
||||
playback_tx: PlaybackWriter,
|
||||
) {
|
||||
let output_sender = Arc::new(Mutex::new(Some(playback_tx)));
|
||||
let output_sender_for_track = Arc::clone(&output_sender);
|
||||
let core_for_track = core.clone();
|
||||
peer.on_track(Box::new(move |track, _receiver, _transceiver| {
|
||||
let output_sender = output_sender_for_track.lock().take();
|
||||
let core = core_for_track.clone();
|
||||
Box::pin(async move {
|
||||
if track.kind() != RTPCodecType::Audio {
|
||||
return;
|
||||
}
|
||||
let Some(output_sender) = output_sender else {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure("Codex live returned more than one remote audio track".to_owned());
|
||||
}
|
||||
return;
|
||||
};
|
||||
tokio::spawn(receive_output_audio(track, output_sender, core));
|
||||
})
|
||||
}));
|
||||
|
||||
let peer_for_state = Arc::downgrade(peer);
|
||||
peer.on_peer_connection_state_change(Box::new(move |state| {
|
||||
let core = core.clone();
|
||||
let peer = peer_for_state.clone();
|
||||
Box::pin(async move {
|
||||
let Some(core) = core.upgrade() else {
|
||||
return;
|
||||
};
|
||||
match state {
|
||||
RTCPeerConnectionState::Failed => {
|
||||
core.report_failure("Live WebRTC peer connection failed".to_owned());
|
||||
},
|
||||
RTCPeerConnectionState::Closed => {
|
||||
if !core.closing.load(Ordering::Acquire) {
|
||||
core.report_failure("Live WebRTC peer connection closed unexpectedly".to_owned());
|
||||
}
|
||||
},
|
||||
RTCPeerConnectionState::Disconnected => {
|
||||
tokio::time::sleep(DISCONNECT_GRACE).await;
|
||||
if peer
|
||||
.upgrade()
|
||||
.is_some_and(|peer| peer.connection_state() == RTCPeerConnectionState::Disconnected)
|
||||
{
|
||||
core.report_failure("Live WebRTC peer connection disconnected".to_owned());
|
||||
}
|
||||
},
|
||||
_ => {},
|
||||
}
|
||||
})
|
||||
}));
|
||||
}
|
||||
|
||||
fn install_data_channel_callbacks(data_channel: &Arc<RTCDataChannel>, core: Weak<LivePeerCore>) {
|
||||
let core_for_open = core.clone();
|
||||
data_channel.on_open(Box::new(move || {
|
||||
let core = core_for_open.clone();
|
||||
Box::pin(async move {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.mark_open();
|
||||
}
|
||||
})
|
||||
}));
|
||||
|
||||
let core_for_message = core.clone();
|
||||
data_channel.on_message(Box::new(move |message: DataChannelMessage| {
|
||||
let core = core_for_message.clone();
|
||||
Box::pin(async move {
|
||||
if !message.is_string {
|
||||
return;
|
||||
}
|
||||
if let (Some(core), Ok(payload)) = (core.upgrade(), String::from_utf8(message.data.to_vec())) {
|
||||
core.report_event(payload);
|
||||
}
|
||||
})
|
||||
}));
|
||||
|
||||
let core_for_close = core.clone();
|
||||
data_channel.on_close(Box::new(move || {
|
||||
let core = core_for_close.clone();
|
||||
Box::pin(async move {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure("Live data channel closed unexpectedly".to_owned());
|
||||
}
|
||||
})
|
||||
}));
|
||||
|
||||
data_channel.on_error(Box::new(move |error| {
|
||||
let core = core.clone();
|
||||
Box::pin(async move {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Live data channel failed: {error}"));
|
||||
}
|
||||
})
|
||||
}));
|
||||
}
|
||||
|
||||
async fn run_input_audio(
|
||||
track: Arc<TrackLocalStaticSample>,
|
||||
input_rx: flume::Receiver<InputCommand>,
|
||||
core: Weak<LivePeerCore>,
|
||||
) {
|
||||
let mut encoder = match Encoder::new(INPUT_SAMPLE_RATE, Channels::Mono, Application::Voip) {
|
||||
Ok(encoder) => encoder,
|
||||
Err(error) => {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Failed to initialize the live Opus encoder: {error}"));
|
||||
}
|
||||
return;
|
||||
},
|
||||
};
|
||||
if let Err(error) = encoder.set_inband_fec(true) {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Failed to configure the live Opus encoder: {error}"));
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
let mut muted = false;
|
||||
let mut pending = Vec::with_capacity(INPUT_FRAME_SAMPLES * 2);
|
||||
let mut encoded = [0u8; MAX_ENCODED_OPUS_BYTES];
|
||||
let mut ticker = tokio::time::interval(INPUT_FRAME_DURATION);
|
||||
ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Burst);
|
||||
ticker.tick().await;
|
||||
loop {
|
||||
tokio::select! {
|
||||
biased;
|
||||
command = input_rx.recv_async() => {
|
||||
let Ok(command) = command else {
|
||||
break;
|
||||
};
|
||||
match command {
|
||||
InputCommand::Audio(samples) => {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.queued_samples.fetch_sub(samples.len(), Ordering::AcqRel);
|
||||
}
|
||||
if muted {
|
||||
continue;
|
||||
}
|
||||
if samples.len() >= MAX_QUEUED_INPUT_SAMPLES {
|
||||
pending.clear();
|
||||
pending.extend_from_slice(&samples[samples.len() - MAX_QUEUED_INPUT_SAMPLES..]);
|
||||
continue;
|
||||
}
|
||||
let overflow = pending
|
||||
.len()
|
||||
.saturating_add(samples.len())
|
||||
.saturating_sub(MAX_QUEUED_INPUT_SAMPLES);
|
||||
if overflow > 0 {
|
||||
pending.drain(..overflow);
|
||||
}
|
||||
pending.extend_from_slice(&samples);
|
||||
},
|
||||
InputCommand::Muted(next_muted) => {
|
||||
muted = next_muted;
|
||||
pending.clear();
|
||||
},
|
||||
InputCommand::Close => break,
|
||||
}
|
||||
},
|
||||
_ = ticker.tick() => {
|
||||
let mut frame = [0.0f32; INPUT_FRAME_SAMPLES];
|
||||
if !muted {
|
||||
let consumed = pending.len().min(INPUT_FRAME_SAMPLES);
|
||||
frame[..consumed].copy_from_slice(&pending[..consumed]);
|
||||
if consumed > 0 {
|
||||
pending.copy_within(consumed.., 0);
|
||||
pending.truncate(pending.len() - consumed);
|
||||
}
|
||||
}
|
||||
let encoded_len = match encoder.encode_float(&frame, &mut encoded) {
|
||||
Ok(encoded_len) => encoded_len,
|
||||
Err(error) => {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Failed to encode live microphone audio: {error}"));
|
||||
}
|
||||
return;
|
||||
},
|
||||
};
|
||||
let sample = Sample {
|
||||
data: Bytes::copy_from_slice(&encoded[..encoded_len]),
|
||||
duration: INPUT_FRAME_DURATION,
|
||||
..Default::default()
|
||||
};
|
||||
if let Err(error) = track.write_sample(&sample).await {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Failed to send live microphone audio: {error}"));
|
||||
}
|
||||
return;
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn drain_rtcp(sender: Arc<RTCRtpSender>) {
|
||||
while sender.read_rtcp().await.is_ok() {}
|
||||
}
|
||||
|
||||
async fn receive_output_audio(
|
||||
track: Arc<TrackRemote>,
|
||||
playback_tx: PlaybackWriter,
|
||||
core: Weak<LivePeerCore>,
|
||||
) {
|
||||
if !track.codec().capability.mime_type.eq_ignore_ascii_case(MIME_TYPE_OPUS) {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!(
|
||||
"Codex live negotiated unsupported audio codec {}",
|
||||
track.codec().capability.mime_type
|
||||
));
|
||||
}
|
||||
return;
|
||||
}
|
||||
let mut decoder = match Decoder::new(OUTPUT_SAMPLE_RATE, Channels::Mono) {
|
||||
Ok(decoder) => decoder,
|
||||
Err(error) => {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Failed to initialize the live Opus decoder: {error}"));
|
||||
}
|
||||
return;
|
||||
},
|
||||
};
|
||||
let mut decoded = [0.0f32; MAX_DECODED_OPUS_SAMPLES];
|
||||
let mut expected_sequence: Option<u16> = None;
|
||||
let mut level = OutputLevel::default();
|
||||
|
||||
loop {
|
||||
let packet = match track.read_rtp().await {
|
||||
Ok((packet, _attributes)) => packet,
|
||||
Err(error) => {
|
||||
if let Some(core) = core.upgrade()
|
||||
&& !core.closing.load(Ordering::Acquire)
|
||||
{
|
||||
core.report_failure(format!("Live remote audio track failed: {error}"));
|
||||
}
|
||||
return;
|
||||
},
|
||||
};
|
||||
let sequence = packet.header.sequence_number;
|
||||
if let Some(expected) = expected_sequence {
|
||||
let gap = sequence.wrapping_sub(expected);
|
||||
if gap >= u16::MAX / 2 {
|
||||
continue;
|
||||
}
|
||||
if gap > 0 {
|
||||
for _ in 1..gap.min(5) {
|
||||
if let Ok(samples) =
|
||||
decoder.decode_float(&[], &mut decoded[..OUTPUT_FRAME_SAMPLES], false)
|
||||
{
|
||||
if !write_output(&playback_tx, &decoded[..samples], &core) {
|
||||
return;
|
||||
}
|
||||
level.observe(&decoded[..samples], &core);
|
||||
}
|
||||
}
|
||||
if let Ok(samples) = decoder.decode_float(&packet.payload, &mut decoded, true) {
|
||||
if !write_output(&playback_tx, &decoded[..samples], &core) {
|
||||
return;
|
||||
}
|
||||
level.observe(&decoded[..samples], &core);
|
||||
}
|
||||
}
|
||||
}
|
||||
expected_sequence = Some(sequence.wrapping_add(1));
|
||||
match decoder.decode_float(&packet.payload, &mut decoded, false) {
|
||||
Ok(samples) => {
|
||||
if !write_output(&playback_tx, &decoded[..samples], &core) {
|
||||
return;
|
||||
}
|
||||
level.observe(&decoded[..samples], &core);
|
||||
},
|
||||
Err(error) => {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_failure(format!("Failed to decode live speaker audio: {error}"));
|
||||
}
|
||||
return;
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn write_output(playback_tx: &PlaybackWriter, samples: &[f32], core: &Weak<LivePeerCore>) -> bool {
|
||||
match playback_tx.write(samples) {
|
||||
Ok(()) => true,
|
||||
Err(error) => {
|
||||
if let Some(core) = core.upgrade()
|
||||
&& !core.closing.load(Ordering::Acquire)
|
||||
{
|
||||
core.report_failure(format!("Live speaker playback failed: {error}"));
|
||||
}
|
||||
false
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
struct OutputLevel {
|
||||
sum_squares: f64,
|
||||
samples: usize,
|
||||
}
|
||||
|
||||
impl OutputLevel {
|
||||
fn observe(&mut self, decoded: &[f32], core: &Weak<LivePeerCore>) {
|
||||
let mut offset = 0;
|
||||
while offset < decoded.len() {
|
||||
let take = (OUTPUT_LEVEL_SAMPLES - self.samples).min(decoded.len() - offset);
|
||||
for &sample in &decoded[offset..offset + take] {
|
||||
self.sum_squares += f64::from(sample) * f64::from(sample);
|
||||
}
|
||||
self.samples += take;
|
||||
offset += take;
|
||||
if self.samples == OUTPUT_LEVEL_SAMPLES {
|
||||
if let Some(core) = core.upgrade() {
|
||||
core.report_level((self.sum_squares / self.samples as f64).sqrt());
|
||||
}
|
||||
self.sum_squares = 0.0;
|
||||
self.samples = 0;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user