openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-02 19:43:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-02-23 14:40:22 -08:00
parent 19d9d7c6a5
commit 128a7d2ae5
4 changed files with 139 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/chatwidget.rs
+19 -8
View File
@@ -255,6 +255,7 @@ use crate::tui::FrameRequester;
mod interrupts;
use self::interrupts::InterruptManager;
mod agent;
use self::agent::ChatWidgetOpSenders;
use self::agent::spawn_agent;
use self::agent::spawn_agent_from_existing;
pub(crate) use self::agent::spawn_op_forwarder;
@@ -530,6 +531,7 @@ enum RealtimeConversationUiState {
pub(crate) struct ChatWidget {
app_event_tx: AppEventSender,
codex_op_tx: UnboundedSender<Op>,
realtime_audio_op_tx: tokio::sync::mpsc::Sender<Op>,
bottom_pane: BottomPane,
active_cell: Option<Box<dyn HistoryCell>>,
/// Monotonic-ish counter used to invalidate transcript overlay caching.
@@ -2704,7 +2706,10 @@ impl ChatWidget {
let prevent_idle_sleep = config.features.enabled(Feature::PreventIdleSleep);
let mut rng = rand::rng();
let placeholder = PLACEHOLDERS[rng.random_range(0..PLACEHOLDERS.len())].to_string();
let codex_op_tx = spawn_agent(config.clone(), app_event_tx.clone(), thread_manager);
let ChatWidgetOpSenders {
codex_op_tx,
realtime_audio_op_tx,
} = spawn_agent(config.clone(), app_event_tx.clone(), thread_manager);
let model_override = model.as_deref();
let model_for_header = model
@@ -2736,6 +2741,7 @@ impl ChatWidget {
app_event_tx: app_event_tx.clone(),
frame_requester: frame_requester.clone(),
codex_op_tx,
realtime_audio_op_tx,
bottom_pane: BottomPane::new(BottomPaneParams {
frame_requester,
app_event_tx,
@@ -2853,10 +2859,7 @@ impl ChatWidget {
widget
}
pub(crate) fn new_with_op_sender(
common: ChatWidgetInit,
codex_op_tx: UnboundedSender<Op>,
) -> Self {
pub(crate) fn new_with_op_sender(common: ChatWidgetInit, senders: ChatWidgetOpSenders) -> Self {
let ChatWidgetInit {
config,
frame_requester,
@@ -2878,6 +2881,10 @@ impl ChatWidget {
let prevent_idle_sleep = config.features.enabled(Feature::PreventIdleSleep);
let mut rng = rand::rng();
let placeholder = PLACEHOLDERS[rng.random_range(0..PLACEHOLDERS.len())].to_string();
let ChatWidgetOpSenders {
codex_op_tx,
realtime_audio_op_tx,
} = senders;
let model_override = model.as_deref();
let model_for_header = model
@@ -2909,6 +2916,7 @@ impl ChatWidget {
app_event_tx: app_event_tx.clone(),
frame_requester: frame_requester.clone(),
codex_op_tx,
realtime_audio_op_tx,
bottom_pane: BottomPane::new(BottomPaneParams {
frame_requester,
app_event_tx,
@@ -3051,8 +3059,10 @@ impl ChatWidget {
.unwrap_or(header_model);
let current_cwd = Some(session_configured.cwd.clone());
let codex_op_tx =
spawn_agent_from_existing(conversation, session_configured, app_event_tx.clone());
let ChatWidgetOpSenders {
codex_op_tx,
realtime_audio_op_tx,
} = spawn_agent_from_existing(conversation, session_configured, app_event_tx.clone());
let fallback_default = Settings {
model: header_model.clone(),
@@ -3071,6 +3081,7 @@ impl ChatWidget {
app_event_tx: app_event_tx.clone(),
frame_requester: frame_requester.clone(),
codex_op_tx,
realtime_audio_op_tx,
bottom_pane: BottomPane::new(BottomPaneParams {
frame_requester,
app_event_tx,
@@ -3409,7 +3420,7 @@ impl ChatWidget {
return;
}
match RealtimeAudioController::start(self.codex_op_tx.clone()) {
match RealtimeAudioController::start(self.realtime_audio_op_tx.clone()) {
Ok(controller) => {
self.realtime_audio_controller = Some(controller);
self.realtime_conversation_state = RealtimeConversationUiState::Working;
codex-rs/tui/src/chatwidget/agent.rs
+44 -6
View File
@@ -7,20 +7,32 @@ use codex_core::config::Config;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use tokio::sync::mpsc::Receiver;
use tokio::sync::mpsc::Sender;
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::mpsc::channel;
use tokio::sync::mpsc::unbounded_channel;
use crate::app_event::AppEvent;
use crate::app_event_sender::AppEventSender;
const REALTIME_MIC_AUDIO_QUEUE_CAPACITY: usize = 64;
pub(crate) struct ChatWidgetOpSenders {
pub(crate) codex_op_tx: UnboundedSender<Op>,
pub(crate) realtime_audio_op_tx: Sender<Op>,
}
/// Spawn the agent bootstrapper and op forwarding loop, returning the
/// `UnboundedSender<Op>` used by the UI to submit operations.
pub(crate) fn spawn_agent(
config: Config,
app_event_tx: AppEventSender,
server: Arc<ThreadManager>,
) -> UnboundedSender<Op> {
) -> ChatWidgetOpSenders {
let (codex_op_tx, mut codex_op_rx) = unbounded_channel::<Op>();
let (realtime_audio_op_tx, realtime_audio_op_rx) =
channel::<Op>(REALTIME_MIC_AUDIO_QUEUE_CAPACITY);
let app_event_tx_clone = app_event_tx;
tokio::spawn(async move {
@@ -52,6 +64,7 @@ pub(crate) fn spawn_agent(
app_event_tx_clone.send(AppEvent::CodexEvent(ev));
let thread_clone = thread.clone();
spawn_bounded_op_forwarder(thread.clone(), realtime_audio_op_rx);
tokio::spawn(async move {
while let Some(op) = codex_op_rx.recv().await {
let id = thread_clone.submit(op).await;
@@ -72,7 +85,10 @@ pub(crate) fn spawn_agent(
}
});
codex_op_tx
ChatWidgetOpSenders {
codex_op_tx,
realtime_audio_op_tx,
}
}
/// Spawn agent loops for an existing thread (e.g., a forked thread).
@@ -82,8 +98,10 @@ pub(crate) fn spawn_agent_from_existing(
thread: std::sync::Arc<CodexThread>,
session_configured: codex_protocol::protocol::SessionConfiguredEvent,
app_event_tx: AppEventSender,
) -> UnboundedSender<Op> {
) -> ChatWidgetOpSenders {
let (codex_op_tx, mut codex_op_rx) = unbounded_channel::<Op>();
let (realtime_audio_op_tx, realtime_audio_op_rx) =
channel::<Op>(REALTIME_MIC_AUDIO_QUEUE_CAPACITY);
let app_event_tx_clone = app_event_tx;
tokio::spawn(async move {
@@ -95,6 +113,7 @@ pub(crate) fn spawn_agent_from_existing(
app_event_tx_clone.send(AppEvent::CodexEvent(ev));
let thread_clone = thread.clone();
spawn_bounded_op_forwarder(thread.clone(), realtime_audio_op_rx);
tokio::spawn(async move {
while let Some(op) = codex_op_rx.recv().await {
let id = thread_clone.submit(op).await;
@@ -115,13 +134,19 @@ pub(crate) fn spawn_agent_from_existing(
}
});
codex_op_tx
ChatWidgetOpSenders {
codex_op_tx,
realtime_audio_op_tx,
}
}
/// Spawn an op-forwarding loop for an existing thread without subscribing to events.
pub(crate) fn spawn_op_forwarder(thread: std::sync::Arc<CodexThread>) -> UnboundedSender<Op> {
pub(crate) fn spawn_op_forwarder(thread: std::sync::Arc<CodexThread>) -> ChatWidgetOpSenders {
let (codex_op_tx, mut codex_op_rx) = unbounded_channel::<Op>();
let (realtime_audio_op_tx, realtime_audio_op_rx) =
channel::<Op>(REALTIME_MIC_AUDIO_QUEUE_CAPACITY);
spawn_bounded_op_forwarder(thread.clone(), realtime_audio_op_rx);
tokio::spawn(async move {
while let Some(op) = codex_op_rx.recv().await {
if let Err(e) = thread.submit(op).await {
@@ -130,5 +155,18 @@ pub(crate) fn spawn_op_forwarder(thread: std::sync::Arc<CodexThread>) -> Unbound
}
});
codex_op_tx
ChatWidgetOpSenders {
codex_op_tx,
realtime_audio_op_tx,
}
}
fn spawn_bounded_op_forwarder(thread: std::sync::Arc<CodexThread>, mut op_rx: Receiver<Op>) {
tokio::spawn(async move {
while let Some(op) = op_rx.recv().await {
if let Err(e) = thread.submit(op).await {
tracing::error!("failed to submit realtime audio op: {e}");
}
}
});
}
codex-rs/tui/src/chatwidget/tests.rs
+2
View File
@@ -1598,9 +1598,11 @@ async fn make_chatwidget_manual(
},
};
let current_collaboration_mode = base_mode;
let (realtime_audio_op_tx, _realtime_audio_op_rx) = tokio::sync::mpsc::channel(1);
let mut widget = ChatWidget {
app_event_tx,
codex_op_tx: op_tx,
realtime_audio_op_tx,
bottom_pane: bottom,
active_cell: None,
active_cell_revision: 0,
codex-rs/tui/src/realtime_audio.rs
+74 -23
View File
@@ -1,15 +1,16 @@
use anyhow::Result;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RealtimeAudioFrame;
use tokio::sync::mpsc::UnboundedSender;
use tokio::sync::mpsc::Sender;
#[cfg(not(test))]
use tokio::sync::mpsc::error::TrySendError;
#[cfg(not(test))]
use anyhow::Context;
#[cfg(not(test))]
use base64::Engine;
#[cfg(not(test))]
use codex_protocol::protocol::ConversationAudioParams;
#[cfg(not(test))]
use cpal::traits::DeviceTrait;
#[cfg(not(test))]
use std::collections::VecDeque;
@@ -32,6 +33,9 @@ const PLAYBACK_BUFFER_SECONDS: usize = 5;
const MAX_AUDIO_FRAME_DECODED_BYTES: usize = 128 * 1024;
#[cfg(not(test))]
const MAX_AUDIO_FRAME_ENCODED_BYTES: usize = 192 * 1024;
const MIN_REALTIME_PLAYBACK_SAMPLE_RATE: u32 = 8_000;
const MAX_REALTIME_PLAYBACK_SAMPLE_RATE: u32 = 192_000;
const MAX_RESAMPLED_MONO_SAMPLES_PER_FRAME: usize = 480_000;
pub(crate) struct RealtimeAudioController {
backend: RealtimeAudioBackend,
@@ -52,10 +56,10 @@ struct LiveRealtimeAudioController {
}
impl RealtimeAudioController {
pub(crate) fn start(codex_op_tx: UnboundedSender<Op>) -> Result<Self> {
pub(crate) fn start(realtime_audio_op_tx: Sender<Op>) -> Result<Self> {
#[cfg(test)]
{
let _ = codex_op_tx;
let _ = realtime_audio_op_tx;
Ok(Self {
backend: RealtimeAudioBackend::Stub,
})
@@ -91,7 +95,7 @@ impl RealtimeAudioController {
output_config.channels,
)));
let mic_state = Arc::new(Mutex::new(MicCaptureState::new(
codex_op_tx,
realtime_audio_op_tx,
input_config.sample_rate.0,
input_config.channels,
)));
@@ -152,7 +156,7 @@ impl RealtimeAudioController {
#[cfg(not(test))]
#[derive(Debug)]
struct MicCaptureState {
codex_op_tx: UnboundedSender<Op>,
realtime_audio_op_tx: Sender<Op>,
source_sample_rate: u32,
source_channels: u16,
source_mono: Vec<f32>,
@@ -163,12 +167,12 @@ struct MicCaptureState {
#[cfg(not(test))]
impl MicCaptureState {
fn new(
codex_op_tx: UnboundedSender<Op>,
realtime_audio_op_tx: Sender<Op>,
source_sample_rate: u32,
source_channels: u16,
) -> Self {
Self {
codex_op_tx,
realtime_audio_op_tx,
source_sample_rate,
source_channels,
source_mono: Vec::new(),
@@ -247,9 +251,15 @@ impl MicCaptureState {
samples_per_channel: Some(TARGET_SAMPLES_PER_CHANNEL),
},
});
if let Err(err) = self.codex_op_tx.send(op) {
warn!("failed to send realtime microphone frame: {err}");
break;
match self.realtime_audio_op_tx.try_send(op) {
Ok(()) => {}
Err(TrySendError::Full(_)) => {
warn!("dropping realtime microphone frame due to full TUI audio queue");
}
Err(TrySendError::Closed(_)) => {
warn!("failed to send realtime microphone frame: channel closed");
break;
}
}
}
}
@@ -317,9 +327,27 @@ impl PlaybackState {
warn!("dropping realtime audio frame with zero sample rate");
return Ok(());
}
if !is_supported_realtime_playback_sample_rate(frame.sample_rate) {
warn!(
sample_rate = frame.sample_rate,
min_sample_rate = MIN_REALTIME_PLAYBACK_SAMPLE_RATE,
max_sample_rate = MAX_REALTIME_PLAYBACK_SAMPLE_RATE,
"dropping realtime audio frame with unsupported sample rate"
);
return Ok(());
}
let resampled =
resample_linear_mono(&mono, frame.sample_rate, self.output_sample_rate.max(1));
let Some(resampled) =
resample_linear_mono(&mono, frame.sample_rate, self.output_sample_rate.max(1))
else {
warn!(
src_rate = frame.sample_rate,
dst_rate = self.output_sample_rate.max(1),
max_samples = MAX_RESAMPLED_MONO_SAMPLES_PER_FRAME,
"dropping realtime audio frame due to oversized resampled output"
);
return Ok(());
};
for sample in resampled {
for _ in 0..self.output_channels {
self.queue.push_back(sample);
@@ -353,18 +381,23 @@ fn interleaved_i16_to_mono_f32(samples: &[i16], num_channels: u16) -> Vec<f32> {
mono
}
fn resample_linear_mono(input: &[f32], src_rate: u32, dst_rate: u32) -> Vec<f32> {
fn is_supported_realtime_playback_sample_rate(sample_rate: u32) -> bool {
(MIN_REALTIME_PLAYBACK_SAMPLE_RATE..=MAX_REALTIME_PLAYBACK_SAMPLE_RATE).contains(&sample_rate)
}
fn resample_linear_mono(input: &[f32], src_rate: u32, dst_rate: u32) -> Option<Vec<f32>> {
if input.is_empty() || src_rate == 0 || dst_rate == 0 {
return Vec::new();
return Some(Vec::new());
}
if src_rate == dst_rate || input.len() == 1 {
return input.to_vec();
return Some(input.to_vec());
}
let out_len = ((input.len() as u64 * dst_rate as u64) / src_rate as u64)
.max(1)
.try_into()
.unwrap_or(usize::MAX);
let out_len = ((input.len() as u64 * dst_rate as u64) / src_rate as u64).max(1);
if out_len > MAX_RESAMPLED_MONO_SAMPLES_PER_FRAME as u64 {
return None;
}
let out_len = usize::try_from(out_len).ok()?;
let step = src_rate as f64 / dst_rate as f64;
let mut pos = 0.0f64;
let mut out = Vec::with_capacity(out_len);
@@ -380,7 +413,7 @@ fn resample_linear_mono(input: &[f32], src_rate: u32, dst_rate: u32) -> Vec<f32>
}
pos += step;
}
out
Some(out)
}
#[cfg(not(test))]
@@ -515,12 +548,30 @@ fn fill_output_buffer<T>(
#[cfg(test)]
mod tests {
use super::MAX_RESAMPLED_MONO_SAMPLES_PER_FRAME;
use super::is_supported_realtime_playback_sample_rate;
use super::resample_linear_mono;
use pretty_assertions::assert_eq;
#[test]
fn resample_linear_passthrough_when_same_rate() {
let input = vec![0.1, -0.2, 0.3];
assert_eq!(resample_linear_mono(&input, 24_000, 24_000), input);
assert_eq!(resample_linear_mono(&input, 24_000, 24_000), Some(input));
}
#[test]
fn playback_sample_rate_validation_rejects_unrealistic_rates() {
assert!(!is_supported_realtime_playback_sample_rate(1));
assert!(!is_supported_realtime_playback_sample_rate(7_999));
assert!(is_supported_realtime_playback_sample_rate(8_000));
assert!(is_supported_realtime_playback_sample_rate(24_000));
assert!(is_supported_realtime_playback_sample_rate(192_000));
assert!(!is_supported_realtime_playback_sample_rate(192_001));
}
#[test]
fn resample_linear_returns_none_when_output_would_exceed_cap() {
let input = vec![0.0; (MAX_RESAMPLED_MONO_SAMPLES_PER_FRAME / 2) + 1];
assert_eq!(resample_linear_mono(&input, 1, 2), None);
}
}