Add live WebRTC voice conversations to the TUI (#43581)

## What changed

- Add feature-gated `/voice`, `/voice mute`, and `/voice stop` commands with local WebRTC audio and app-server signaling.
- Show live transcripts, conversation status, and microphone and speaker levels. Preserve captions and undelivered answers across thread switches.
- Speak final answers from voice handoffs while keeping delegated reasoning and commentary hidden and typed answers unspoken.
- Retry eligible startup failures once and clean up voice sessions on thread switches and disconnects.
- Stop voice and block late handoffs after a misalignment policy violation. Remove realtime event payloads and spoken text from receipt and debug logs.

## Testing

Add coverage for voice command mapping, rejected starts, disconnect cleanup, delegated final-answer speech, late handoff rejection, and transcript and footer rendering.

GitOrigin-RevId: 11def1f86b0e023b21d0d92072d6636ccad89815
This commit is contained in:
Benjamin Carlsson
2026-09-07 20:25:30 +00:00
committed by copyberry
parent 4110342321
commit b01c3986fd
70 changed files with 4421 additions and 69 deletions
+1
View File
@@ -4677,6 +4677,7 @@ dependencies = [
"codex-otel",
"codex-plugin",
"codex-protocol",
"codex-realtime-webrtc",
"codex-rollout",
"codex-sandboxing",
"codex-shell-command",
+61 -9
View File
@@ -76,6 +76,7 @@ use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::Mutex;
use tokio::sync::Semaphore;
use tokio::task::JoinHandle;
use tokio_util::sync::CancellationToken;
use tracing::debug;
@@ -129,6 +130,37 @@ enum RealtimeFanoutTaskStop {
Detach,
}
/// Serializes turn admission with the safety cutoff for one voice session.
struct RealtimeHandoffAdmission {
gate: Semaphore,
retired: AtomicBool,
}
impl RealtimeHandoffAdmission {
fn new() -> Self {
Self {
gate: Semaphore::new(1),
retired: AtomicBool::new(false),
}
}
async fn route(&self, session: &Arc<Session>, text: String) {
let Ok(_permit) = self.gate.acquire().await else {
return;
};
if !self.retired.load(Ordering::Acquire) {
session.route_realtime_text_input(text).await;
}
}
async fn retire(&self) {
let Ok(_permit) = self.gate.acquire().await else {
return;
};
self.retired.store(true, Ordering::Release);
}
}
pub(crate) struct RealtimeConversationManager {
state: Mutex<Option<ConversationState>>,
mode_instructions: Mutex<Option<RealtimeModeInstructions>>,
@@ -467,6 +499,8 @@ struct ConversationState {
input_task: JoinHandle<()>,
fanout_task: Option<JoinHandle<()>>,
realtime_active: Arc<AtomicBool>,
// A misalignment failure retires only this session's queued handoffs.
route_handoffs: Arc<RealtimeHandoffAdmission>,
stop_token: CancellationToken,
}
@@ -489,6 +523,7 @@ struct RealtimeStart {
struct RealtimeStartOutput {
realtime_active: Arc<AtomicBool>,
route_handoffs: Arc<RealtimeHandoffAdmission>,
events_rx: Receiver<RealtimeEvent>,
transcript_tail_rx: Receiver<String>,
sdp: Option<String>,
@@ -529,6 +564,20 @@ impl RealtimeConversationManager {
)
}
/// Prevent already parsed handoffs from starting turns after a safety failure.
/// Keep the gate on the old session so a later voice session starts fresh.
pub(crate) async fn retire_handoffs_for_misalignment(&self) {
let route_handoffs = self
.state
.lock()
.await
.as_ref()
.map(|state| Arc::clone(&state.route_handoffs));
if let Some(route_handoffs) = route_handoffs {
route_handoffs.retire().await;
}
}
async fn start(
&self,
start: RealtimeStart,
@@ -581,6 +630,7 @@ impl RealtimeConversationManager {
let (transcript_tail_tx, transcript_tail_rx) = async_channel::bounded::<String>(1);
let realtime_active = Arc::new(AtomicBool::new(true));
let route_handoffs = Arc::new(RealtimeHandoffAdmission::new());
let stop_token = CancellationToken::new();
let handoff = RealtimeHandoffState {
output_tx: handoff_output_tx,
@@ -690,10 +740,12 @@ impl RealtimeConversationManager {
input_task: task,
fanout_task: None,
realtime_active: Arc::clone(&realtime_active),
route_handoffs: Arc::clone(&route_handoffs),
stop_token,
});
Ok(RealtimeStartOutput {
realtime_active,
route_handoffs,
events_rx,
transcript_tail_rx,
sdp,
@@ -1571,6 +1623,7 @@ async fn handle_start_inner(
let RealtimeStartOutput {
realtime_active,
route_handoffs,
events_rx,
transcript_tail_rx,
sdp,
@@ -1597,10 +1650,9 @@ async fn handle_start_inner(
match &event {
RealtimeEvent::AudioOut(_) => {}
_ => {
info!(
event = ?event,
"received realtime conversation event"
);
// Transcripts, handoffs, and conversation items may contain
// credentials. Keep this receipt free of event payloads.
info!("received realtime conversation event");
}
}
if let RealtimeEvent::Error(_) = &event {
@@ -1613,9 +1665,9 @@ async fn handle_start_inner(
_ => None,
};
if let Some(text) = maybe_routed_text {
debug!(text = %text, "[realtime-text] realtime conversation text output");
let sess_for_routed_text = Arc::clone(&sess_clone);
sess_for_routed_text.route_realtime_text_input(text).await;
// The routed text can contain spoken prompts or workspace secrets.
debug!("[realtime-text] realtime conversation text output");
route_handoffs.route(&sess_clone, text).await;
}
sess_clone
.send_event_raw(ev(EventMsg::RealtimeConversationRealtime(
@@ -1626,7 +1678,7 @@ async fn handle_start_inner(
.await;
}
if let Ok(text) = transcript_tail_rx.recv().await {
sess_clone.route_realtime_text_input(text).await;
route_handoffs.route(&sess_clone, text).await;
}
if fanout_realtime_active.swap(false, Ordering::Relaxed) {
match end {
@@ -1785,7 +1837,7 @@ pub(crate) async fn handle_speech(
sub_id: String,
params: ConversationSpeechParams,
) {
debug!(text = %params.text, "[realtime-text] appending realtime speech");
debug!("[realtime-text] appending realtime speech");
if let Err(err) = sess.conversation.append_speech(params.text).await {
error!("failed to append realtime speech: {err}");
if sess.conversation.running_state().await.is_some() {
+6
View File
@@ -636,6 +636,12 @@ pub(crate) async fn run_turn(
}
Err(e) => {
info!("Turn error: {e:#}");
if matches!(
e.details(),
CodexErrorDetails::MisalignmentPolicyViolation { .. }
) {
sess.conversation.retire_handoffs_for_misalignment().await;
}
let error = e.to_codex_protocol_error();
sess.emit_turn_error_lifecycle(turn_context.as_ref(), error.clone())
.await;
+1
View File
@@ -128,6 +128,7 @@ mod prompt_debug_tests;
mod quota_exceeded;
mod realtime_conversation;
mod realtime_initial_items;
mod realtime_misalignment;
mod realtime_sideband_endpoint;
mod reasoning_effort_override;
mod remote_env;
@@ -0,0 +1,151 @@
//! A failed turn must not admit handoffs still arriving from its voice session.
use anyhow::Context;
use anyhow::Result;
use codex_config::config_toml::RealtimeWsVersion;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeOutputModality;
use core_test_support::responses;
use core_test_support::skip_if_no_network;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event_match;
use futures::SinkExt;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::time::Duration;
use tokio::net::TcpListener;
use tokio::sync::oneshot;
use tokio_tungstenite::accept_async;
use tokio_tungstenite::tungstenite::Message;
use wiremock::ResponseTemplate;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn misalignment_retires_late_voice_handoff_before_it_starts_a_turn() -> Result<()> {
skip_if_no_network!(Ok(()));
let api_server = responses::start_mock_server().await;
let first_response = responses::mount_response_once(
&api_server,
ResponseTemplate::new(403).set_body_json(json!({
"error": {
"message": "This request violated the misalignment policy.",
"type": "invalid_request_error",
"code": "misalignment_policy_violation"
}
})),
)
.await;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let realtime_url = format!("ws://{}", listener.local_addr()?);
let (late_handoff_tx, mut late_handoff_rx) = oneshot::channel();
let (finish_tx, finish_rx) = oneshot::channel();
let sideband = tokio::spawn(async move {
let (stream, _) = listener.accept().await?;
let mut websocket = accept_async(stream).await?;
websocket
.send(Message::Text(
json!({
"type": "session.updated",
"session": {"id": "session_misalignment", "instructions": "backend prompt"}
})
.to_string()
.into(),
))
.await?;
for (handoff_id, text) in [
("initial_handoff", "first request"),
("late_handoff", "must not start a turn"),
] {
if handoff_id == "late_handoff" {
(&mut late_handoff_rx).await?;
}
websocket
.send(Message::Text(
json!({
"type": "conversation.handoff.requested",
"handoff_id": handoff_id,
"item_id": handoff_id,
"input_transcript": text
})
.to_string()
.into(),
))
.await?;
}
finish_rx.await?;
websocket.close(None).await?;
Ok::<_, anyhow::Error>(())
});
let mut builder = test_codex().with_config(move |config| {
config.experimental_realtime_ws_base_url = Some(realtime_url);
config.realtime.version = RealtimeWsVersion::V1;
});
let test = builder.build_with_auto_env(&api_server).await?;
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
client_managed_handoffs: false,
delegation_ack_filler: None,
flush_transcript_tail_on_session_end: false,
codex_responses_as_items: false,
codex_response_item_prefix: None,
codex_response_handoff_mode:
codex_protocol::protocol::CodexResponseHandoffMode::Thinking,
codex_response_handoff_channel_prefixes: None,
model: None,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: true,
initial_items: Vec::new(),
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: Some(Some("backend prompt".to_string())),
realtime_session_id: None,
transport: None,
version: None,
voice: None,
}))
.await?;
wait_for_event_match(&test.codex, |event| match event {
EventMsg::Error(error)
if error.codex_error_info == Some(CodexErrorInfo::MisalignmentPolicyViolation) =>
{
Some(())
}
_ => None,
})
.await;
assert_eq!(first_response.requests().len(), 1);
late_handoff_tx.send(()).expect("sideband still open");
wait_for_event_match(&test.codex, |event| match event {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::HandoffRequested(handoff),
}) if handoff.handoff_id == "late_handoff" => Some(()),
_ => None,
})
.await;
finish_tx.send(()).expect("sideband still open");
sideband.await??;
// The fanout forwards a handoff only after deciding whether to route it.
// Give any spawned turn a chance to reach the mock API before checking.
tokio::time::sleep(Duration::from_millis(200)).await;
let requests = api_server
.received_requests()
.await
.context("API requests")?;
assert_eq!(
requests
.iter()
.filter(|request| request.url.path() == "/v1/responses")
.count(),
1,
);
Ok(())
}
+22
View File
@@ -19,6 +19,22 @@ use crate::message_reader::MessageReader;
const DEADLINE: Duration = Duration::from_secs(/*secs*/ 5);
const RUNTIME_INITIALIZATION_DEADLINE: Duration = Duration::from_secs(/*secs*/ 30);
/// Startup failures eligible for recovery remain distinct from all other failures.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ConnectionError {
NegotiationTimedOut,
Failed,
}
impl std::fmt::Display for ConnectionError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.write_str(match self {
Self::NegotiationTimedOut => "voice negotiation timed out",
Self::Failed => "voice connection failed",
})
}
}
impl std::error::Error for ConnectionError {}
/// Owns one helper. Dropping it terminates the process and leaves its waiter to reap it.
/// A successful handshake establishes compatibility only, not an active audio session.
pub struct VoiceHost {
@@ -94,6 +110,12 @@ impl VoiceHost {
Duration::from_secs(/*secs*/ 20),
)
.await?;
if response == (Message::TransportTimedOut {}) {
self.process.terminate();
// A lost exit notification or failed cleanup is not retryable.
timeout(DEADLINE, &mut self.exit).await??;
return Err(ConnectionError::NegotiationTimedOut.into());
}
ensure!(
response == Message::TransportReady {},
"unexpected voice helper response"
@@ -61,3 +61,52 @@ async fn cancelling_initialization_terminates_the_owned_helper() -> anyhow::Resu
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 1), spawned.exit_rx).await??;
Ok(())
}
#[cfg(unix)]
#[tokio::test]
async fn negotiation_timeout_waits_for_helper_exit_before_allowing_retry() -> anyhow::Result<()> {
let spawned = codex_utils_pty::spawn_pipe_process(
std::path::Path::new("/bin/sleep"),
&["30".to_owned()],
std::path::Path::new("/"),
&child_environment(std::iter::empty()),
/*arg0*/ &None,
&[],
)
.await?;
let (output, receiver) = tokio::sync::mpsc::channel(/*buffer*/ 1);
output
.send(crate::encode_frame(&crate::Message::TransportTimedOut {})?)
.await?;
let (reaped, exit) = tokio::sync::oneshot::channel();
let host = super::VoiceHost {
process: spawned.session,
output: crate::message_reader::MessageReader::new(receiver),
exit,
};
let answer = crate::SessionDescription::try_from("synthetic-answer".to_owned()).unwrap();
let mut connection = Box::pin(host.apply_answer(answer));
std::future::poll_fn(|cx| {
assert!(std::future::Future::poll(connection.as_mut(), cx).is_pending());
std::task::Poll::Ready(())
})
.await;
let code =
tokio::time::timeout(std::time::Duration::from_secs(/*secs*/ 1), spawned.exit_rx).await??;
// The process has exited, but the explicit reaping acknowledgement is still required.
std::future::poll_fn(|cx| {
assert!(std::future::Future::poll(connection.as_mut(), cx).is_pending());
std::task::Poll::Ready(())
})
.await;
reaped.send(code).unwrap();
let error = connection
.await
.err()
.expect("timeout must not return a connected helper");
assert_eq!(
error.downcast_ref::<crate::ConnectionError>(),
Some(&crate::ConnectionError::NegotiationTimedOut)
);
Ok(())
}
+1
View File
@@ -3,6 +3,7 @@ mod message_reader;
mod protocol;
mod session;
pub use client::ConnectionError;
pub use client::VoiceHost;
pub use protocol::AudioControls;
pub use protocol::AudioState;
+1
View File
@@ -81,6 +81,7 @@ pub enum Message {
Offer { sdp: SessionDescription },
ApplyAnswer { sdp: SessionDescription },
TransportReady {},
TransportTimedOut {},
OpenDevices {},
DevicesOpened {},
SetAudioControls { controls: AudioControls },
+30 -10
View File
@@ -26,7 +26,10 @@ const ANSWER_WAIT: Duration = Duration::from_secs(/*secs*/ 40);
static RUNTIME: Mutex<Option<Arc<tokio::runtime::Runtime>>> = Mutex::new(None);
enum Command {
Answer(SessionDescription, blocking::SyncSender<()>),
Answer(
SessionDescription,
blocking::SyncSender<std::result::Result<(), crate::ConnectionError>>,
),
Controls(AudioControls),
}
@@ -196,15 +199,22 @@ impl RealtimeWebrtcSessionHandle {
}
/// Called off the UI thread; return only after negotiation, device startup and restored controls.
pub fn apply_answer_sdp(&self, answer: String) -> Result<()> {
let sdp = SessionDescription::try_from(answer).map_err(anyhow::Error::msg)?;
pub fn apply_answer_sdp(
&self,
answer: String,
) -> std::result::Result<(), crate::ConnectionError> {
let sdp =
SessionDescription::try_from(answer).map_err(|_| crate::ConnectionError::Failed)?;
let (complete, result) = blocking::sync_channel(/*bound*/ 1);
self.send(Command::Answer(sdp, complete))?;
if result.recv_timeout(ANSWER_WAIT).is_err() {
self.close();
anyhow::bail!("voice connection failed");
self.send(Command::Answer(sdp, complete))
.map_err(|_| crate::ConnectionError::Failed)?;
match result.recv_timeout(ANSWER_WAIT) {
Ok(result) => result,
Err(_) => {
self.close();
Err(crate::ConnectionError::Failed)
}
}
Ok(())
}
pub fn take_error(&self) -> Option<String> {
@@ -254,13 +264,23 @@ async fn run(
biased;
command = commands.recv() => match command {
Some(Command::Answer(sdp, complete)) if !connected => {
host = host.apply_answer(sdp).await?.open_devices().await?;
host = match host.apply_answer(sdp).await {
Ok(host) => host,
Err(error) => {
let failure = error.downcast_ref::<crate::ConnectionError>()
.copied().unwrap_or(crate::ConnectionError::Failed);
let _ = complete.send(Err(failure));
// This failure is delivered by the startup completion only.
return Ok(());
}
};
host = host.open_devices().await?;
let applied = startup_controls(&mut commands, controls, |initial| {
host.begin_audio_controls(initial)
})?;
applied.await?;
connected = true;
let _ = complete.send(());
let _ = complete.send(Ok(()));
}
Some(Command::Controls(next)) => {
if connected {
@@ -180,3 +180,21 @@ fn startup_dispatch_is_ordered_but_acknowledgement_does_not_block_setters() {
);
drop(applied);
}
#[tokio::test]
async fn startup_completion_preserves_classified_failure() {
for failure in [
crate::ConnectionError::NegotiationTimedOut,
crate::ConnectionError::Failed,
] {
let (handle, mut commands, _) = handles();
let caller = handle.clone();
let thread = std::thread::spawn(move || caller.apply_answer_sdp("synthetic-answer".into()));
let Some(Command::Answer(_, complete)) = commands.recv().await else {
panic!("answer expected")
};
complete.send(Err(failure)).unwrap();
assert_eq!(thread.join().unwrap(), Err(failure));
assert_eq!(handle.take_error(), None);
}
}
+2
View File
@@ -53,6 +53,8 @@ codex-models-manager = { workspace = true }
codex-otel = { workspace = true }
codex-plugin = { workspace = true }
codex-protocol = { workspace = true }
codex-realtime-webrtc = { workspace = true }
futures = { workspace = true }
codex-rollout = { workspace = true }
codex-sandboxing = { workspace = true }
codex-shell-command = { workspace = true }
+5
View File
@@ -235,6 +235,7 @@ mod pets;
mod platform_actions;
mod plugin_mentions;
mod rate_limit_refresh;
mod realtime_delivery;
mod recap;
mod reconnect;
mod replay_filter;
@@ -617,6 +618,10 @@ pub(crate) struct App {
windows_sandbox: WindowsSandboxState,
thread_event_channels: HashMap<ThreadId, ThreadEventChannel>,
pending_realtime_speech_replay: HashMap<ThreadId, Vec<(String, ThreadItem)>>,
pending_realtime_transcript_replay:
HashMap<ThreadId, VecDeque<crate::chatwidget::RealtimeTranscriptRecord>>,
realtime_replay_order: VecDeque<ThreadId>,
temporary_structured_requests: HashMap<ThreadId, mpsc::UnboundedSender<ServerNotification>>,
/// Track title generation across thread switches and deduplicate automatic requests.
pending_thread_titles: HashSet<(ThreadId, ThreadTitleDestination)>,
+78 -3
View File
@@ -21,7 +21,7 @@ use codex_app_server_protocol::ThreadGoalStatus;
#[cfg(target_os = "windows")]
use codex_config::types::WindowsSandboxModeToml;
const SHUTDOWN_FIRST_EXIT_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 2);
pub(super) const SHUTDOWN_FIRST_EXIT_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 2);
impl App {
pub(super) async fn handle_event(
@@ -785,6 +785,23 @@ impl App {
.apply_reserve_fallback_to_pending_turn(&mut op);
}
let is_user_turn = matches!(&op, AppCommand::UserTurn { .. });
let is_realtime_stop = matches!(&op, AppCommand::RealtimeConversationStop { .. });
let realtime_stop_thread_id = match &op {
AppCommand::RealtimeConversationStop { thread_id } => Some(*thread_id),
_ => None,
};
let realtime_speech_delivery_id = match &op {
AppCommand::RealtimeConversationSpeech { delivery_id, .. } => {
Some(*delivery_id)
}
_ => None,
};
let is_realtime_conversation = matches!(
&op,
AppCommand::RealtimeConversationStart { .. }
| AppCommand::RealtimeConversationStop { .. }
| AppCommand::RealtimeConversationSpeech { .. }
);
if is_user_turn {
let screen_size = tui.terminal.last_known_screen_size;
self.handle_draw_pre_render(tui, screen_size)?;
@@ -797,6 +814,10 @@ impl App {
}
self.chat_widget.prepare_local_op_submission(&op);
if let Err(err) = self.submit_active_thread_op(app_server, op).await {
if let Some(delivery_id) = realtime_speech_delivery_id {
self.chat_widget
.restore_undelivered_realtime_speech(delivery_id);
}
if self.recover_transport_error(&err)
{
return Ok(AppRunControl::Continue);
@@ -817,10 +838,22 @@ impl App {
&& self
.chat_widget
.handle_turn_start_rejection(format!("Failed to start turn: {err:#}"));
if !handled {
if is_realtime_conversation {
let message = format!("Voice conversation failed: {err:#}");
if is_realtime_stop {
if self.chat_widget.thread_id() == realtime_stop_thread_id {
self.chat_widget.reset_realtime_conversation();
self.chat_widget.add_error_message(message);
}
} else {
self.chat_widget.on_realtime_error(message);
}
tracing::error!(error = ?err, "realtime conversation request failed");
} else if handled {
tracing::error!(error = ?err, "failed to start turn through app server");
} else {
return Err(err);
}
tracing::error!(error = ?err, "failed to start turn through app server");
}
}
AppEvent::ConfirmSafetyBufferedRetry {
@@ -1656,6 +1689,47 @@ impl App {
self.sync_active_thread_personality_setting(app_server, personality)
.await;
}
AppEvent::RealtimeWebrtcOfferCreated {
thread_id,
attempt_id,
result,
} => {
if self.chat_widget.thread_id() == Some(thread_id) {
self.chat_widget
.on_realtime_webrtc_offer_created(thread_id, attempt_id, result);
} else if let Ok(offer) = result {
offer.handle.close();
}
}
AppEvent::RealtimeWebrtcConnected {
thread_id,
attempt_id,
result,
} => {
if self.chat_widget.thread_id() == Some(thread_id) {
self.chat_widget
.on_realtime_webrtc_connected(attempt_id, result);
}
}
AppEvent::StopRealtimeConversation { thread_id } => {
match tokio::time::timeout(
SHUTDOWN_FIRST_EXIT_TIMEOUT,
app_server.thread_realtime_stop(thread_id),
)
.await
{
Ok(Ok(())) => {}
Ok(Err(error)) => tracing::warn!(
%thread_id,
%error,
"failed to stop voice conversation after switching threads"
),
Err(_) => tracing::warn!(
%thread_id,
"timed out stopping voice conversation after switching threads"
),
}
}
AppEvent::SettingsSelectionClosed => {
self.app_event_tx.send(AppEvent::SettingsSelectionSettled);
}
@@ -3369,6 +3443,7 @@ impl App {
})
}
ExitMode::Immediate => {
self.stop_realtime_conversation(app_server).await;
self.pending_shutdown_exit_thread_id = None;
AppRunControl::Exit(ExitReason::UserRequested)
}
@@ -4,6 +4,7 @@
use super::*;
use crate::app_server_session::turn_permissions_overrides;
use crate::chatwidget::MisalignmentReview;
use crate::chatwidget::MisalignmentTurnSource;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
@@ -161,6 +162,10 @@ impl App {
Ok(response) => {
let store = &self.ensure_thread_channel(review.thread_id).store;
store.lock().await.active_turn_id = Some(response.turn.id.clone());
self.chat_widget.clear_misalignment_for_new_turn(
&response.turn.id,
MisalignmentTurnSource::AcknowledgedContinuation,
);
self.chat_widget.handle_server_notification(
ServerNotification::TurnStarted(
codex_app_server_protocol::TurnStartedNotification {
+392
View File
@@ -0,0 +1,392 @@
//! Preserve voice answers and accepted transcripts across thread-widget replacement.
//!
//! A queued speech command can outlive its widget. Retain its bounded final
//! item by original thread until replay either renders that item or needs a
//! fallback, without adding history to whichever thread is currently visible.
use super::*;
use codex_protocol::models::MessagePhase;
use std::collections::HashMap;
use std::collections::HashSet;
type ReplayedVoiceTextCounts = HashMap<(String, String), usize>;
const MAX_INACTIVE_REALTIME_REPLAY_THREADS: usize = 16;
fn normalized_replay_text(text: &str) -> String {
text.split_whitespace().collect::<Vec<_>>().join(" ")
}
fn record_voice_item(
item: &ThreadItem,
turn_id: &str,
delegated: &mut HashSet<String>,
items: &mut HashMap<(String, String, &'static str), String>,
) {
match item {
ThreadItem::UserMessage { id, content, .. } => {
if let Some(input) = crate::chatwidget::realtime_delegation_input(content) {
delegated.insert(turn_id.to_string());
let display = crate::chatwidget::realtime_delegation_display_text(input);
items.insert(
(turn_id.to_string(), id.clone(), "user"),
normalized_replay_text(&display),
);
}
}
ThreadItem::AgentMessage {
id,
text,
phase: Some(MessagePhase::FinalAnswer) | None,
..
} if delegated.contains(turn_id)
&& !crate::chatwidget::is_private_realtime_agent_item(item)
&& !text.trim().is_empty() =>
{
items.insert(
(turn_id.to_string(), id.clone(), "assistant"),
normalized_replay_text(text),
);
}
_ => {}
}
}
pub(super) fn replayed_voice_texts_from_turns(turns: &[Turn]) -> ReplayedVoiceTextCounts {
let mut delegated = HashSet::new();
let mut items = HashMap::new();
for turn in turns {
for item in &turn.items {
record_voice_item(item, &turn.id, &mut delegated, &mut items);
}
}
count_replayed_voice_texts(items)
}
fn count_replayed_voice_texts(
items: HashMap<(String, String, &'static str), String>,
) -> ReplayedVoiceTextCounts {
let mut counts = HashMap::new();
for ((_, _, role), text) in items {
if !text.is_empty() {
*counts.entry((role.to_string(), text)).or_insert(0) += 1;
}
}
counts
}
pub(super) fn replayed_voice_texts(snapshot: &ThreadEventSnapshot) -> ReplayedVoiceTextCounts {
let mut delegated = snapshot
.delegated_turns
.iter()
.cloned()
.collect::<HashSet<_>>();
let mut items = HashMap::new();
for turn in &snapshot.turns {
for item in &turn.items {
record_voice_item(item, &turn.id, &mut delegated, &mut items);
}
}
for event in &snapshot.events {
let ThreadBufferedEvent::Notification(notification) = event else {
continue;
};
match notification.as_ref() {
ServerNotification::ItemStarted(n) => {
if let ThreadItem::UserMessage { content, .. } = &n.item
&& crate::chatwidget::realtime_delegation_input(content).is_some()
{
delegated.insert(n.turn_id.clone());
}
}
ServerNotification::ItemCompleted(n) => {
record_voice_item(&n.item, &n.turn_id, &mut delegated, &mut items);
}
ServerNotification::TurnCompleted(n) if n.turn.status == TurnStatus::Completed => {
if let Some(item) = n.turn.items.iter().rev().find(|item| {
matches!(
item,
ThreadItem::AgentMessage {
phase: Some(MessagePhase::FinalAnswer) | None,
..
}
) && !crate::chatwidget::is_private_realtime_agent_item(item)
}) {
record_voice_item(item, &n.turn.id, &mut delegated, &mut items);
}
}
_ => {}
}
}
count_replayed_voice_texts(items)
}
impl App {
pub(super) fn note_realtime_replay_thread(&mut self, thread_id: ThreadId) {
if self.realtime_replay_order.contains(&thread_id) {
return;
}
if self.realtime_replay_order.len() >= MAX_INACTIVE_REALTIME_REPLAY_THREADS
&& let Some(oldest) = self.realtime_replay_order.pop_front()
{
self.pending_realtime_transcript_replay.remove(&oldest);
self.pending_realtime_speech_replay.remove(&oldest);
}
self.realtime_replay_order.push_back(thread_id);
}
pub(super) fn forget_realtime_replay_thread(&mut self, thread_id: ThreadId) {
self.realtime_replay_order
.retain(|saved| *saved != thread_id);
self.pending_realtime_transcript_replay.remove(&thread_id);
self.pending_realtime_speech_replay.remove(&thread_id);
}
pub(super) fn retain_inactive_realtime_transcript(
&mut self,
thread_id: ThreadId,
notification: &ServerNotification,
) {
let (role, text, complete) = match notification {
ServerNotification::ThreadRealtimeTranscriptDelta(n) => {
(n.role.as_str(), n.delta.as_str(), false)
}
ServerNotification::ThreadRealtimeTranscriptDone(n) => {
(n.role.as_str(), n.text.as_str(), true)
}
_ => return,
};
if !matches!(role, "user" | "assistant") {
return;
}
self.note_realtime_replay_thread(thread_id);
let records = self
.pending_realtime_transcript_replay
.entry(thread_id)
.or_default();
if complete {
if text.trim().is_empty() {
if let Some(index) = records
.iter()
.rposition(|record| record.role == role && !record.complete)
{
records.remove(index);
}
return;
}
let mut bounded = text.to_string();
if bounded.len() > crate::chatwidget::MAX_TRANSCRIPT_BYTES {
let mut end = crate::chatwidget::MAX_TRANSCRIPT_BYTES;
while !bounded.is_char_boundary(end) {
end -= 1;
}
bounded.truncate(end);
}
if let Some(partial) = records
.iter_mut()
.rev()
.find(|record| record.role == role && !record.complete)
{
partial.text = bounded;
partial.complete = true;
return;
}
if records.len() >= crate::chatwidget::MAX_REPLAY_TRANSCRIPT_CELLS {
records.pop_front();
}
records.push_back(crate::chatwidget::RealtimeTranscriptRecord {
role: role.to_string(),
text: bounded,
complete: true,
});
} else {
if text.is_empty() {
return;
}
let partial = records
.iter_mut()
.rev()
.find(|record| record.role == role && !record.complete);
if let Some(partial) = partial {
partial.text.push_str(text);
if partial.text.len() > crate::chatwidget::MAX_TRANSCRIPT_BYTES {
let mut start = partial.text.len() - crate::chatwidget::MAX_TRANSCRIPT_BYTES;
while !partial.text.is_char_boundary(start) {
start += 1;
}
partial.text.drain(..start);
}
} else {
let mut bounded = text.to_string();
if bounded.len() > crate::chatwidget::MAX_TRANSCRIPT_BYTES {
let mut start = bounded.len() - crate::chatwidget::MAX_TRANSCRIPT_BYTES;
while !bounded.is_char_boundary(start) {
start += 1;
}
bounded.drain(..start);
}
if records.len() >= crate::chatwidget::MAX_REPLAY_TRANSCRIPT_CELLS {
records.pop_front();
}
records.push_back(crate::chatwidget::RealtimeTranscriptRecord {
role: role.to_string(),
text: bounded,
complete: false,
});
}
}
}
pub(super) fn retain_realtime_replay_state_before_replace(&mut self) {
if let Some(thread_id) = self.chat_widget.thread_id() {
if self.chat_widget.may_receive_realtime_transcripts() {
self.note_realtime_replay_thread(thread_id);
// Even a session without captions can receive a final queued caption
// after this widget has been replaced.
self.pending_realtime_transcript_replay
.entry(thread_id)
.or_default();
}
let cells = self.chat_widget.take_realtime_transcript_cells_for_replay();
if !cells.is_empty() {
self.note_realtime_replay_thread(thread_id);
let pending = self
.pending_realtime_transcript_replay
.entry(thread_id)
.or_default();
// Only the active widget produces these cells; reattachment consumes them
// before another widget for this thread can become active.
// Up to 32 accepted cells plus one partial for each speaker.
debug_assert!(pending.len() + cells.len() <= 34);
pending.extend(cells);
}
}
for (thread_id, turn_id, item) in self
.chat_widget
.take_undelivered_realtime_speech_for_replay()
{
self.note_realtime_replay_thread(thread_id);
let pending = self
.pending_realtime_speech_replay
.entry(thread_id)
.or_default();
// The active widget is the only producer for a thread; selecting
// that thread reconciles these records before it can produce more.
debug_assert!(pending.len() < 16);
pending.push((turn_id, item));
}
}
pub(super) fn restore_realtime_replay_state_after_replay(
&mut self,
replayed_final_items: &HashMap<(String, String), String>,
mut replayed_voice_texts: ReplayedVoiceTextCounts,
) {
let Some(thread_id) = self.chat_widget.thread_id() else {
return;
};
self.realtime_replay_order
.retain(|saved| *saved != thread_id);
let mut retained_assistant_captions = HashMap::<String, usize>::new();
if let Some(mut cells) = self.pending_realtime_transcript_replay.remove(&thread_id) {
cells.retain(|record| {
if !record.complete {
return true;
}
let key = (record.role.clone(), normalized_replay_text(&record.text));
if let Some(count) = replayed_voice_texts.get_mut(&key)
&& *count > 0
{
*count -= 1;
return false;
}
true
});
for record in &cells {
if record.complete && record.role == "assistant" {
*retained_assistant_captions
.entry(normalized_replay_text(&record.text))
.or_default() += 1;
}
}
self.chat_widget.restore_realtime_transcript_cells(cells);
}
let Some(pending) = self.pending_realtime_speech_replay.remove(&thread_id) else {
return;
};
for (turn_id, item) in pending {
let ThreadItem::AgentMessage { id, text, .. } = &item else {
continue;
};
// A matching ID with empty text did not render the saved answer.
// Any nonempty snapshot text is authoritative even if it changed.
if replayed_final_items
.get(&(turn_id.clone(), id.clone()))
.is_none_or(|text| text.trim().is_empty())
{
// A completed caption restored above already displays this
// answer. Consume only one exact match; partial or distinct
// captions must not hide an undelivered fallback.
let spoken_text = text.trim().strip_prefix("[FINAL]").unwrap_or(text);
if let Some(count) =
retained_assistant_captions.get_mut(&normalized_replay_text(spoken_text))
&& *count > 0
{
*count -= 1;
continue;
}
self.chat_widget
.render_undelivered_realtime_speech(thread_id, turn_id, item);
}
}
}
}
pub(super) fn completed_agent_items(
snapshot: &ThreadEventSnapshot,
) -> HashMap<(String, String), String> {
let mut completed = completed_agent_items_from_turns(&snapshot.turns);
for event in &snapshot.events {
let ThreadBufferedEvent::Notification(notification) = event else {
continue;
};
match notification.as_ref() {
ServerNotification::ItemCompleted(completed_item) => {
if let ThreadItem::AgentMessage { id, text, .. } = &completed_item.item
&& !text.trim().is_empty()
{
completed.insert((completed_item.turn_id.clone(), id.clone()), text.clone());
}
}
ServerNotification::TurnCompleted(completed_turn)
if completed_turn.turn.status == TurnStatus::Completed =>
{
for item in &completed_turn.turn.items {
if let ThreadItem::AgentMessage { id, text, .. } = item
&& !text.trim().is_empty()
{
completed
.insert((completed_turn.turn.id.clone(), id.clone()), text.clone());
}
}
}
_ => {}
}
}
completed
}
pub(super) fn completed_agent_items_from_turns(
turns: &[Turn],
) -> HashMap<(String, String), String> {
let mut completed = HashMap::new();
for turn in turns {
for item in &turn.items {
if let ThreadItem::AgentMessage { id, text, .. } = item
&& !text.trim().is_empty()
{
completed.insert((turn.id.clone(), id.clone()), text.clone());
}
}
}
completed
}
+18
View File
@@ -484,6 +484,7 @@ impl App {
/// This helper copies every known nickname/role from `AgentNavigationState` into the
/// replacement widget so that replayed collab items render agent names immediately.
pub(super) fn replace_chat_widget(&mut self, mut chat_widget: ChatWidget) {
self.retain_realtime_replay_state_before_replace();
chat_widget.cyber_policy_notice = self.chat_widget.cyber_policy_notice.clone();
self.commit_animation = None;
// Transfer the last-written terminal title to the replacement widget
@@ -581,6 +582,16 @@ impl App {
return Ok(());
}
let previous_thread_id = self.active_thread_id;
// Notifications already routed to the active channel must reach the old
// widget before its transcript state is transferred. Events routed after
// deactivation are retained separately for that thread.
if let Some(mut receiver) = self.active_thread_rx.take() {
while let Ok(event) = receiver.try_recv() {
self.handle_thread_event_now_recovering_file_changes(event)
.await;
}
self.active_thread_rx = Some(receiver);
}
self.store_active_thread_receiver().await;
self.active_thread_id = None;
let Some((receiver, mut snapshot)) = self.activate_thread_for_replay(thread_id).await
@@ -656,6 +667,10 @@ impl App {
self.recap.seed_from_progress(recap_progress, now);
self.schedule_recap_check(thread_id, now);
// The old widget owns the local helper and its undelivered answers.
// Transfer replay state before stopping its backend voice session.
self.retain_realtime_replay_state_before_replace();
self.stop_realtime_conversation(app_server).await;
self.render_thread_snapshot(tui, app_server, thread_id, snapshot, !is_replay_only)?;
if is_replay_only
&& self
@@ -749,6 +764,9 @@ impl App {
pub(super) fn reset_thread_event_state(&mut self) {
self.abort_all_thread_event_listeners();
self.thread_event_channels.clear();
self.pending_realtime_speech_replay.clear();
self.pending_realtime_transcript_replay.clear();
self.realtime_replay_order.clear();
self.pending_server_profiles.clear();
self.agents_overview.activity.clear();
self.agent_navigation.clear();
+1
View File
@@ -524,6 +524,7 @@ impl App {
} else {
self.refresh_pending_thread_approvals().await;
}
self.forget_realtime_replay_thread(thread_id);
self.sync_active_agent_label();
}
+3
View File
@@ -707,6 +707,9 @@ See the Codex keymap documentation for supported actions and examples."
pending_shutdown_exit_thread_id: None,
windows_sandbox: WindowsSandboxState::default(),
thread_event_channels: HashMap::new(),
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashSet::new(),
thread_event_listener_tasks: HashMap::new(),
+3
View File
@@ -64,6 +64,9 @@ pub(super) async fn make_test_app() -> App {
pending_shutdown_exit_thread_id: None,
windows_sandbox: WindowsSandboxState::default(),
thread_event_channels: HashMap::new(),
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashSet::new(),
thread_event_listener_tasks: HashMap::new(),
+27
View File
@@ -42,6 +42,10 @@ mod patch_approval_tests;
mod permission_shortcuts_tests;
mod plugin_catalog;
mod rate_limits;
#[path = "tests/realtime_handoff_e2e.rs"]
mod realtime_handoff_e2e;
#[path = "tests/realtime_start.rs"]
mod realtime_start;
#[path = "tests/recap_generation_tests.rs"]
mod recap_generation;
mod safety_buffering;
@@ -1334,6 +1338,7 @@ async fn replayed_turn_complete_submits_restored_queued_follow_up() {
while new_op_rx.try_recv().is_ok() {}
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(Box::new(
@@ -1387,6 +1392,7 @@ async fn replay_only_thread_keeps_restored_queue_visible() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(Box::new(
@@ -1438,6 +1444,7 @@ async fn replay_thread_snapshot_keeps_queue_when_running_state_only_comes_from_s
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![],
@@ -1487,6 +1494,7 @@ async fn replay_thread_snapshot_in_progress_turn_restores_running_queue_state()
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: vec![test_turn("turn-1", TurnStatus::InProgress, Vec::new())],
events: Vec::new(),
@@ -1516,6 +1524,7 @@ async fn replay_thread_snapshot_in_progress_turn_restores_running_state_without_
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: vec![test_turn("turn-1", TurnStatus::InProgress, Vec::new())],
events: Vec::new(),
@@ -1558,6 +1567,7 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up()
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![
@@ -1699,6 +1709,7 @@ async fn replay_thread_snapshot_restores_collaboration_mode_for_draft_submit() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![],
@@ -1779,6 +1790,7 @@ async fn replay_thread_snapshot_restores_collaboration_mode_without_input() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![],
@@ -1829,6 +1841,7 @@ async fn replayed_interrupted_turn_restores_queued_input_to_composer() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(Box::new(
@@ -3714,6 +3727,7 @@ async fn replay_snapshot_with_pending_request_suppresses_replay_notices() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(test_thread_session(thread_id, test_path_buf("/tmp/main"))),
turns: Vec::new(),
events: vec![
@@ -4096,6 +4110,7 @@ async fn replayed_file_change_approval_recovers_snapshot_changes() {
let cwd = test_path_buf("/tmp/project").abs();
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(test_thread_session(thread_id, cwd.clone().into_path_buf())),
turns: vec![test_turn(
"turn-replayed-approval",
@@ -5053,6 +5068,7 @@ async fn side_thread_snapshot_does_not_refresh_from_fork_history() {
.insert(side_thread_id, SideThreadState::new(parent_thread_id));
let snapshot = ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(ThreadSessionState {
rollout_path: None,
..test_thread_session(side_thread_id, test_path_buf("/tmp/side"))
@@ -5081,6 +5097,7 @@ async fn side_thread_snapshot_skips_session_header_preamble() {
.insert(side_thread_id, SideThreadState::new(parent_thread_id));
let snapshot = ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(ThreadSessionState {
forked_from_id: Some(parent_thread_id),
fork_parent_title: None,
@@ -5256,6 +5273,7 @@ async fn active_side_thread_renders_live_mcp_startup_notifications() {
app.activate_thread_channel(side_thread_id).await;
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(test_thread_session(
side_thread_id,
test_path_buf("/tmp/side"),
@@ -5839,6 +5857,9 @@ async fn make_test_app() -> App {
pending_shutdown_exit_thread_id: None,
windows_sandbox: WindowsSandboxState::default(),
thread_event_channels: HashMap::new(),
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashSet::new(),
thread_event_listener_tasks: HashMap::new(),
@@ -5933,6 +5954,9 @@ pub(super) async fn make_test_app_with_channels() -> (
pending_shutdown_exit_thread_id: None,
windows_sandbox: WindowsSandboxState::default(),
thread_event_channels: HashMap::new(),
pending_realtime_speech_replay: HashMap::new(),
pending_realtime_transcript_replay: HashMap::new(),
realtime_replay_order: VecDeque::new(),
temporary_structured_requests: HashMap::new(),
pending_thread_titles: HashSet::new(),
thread_event_listener_tasks: HashMap::new(),
@@ -8049,6 +8073,7 @@ async fn replay_thread_snapshot_replays_turn_history_in_order() {
let thread_id = ThreadId::new();
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(test_thread_session(
thread_id,
test_path_buf("/home/user/project"),
@@ -8165,6 +8190,7 @@ async fn replace_chat_widget_reseeds_collab_agent_metadata_for_replay() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: None,
turns: Vec::new(),
events: vec![ThreadBufferedEvent::Notification(Box::new(
@@ -8240,6 +8266,7 @@ async fn refreshed_snapshot_session_persists_resumed_turns() {
..initial_session.clone()
};
let mut snapshot = ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(initial_session),
turns: Vec::new(),
events: Vec::new(),
@@ -65,6 +65,7 @@ async fn switching_from_ultra_thread_restores_configured_plan_effort() {
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(ultra_session),
turns: Vec::new(),
events: Vec::new(),
@@ -74,6 +75,7 @@ async fn switching_from_ultra_thread_restores_configured_plan_effort() {
);
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(normal_session),
turns: Vec::new(),
events: Vec::new(),
@@ -258,6 +258,38 @@ async fn disconnected_command_center_keeps_input_and_blocks_actions() -> Result<
Ok(())
}
#[tokio::test]
async fn remote_disconnect_retires_voice_before_reconnecting() -> Result<()> {
let (mut app, _events, _ops) = make_test_app_with_channels().await;
let thread_id = ThreadId::new();
app.active_thread_id = Some(thread_id);
app.chat_widget
.handle_thread_session_quiet(test_thread_session(thread_id, app.config.cwd.to_path_buf()));
app.app_server_target = AppServerTarget::Remote {
endpoint: crate::resolve_remote_addr("ws://127.0.0.1:9")?,
};
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, thread_id);
assert!(app.chat_widget.is_current_realtime_attempt(
thread_id, /*attempt_id*/ 0, /*input_generation*/ 0
));
assert!(app.begin_reconnect());
assert_eq!(
(
app.reconnect.offline,
app.chat_widget.is_current_realtime_attempt(
thread_id, /*attempt_id*/ 0, /*input_generation*/ 0
),
),
(true, false)
);
assert!(app.begin_reconnect());
assert!(!app.chat_widget.is_current_realtime_attempt(
thread_id, /*attempt_id*/ 0, /*input_generation*/ 0
));
Ok(())
}
// Common framing/bootstrap for both transports. Returning None deliberately loses a reply.
pub(super) async fn serve_reconnect_requests<S, F>(
mut socket: tokio_tungstenite::WebSocketStream<S>,
@@ -36,6 +36,7 @@ async fn subagent_mcp_startup_settles_while_cached_servers_remain_deferred() {
app.activate_thread_channel(subagent_thread_id).await;
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(test_thread_session(
subagent_thread_id,
test_path_buf("/tmp/subagent"),
@@ -287,6 +287,7 @@ async fn replayed_patch_approval_pager_recovers_stored_turn_changes() {
let cwd = app.chat_widget.config_ref().cwd.to_path_buf();
app.replay_thread_snapshot(
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(test_thread_session(thread_id, cwd)),
turns: vec![test_turn(
TURN_ID,
@@ -0,0 +1,270 @@
//! Real Core-to-app-server-to-TUI handoff coverage without an audio device.
use super::*;
use crate::chatwidget::UserMessage;
use app_test_support::MockResponsesConfig;
use app_test_support::create_mock_responses_server_sequence_unchecked;
use codex_app_server_client::AppServerEvent;
use codex_app_server_protocol::ServerNotification;
use codex_login::AuthCredentialsStoreMode;
use codex_login::AuthKeyringBackendKind;
use codex_model_provider_info::ModelProviderInfo;
use core_test_support::responses;
use core_test_support::responses::WebSocketConnectionConfig;
use pretty_assertions::assert_eq;
use ratatui::buffer::Buffer;
use ratatui::layout::Rect;
use serde_json::json;
use std::time::Duration;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn delegated_core_events_keep_private_output_hidden_and_deliver_final_speech() -> Result<()> {
core_test_support::skip_if_no_network!(Ok(()));
let mut commentary =
responses::ev_assistant_message("private-commentary", "PRIVATE COMMENTARY");
commentary["item"]["phase"] = json!("commentary");
let mut final_answer = responses::ev_assistant_message(
"public-final",
"[ANALYSIS] is the marker you asked about.",
);
final_answer["item"]["phase"] = json!("final_answer");
let model_server = create_mock_responses_server_sequence_unchecked(vec![
responses::sse(vec![
responses::ev_response_created("delegated-response"),
responses::ev_reasoning_item("private-reasoning", &["PRIVATE REASONING"], &[]),
commentary,
final_answer,
responses::ev_completed("delegated-response"),
]),
responses::sse(vec![
responses::ev_response_created("typed-response"),
responses::ev_assistant_message("typed-final", "Typed answer stays visible."),
responses::ev_completed("typed-response"),
]),
])
.await;
Mock::given(method("POST"))
.and(path("/v1/live"))
.respond_with(
ResponseTemplate::new(200)
.insert_header("Location", "/v1/live/rtc_tui_handoff")
.set_body_string("v=answer\r\n"),
)
.mount(&model_server)
.await;
let realtime_server =
responses::start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
requests: vec![
vec![
json!({
"type": "session.started",
"session": { "id": "sess_tui_handoff", "instructions": "backend prompt" }
}),
json!({
"type": "delegation.created",
"offset_ms": 100,
"item": {
"id": "delegation_tui",
"type": "delegation",
"target": "client",
"content": [{"type": "input_text", "text": "explain the marker"}]
}
}),
],
vec![],
vec![],
vec![],
],
response_headers: Vec::new(),
accept_delay: None,
close_after_requests: false,
}])
.await;
let (mut app, mut app_events, mut ops) = make_test_app_with_channels().await;
let codex_home = tempfile::tempdir()?;
MockResponsesConfig::new(&model_server.uri())
.with_root_config(&format!(
"experimental_realtime_ws_base_url = {:?}\nexperimental_realtime_webrtc_call_base_url = {:?}",
realtime_server.uri(),
format!("{}/v1", model_server.uri()),
))
.with_extra_config("[realtime]\nversion = \"v3\"\ntype = \"conversational\"")
.write(codex_home.path())?;
codex_login::login_with_api_key(
codex_home.path(),
"sk-test-key",
AuthCredentialsStoreMode::File,
AuthKeyringBackendKind::default(),
)?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
app.config.cli_auth_credentials_store_mode = AuthCredentialsStoreMode::File;
app.config.model = Some("mock-model".to_string());
app.config.model_provider_id = "mock_provider".to_string();
app.config.model_provider = ModelProviderInfo {
name: "Mock provider for test".to_string(),
base_url: Some(format!("{}/v1", model_server.uri())),
request_max_retries: Some(0),
stream_max_retries: Some(0),
..ModelProviderInfo::default()
};
app.config.experimental_realtime_ws_base_url = Some(realtime_server.uri().to_string());
app.config.experimental_realtime_webrtc_call_base_url =
Some(format!("{}/v1", model_server.uri()));
let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?;
let started = app_server.start_thread(&app.config).await?;
let thread_id = started.session.thread_id;
app.enqueue_primary_thread_session(started.session, started.turns)
.await?;
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, thread_id);
while ops.try_recv().is_ok() {}
let mut tui = crate::tui::test_support::make_test_tui()?;
app_server
.thread_realtime_start(thread_id, "v=offer\r\n".to_string(), /*model*/ None)
.await?;
// The sideband fixture sends its delegation after the first outbound request.
app_server
.thread_realtime_append_speech(thread_id, "fixture prompt".to_string())
.await?;
let mut rendered = Vec::new();
let speech = timeout(Duration::from_secs(15), async {
loop {
tokio::select! {
event = app_server.next_event() => {
app.handle_app_server_event(
&app_server,
event.expect("embedded app-server should stay connected"),
).await;
app.drain_active_thread_events(&mut tui).await?;
}
event = app_events.recv() => {
let event = event.expect("TUI event stream should stay open");
if let AppEvent::InsertHistoryCell(cell) = &event {
rendered.extend(cell.display_lines(/*width*/ 80).into_iter().map(|line| line.to_string()));
}
app.handle_event(&mut tui, &mut app_server, event).await?;
}
op = ops.recv() => {
let op = op.expect("TUI command stream should stay open");
if matches!(op, Op::RealtimeConversationSpeech { .. }) {
break Ok::<_, color_eyre::Report>(op);
}
app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(op)).await?;
}
}
}
})
.await??;
assert!(
matches!(&speech, Op::RealtimeConversationSpeech { text, .. } if text.as_str() == "[ANALYSIS] is the marker you asked about."),
"unexpected speech: {speech:?}"
);
app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(speech))
.await?;
assert_eq!(
timeout(
Duration::from_secs(5),
realtime_server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1),
)
.await?
.body_json(),
json!({
"type": "session.context.append",
"content": [{
"type": "input_text",
"text": "[ANALYSIS] is the marker you asked about."
}],
"channel": "speakable"
})
);
let rendered = rendered.join("\n");
assert!(!rendered.contains("PRIVATE REASONING"), "{rendered}");
assert!(!rendered.contains("PRIVATE COMMENTARY"), "{rendered}");
let area = Rect::new(
/*x*/ 0, /*y*/ 0, /*width*/ 80, /*height*/ 24,
);
let mut buffer = Buffer::empty(area);
app.chat_widget.render(area, &mut buffer);
let visible = buffer
.content()
.iter()
.map(ratatui::buffer::Cell::symbol)
.collect::<String>();
assert!(!visible.contains("PRIVATE REASONING"), "{visible}");
assert!(!visible.contains("PRIVATE COMMENTARY"), "{visible}");
assert!(
ops.try_recv().is_err(),
"a final answer must be sent only once"
);
// A typed request on the same live thread must follow the normal visible path.
app.chat_widget
.restore_user_message_to_composer(UserMessage::from("typed follow-up"));
app.chat_widget
.handle_key_event(crossterm::event::KeyEvent::new(
crossterm::event::KeyCode::Enter,
crossterm::event::KeyModifiers::NONE,
));
let typed_op = next_user_turn_op(&mut ops);
app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(typed_op))
.await?;
let mut typed_rendered = Vec::new();
timeout(Duration::from_secs(10), async {
loop {
tokio::select! {
event = app_server.next_event() => {
let event = event.expect("embedded app-server should stay connected");
let completed = matches!(&event, AppServerEvent::ServerNotification(notification)
if matches!(notification.as_ref(), ServerNotification::TurnCompleted(_)));
app.handle_app_server_event(&app_server, event).await;
app.drain_active_thread_events(&mut tui).await?;
if completed {
break Ok::<_, color_eyre::Report>(());
}
}
event = app_events.recv() => {
let event = event.expect("TUI event stream should stay open");
if let AppEvent::InsertHistoryCell(cell) = &event {
typed_rendered.extend(cell.display_lines(/*width*/ 80).into_iter().map(|line| line.to_string()));
}
app.handle_event(&mut tui, &mut app_server, event).await?;
}
op = ops.recv() => {
let op = op.expect("TUI command stream should stay open");
assert!(!matches!(op, Op::RealtimeConversationSpeech { .. }), "typed answer must not be spoken");
app.handle_event(&mut tui, &mut app_server, AppEvent::CodexOp(op)).await?;
}
}
}
})
.await??;
while let Ok(event) = app_events.try_recv() {
if let AppEvent::InsertHistoryCell(cell) = &event {
typed_rendered.extend(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string()),
);
}
app.handle_event(&mut tui, &mut app_server, event).await?;
}
assert!(
typed_rendered
.join("\n")
.contains("Typed answer stays visible."),
"typed final must appear in history: {typed_rendered:?}"
);
assert!(ops.try_recv().is_err(), "typed answer must not be spoken");
realtime_server.shutdown().await;
app_server.shutdown().await?;
Ok(())
}
@@ -0,0 +1,64 @@
//! Voice startup rejection coverage without an audio device.
use super::*;
use crate::app::tests::session_lifecycle_requests::recorded_params;
use crate::app::tests::session_lifecycle_requests::start_recording_remote_app_server;
#[tokio::test]
async fn rejected_voice_start_does_not_leave_local_capture_active() -> Result<()> {
for replay_only in [false, true] {
let (mut app, mut events, _ops) = make_test_app_with_channels().await;
let (mut app_server, requests, proxy) =
start_recording_remote_app_server(&app.config).await?;
let thread_id = ThreadId::new();
app.chat_widget
.handle_thread_session_quiet(test_thread_session(
thread_id,
app.config.cwd.to_path_buf(),
));
crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, thread_id);
assert!(app.chat_widget.may_receive_realtime_transcripts());
if replay_only {
app.app_server_target = AppServerTarget::Remote {
endpoint: crate::resolve_remote_addr("ws://127.0.0.1:9")?,
};
app.active_thread_id = Some(thread_id);
app.ensure_thread_channel(thread_id).mark_replay_only();
assert!(app.thread_unavailable(thread_id));
}
let mut tui = crate::tui::test_support::make_test_tui()?;
Box::pin(app.handle_event(
&mut tui,
&mut app_server,
AppEvent::CodexOp(Op::RealtimeConversationStart {
thread_id,
offer_sdp: String::from("v=0\r\n").into(),
}),
))
.await?;
assert!(!app.chat_widget.may_receive_realtime_transcripts());
assert!(recorded_params(&requests, "thread/realtime/start").is_empty());
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
if replay_only {
insta::assert_snapshot!("rejected_voice_start_replay_only", rendered);
} else {
insta::assert_snapshot!("rejected_voice_start_no_active_thread", rendered);
}
app_server.shutdown().await?;
proxy.await??;
}
Ok(())
}
@@ -0,0 +1,6 @@
---
source: tui/src/app/tests/realtime_start.rs
assertion_line: 59
expression: rendered
---
■ No active thread is available.
@@ -0,0 +1,6 @@
---
source: tui/src/app/tests/realtime_start.rs
assertion_line: 57
expression: rendered
---
■ This conversation is read-only or unavailable; no operation was sent.
+1
View File
@@ -1406,6 +1406,7 @@ async fn ignore_same_thread_resume_allows_retrying_read_only_view() -> Result<()
&app_server,
thread_id,
ThreadEventSnapshot {
delegated_turns: Vec::new(),
session: Some(session),
turns: vec![test_turn("running", TurnStatus::InProgress, Vec::new())],
events: Vec::new(),
+241
View File
@@ -11,6 +11,7 @@ use std::borrow::Cow;
#[derive(Debug, Clone)]
pub(super) struct ThreadEventSnapshot {
pub(super) session: Option<ThreadSessionState>,
pub(super) delegated_turns: Vec<String>,
pub(super) turns: Vec<Turn>,
pub(super) events: Vec<ThreadBufferedEvent>,
pub(super) input_state: Option<ThreadInputState>,
@@ -24,6 +25,24 @@ pub(super) enum ThreadBufferedEvent {
FeedbackSubmission(FeedbackThreadEvent),
}
fn is_voice_handoff_item(item: &ThreadItem) -> bool {
matches!(item, ThreadItem::UserMessage { content, .. }
if crate::chatwidget::realtime_delegation_input(content).is_some())
}
fn hide_private_items_after_voice_handoff(items: &mut Vec<ThreadItem>, delegated: bool) {
let has_marker = items.iter().any(is_voice_handoff_item);
// An evicted marker leaves no reliable order, so hide private items
// conservatively. A present marker lets earlier typed output survive.
let mut voice_started = delegated && !has_marker;
items.retain(|item| {
if is_voice_handoff_item(item) {
voice_started = true;
}
!voice_started || !crate::chatwidget::is_private_realtime_agent_item(item)
});
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) struct FeedbackThreadEvent {
pub(super) category: FeedbackCategory,
@@ -53,6 +72,7 @@ pub(super) struct ThreadEventStore {
pub(super) capacity: usize,
pub(super) active: bool,
pub(super) buffered_agent_message_delta_bytes: usize,
delegated_turns: VecDeque<String>,
recap_progress: recap::RecapProgress,
}
@@ -95,6 +115,7 @@ impl ThreadEventStore {
capacity,
active: false,
buffered_agent_message_delta_bytes: 0,
delegated_turns: VecDeque::new(),
recap_progress: recap::RecapProgress::default(),
}
}
@@ -147,6 +168,18 @@ impl ThreadEventStore {
}
fn push_notification_inner(&mut self, notification: Cow<'_, ServerNotification>) {
let user_item = match notification.as_ref() {
ServerNotification::ItemStarted(n) => Some((&n.turn_id, &n.item)),
ServerNotification::ItemCompleted(n) => Some((&n.turn_id, &n.item)),
_ => None,
};
if let Some((turn_id, ThreadItem::UserMessage { content, .. })) = user_item
&& turn_id.len() <= 512
&& crate::chatwidget::realtime_delegation_input(content).is_some()
&& !self.delegated_turns.contains(turn_id)
{
self.delegated_turns.push_back(turn_id.clone());
}
self.pending_interactive_replay
.note_server_notification(notification.as_ref());
match notification.as_ref() {
@@ -204,6 +237,27 @@ impl ThreadEventStore {
}
self.push_replay_notification(notification);
self.trim_delegated_turns();
}
fn trim_delegated_turns(&mut self) {
while self.delegated_turns.len() > self.capacity {
// Keep an evicted user marker while this turn still has buffered output.
let stale = self.delegated_turns.iter().position(|turn_id| {
!self.buffer.iter().any(|event| {
matches!(event, ThreadBufferedEvent::Notification(n) if match n.as_ref() {
ServerNotification::ItemCompleted(item) => &item.turn_id == turn_id,
ServerNotification::AgentMessageDelta(delta) => &delta.turn_id == turn_id,
ServerNotification::ReasoningSummaryTextDelta(delta) => &delta.turn_id == turn_id,
ServerNotification::ReasoningTextDelta(delta) => &delta.turn_id == turn_id,
ServerNotification::ReasoningSummaryPartAdded(part) => &part.turn_id == turn_id,
ServerNotification::TurnCompleted(turn) => &turn.turn.id == turn_id,
_ => false,
})
})
});
self.delegated_turns.remove(stale.unwrap_or(0));
}
}
pub(super) fn push_request(&mut self, request: ServerRequest) {
@@ -242,6 +296,7 @@ impl ThreadEventStore {
pub(super) fn snapshot(&self) -> ThreadEventSnapshot {
let mut snapshot = ThreadEventSnapshot {
session: self.session.clone(),
delegated_turns: self.delegated_turns.iter().cloned().collect(),
turns: self.turns.clone(),
// Thread switches replay buffered events into a rebuilt ChatWidget. Only replay
// interactive prompts that are still pending, or answered approvals/input will reappear.
@@ -263,6 +318,99 @@ impl ThreadEventStore {
if let Some(latest_turn_id) = &self.latest_turn_id {
replay_filter::omit_resolved_misalignment_errors(&mut snapshot, latest_turn_id);
}
let delegated = snapshot
.turns
.iter()
.filter(|turn| {
turn.items.iter().any(|item| {
matches!(item, ThreadItem::UserMessage { content, .. }
if crate::chatwidget::realtime_delegation_input(content).is_some())
})
})
.map(|turn| turn.id.clone())
.chain(self.delegated_turns.iter().cloned())
.collect::<std::collections::HashSet<_>>();
for turn in &mut snapshot.turns {
hide_private_items_after_voice_handoff(&mut turn.items, delegated.contains(&turn.id));
}
let buffered_markers = snapshot
.events
.iter()
.filter_map(|event| match event {
ThreadBufferedEvent::Notification(notification) => match notification.as_ref() {
ServerNotification::ItemStarted(n) if is_voice_handoff_item(&n.item) => {
Some(n.turn_id.clone())
}
ServerNotification::ItemCompleted(n) if is_voice_handoff_item(&n.item) => {
Some(n.turn_id.clone())
}
_ => None,
},
_ => None,
})
.collect::<std::collections::HashSet<_>>();
let mut voice_started = delegated
.difference(&buffered_markers)
.cloned()
.collect::<std::collections::HashSet<_>>();
let mut typed_items = std::collections::HashSet::new();
snapshot.events.retain_mut(|event| {
let ThreadBufferedEvent::Notification(notification) = event else {
return true;
};
match notification.as_ref() {
ServerNotification::ItemStarted(n) if is_voice_handoff_item(&n.item) => {
voice_started.insert(n.turn_id.clone());
}
ServerNotification::ItemCompleted(n) if is_voice_handoff_item(&n.item) => {
voice_started.insert(n.turn_id.clone());
}
ServerNotification::ItemStarted(n) => {
if let ThreadItem::AgentMessage { id, .. } | ThreadItem::Reasoning { id, .. } =
&n.item
{
let key = (n.turn_id.clone(), id.clone());
if voice_started.contains(&n.turn_id) {
typed_items.remove(&key);
} else {
typed_items.insert(key);
}
}
}
_ => {}
}
match notification.as_mut() {
ServerNotification::AgentMessageDelta(n) => {
!voice_started.contains(&n.turn_id)
|| typed_items.contains(&(n.turn_id.clone(), n.item_id.clone()))
}
ServerNotification::ReasoningSummaryTextDelta(n) => {
!voice_started.contains(&n.turn_id)
|| typed_items.contains(&(n.turn_id.clone(), n.item_id.clone()))
}
ServerNotification::ReasoningTextDelta(n) => {
!voice_started.contains(&n.turn_id)
|| typed_items.contains(&(n.turn_id.clone(), n.item_id.clone()))
}
ServerNotification::ReasoningSummaryPartAdded(n) => {
!voice_started.contains(&n.turn_id)
|| typed_items.contains(&(n.turn_id.clone(), n.item_id.clone()))
}
ServerNotification::ItemCompleted(n) if voice_started.contains(&n.turn_id) => {
matches!(&n.item, ThreadItem::AgentMessage { id, .. } | ThreadItem::Reasoning { id, .. }
if typed_items.contains(&(n.turn_id.clone(), id.clone())))
|| !crate::chatwidget::is_private_realtime_agent_item(&n.item)
}
ServerNotification::TurnCompleted(n) if delegated.contains(&n.turn.id) => {
hide_private_items_after_voice_handoff(
&mut n.turn.items,
/*delegated*/ true,
);
true
}
_ => true,
}
});
snapshot
}
@@ -430,12 +578,16 @@ mod tests {
use codex_app_server_protocol::HookRunSummary as AppServerHookRunSummary;
use codex_app_server_protocol::HookScope as AppServerHookScope;
use codex_app_server_protocol::HookStartedNotification;
use codex_app_server_protocol::ItemCompletedNotification;
use codex_app_server_protocol::ItemStartedNotification;
use codex_app_server_protocol::McpToolCallProgressNotification;
use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification;
use codex_app_server_protocol::RequestId as AppServerRequestId;
use codex_app_server_protocol::ThreadRealtimeAudioChunk;
use codex_app_server_protocol::ThreadRealtimeOutputAudioDeltaNotification;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartedNotification;
use codex_app_server_protocol::UserInput;
use codex_config::types::ApprovalsReviewer;
use codex_protocol::models::PermissionProfile;
use pretty_assertions::assert_eq;
@@ -818,4 +970,93 @@ mod tests {
serde_json::to_value(notification).expect("MCP notification should serialize"),
);
}
#[test]
fn buffered_voice_handoff_retains_only_preexisting_typed_reasoning() {
let thread_id = ThreadId::new().to_string();
let turn_id = "mixed-reasoning-turn".to_string();
let mut store = ThreadEventStore::new(/*capacity*/ 16);
let typed = ThreadItem::Reasoning {
id: "typed-reasoning".into(),
summary: Vec::new(),
content: Vec::new(),
};
store.push_notification(ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
item: typed,
started_at_ms: 0,
}));
store.push_notification(ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
item: ThreadItem::UserMessage {
id: "voice-marker".into(),
client_id: None,
content: vec![UserInput::Text {
text:
"<realtime_delegation><input>spoken follow-up</input></realtime_delegation>"
.into(),
text_elements: Vec::new(),
}],
},
started_at_ms: 0,
}));
for (id, text) in [
("typed-reasoning", "Typed tail"),
("private-reasoning", "Private voice reasoning"),
] {
if id == "private-reasoning" {
store.push_notification(ServerNotification::ItemStarted(ItemStartedNotification {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
item: ThreadItem::Reasoning {
id: id.into(),
summary: Vec::new(),
content: Vec::new(),
},
started_at_ms: 0,
}));
}
store.push_notification(ServerNotification::ReasoningSummaryTextDelta(
ReasoningSummaryTextDeltaNotification {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
item_id: id.into(),
delta: text.into(),
summary_index: 0,
},
));
store.push_notification(ServerNotification::ItemCompleted(
ItemCompletedNotification {
thread_id: thread_id.clone(),
turn_id: turn_id.clone(),
item: ThreadItem::Reasoning {
id: id.into(),
summary: vec![text.into()],
content: Vec::new(),
},
completed_at_ms: 0,
},
));
}
let retained = store
.snapshot()
.events
.into_iter()
.filter_map(|event| match event {
ThreadBufferedEvent::Notification(notification) => match *notification {
ServerNotification::ReasoningSummaryTextDelta(n) => Some(n.delta),
ServerNotification::ItemCompleted(n) => match n.item {
ThreadItem::Reasoning { summary, .. } => Some(summary.join("")),
_ => None,
},
_ => None,
},
_ => None,
})
.collect::<Vec<_>>();
assert_eq!(retained, vec!["Typed tail", "Typed tail"]);
}
}
+117
View File
@@ -13,8 +13,32 @@ use codex_app_server_protocol::TurnInterruptParams;
use codex_app_server_protocol::TurnInterruptResponse;
use codex_app_server_protocol::WarningNotification;
// Leave time for side-thread cleanup and unsubscribe inside the two-second exit budget.
const REALTIME_STOP_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 1);
impl App {
pub(super) async fn stop_realtime_conversation(&mut self, app_server: &mut AppServerSession) {
let Some(thread_id) = self.chat_widget.reset_realtime_conversation() else {
return;
};
match tokio::time::timeout(
REALTIME_STOP_TIMEOUT,
app_server.thread_realtime_stop(thread_id),
)
.await
{
Ok(Ok(())) => {}
Ok(Err(error)) => {
tracing::warn!(%thread_id, %error, "failed to stop voice conversation");
}
Err(_) => {
tracing::warn!(%thread_id, "timed out stopping voice conversation");
}
}
}
pub(super) async fn shutdown_current_thread(&mut self, app_server: &mut AppServerSession) {
self.stop_realtime_conversation(app_server).await;
self.shutdown_side_threads(app_server).await;
if let Some(thread_id) = self.chat_widget.thread_id() {
if let Err(err) = app_server.thread_unsubscribe(thread_id).await {
@@ -446,6 +470,15 @@ impl App {
op: AppCommand,
) -> Result<()> {
let Some(thread_id) = self.active_thread_id else {
if let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = &op {
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
}
if let AppCommand::RealtimeConversationStart { thread_id, .. } = &op
&& self.chat_widget.thread_id() == Some(*thread_id)
{
self.chat_widget.reset_realtime_conversation();
}
self.chat_widget
.add_error_message("No active thread is available.".to_string());
return Ok(());
@@ -461,12 +494,25 @@ impl App {
op: AppCommand,
) -> Result<()> {
if self.thread_unavailable(thread_id) {
if let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = &op {
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
}
if let AppCommand::RealtimeConversationStart { thread_id, .. } = &op
&& self.chat_widget.thread_id() == Some(*thread_id)
{
self.chat_widget.reset_realtime_conversation();
}
self.chat_widget.add_error_message(
"This conversation is read-only or unavailable; no operation was sent.".into(),
);
return Ok(());
}
if self.chat_widget.rejects_misalignment_policy_op(&op) {
if let AppCommand::RealtimeConversationSpeech { delivery_id, .. } = &op {
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
}
return Ok(());
}
@@ -867,6 +913,60 @@ impl App {
.await?;
Ok(true)
}
AppCommand::RealtimeConversationStart {
thread_id: realtime_thread_id,
offer_sdp,
} => {
if *realtime_thread_id != thread_id {
return Ok(true);
}
let model = self
.chat_widget
.config_ref()
.experimental_realtime_ws_model
.clone();
app_server
.thread_realtime_start(
*realtime_thread_id,
String::from(offer_sdp.clone()),
model,
)
.await?;
Ok(true)
}
AppCommand::RealtimeConversationStop {
thread_id: realtime_thread_id,
} => {
app_server.thread_realtime_stop(*realtime_thread_id).await?;
Ok(true)
}
AppCommand::RealtimeConversationSpeech {
thread_id: realtime_thread_id,
attempt_id,
input_generation,
delivery_id,
text,
} => {
if !self.chat_widget.has_pending_realtime_speech(*delivery_id) {
return Ok(true);
}
if *realtime_thread_id != thread_id
|| !self.chat_widget.is_current_realtime_attempt(
*realtime_thread_id,
*attempt_id,
*input_generation,
)
{
self.chat_widget
.restore_undelivered_realtime_speech(*delivery_id);
return Ok(true);
}
app_server
.thread_realtime_append_speech(*realtime_thread_id, text.as_str().to_owned())
.await?;
self.chat_widget.accept_realtime_speech(*delivery_id);
Ok(true)
}
AppCommand::RunUserShellCommand { command } => {
app_server
.thread_shell_command(thread_id, command.to_string())
@@ -1136,6 +1236,7 @@ impl App {
guard.push_notification_ref(&notification);
Some(notification)
} else {
self.retain_inactive_realtime_transcript(thread_id, &notification);
guard.push_notification(notification);
None
};
@@ -1423,6 +1524,8 @@ impl App {
}
}
let should_buffer_initial_replay = !turns.is_empty();
let replayed_final_items = realtime_delivery::completed_agent_items_from_turns(&turns);
let replayed_voice_texts = realtime_delivery::replayed_voice_texts_from_turns(&turns);
if should_buffer_initial_replay {
self.app_event_tx
.send(AppEvent::BeginInitialHistoryReplayBuffer);
@@ -1438,6 +1541,10 @@ impl App {
self.app_event_tx
.send(AppEvent::EndInitialHistoryReplayBuffer);
}
self.restore_realtime_replay_state_after_replay(
&replayed_final_items,
replayed_voice_texts,
);
if matches!(presentation, ThreadAttachPresentation::PromptEdit) {
self.chat_widget.emit_prompt_edit_thread_event();
}
@@ -1651,6 +1758,8 @@ impl App {
mut snapshot: ThreadEventSnapshot,
resume_restored_queue: bool,
) {
let replayed_final_items = realtime_delivery::completed_agent_items(&snapshot);
let replayed_voice_texts = realtime_delivery::replayed_voice_texts(&snapshot);
replay_filter::omit_completed_agent_deltas(&mut snapshot.events);
let request_changes = snapshot
.events
@@ -1694,6 +1803,10 @@ impl App {
self.chat_widget.handle_thread_session(session);
}
}
for turn_id in &snapshot.delegated_turns {
self.chat_widget
.remember_realtime_delegated_reasoning_turn(turn_id);
}
let recovered_input = snapshot
.input_state
.as_ref()
@@ -1728,6 +1841,10 @@ impl App {
if recovered_input.is_some() {
self.chat_widget.restore_reconnected_input(recovered_input);
}
self.restore_realtime_replay_state_after_replay(
&replayed_final_items,
replayed_voice_texts,
);
self.chat_widget
.set_queue_autosend_suppressed(/*suppressed*/ false);
self.chat_widget
+91
View File
@@ -1,3 +1,4 @@
use std::fmt;
use std::path::PathBuf;
use codex_app_server_protocol::AskForApproval;
@@ -9,6 +10,7 @@ use codex_app_server_protocol::ReviewTarget;
use codex_app_server_protocol::ToolRequestUserInputResponse;
use codex_app_server_protocol::UserInput;
use codex_config::types::ApprovalsReviewer;
use codex_protocol::ThreadId;
use codex_protocol::approvals::GuardianAssessmentEvent;
use codex_protocol::config_types::CollaborationMode;
use codex_protocol::config_types::Personality;
@@ -19,13 +21,98 @@ use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig;
use codex_protocol::request_permissions::RequestPermissionsResponse;
use serde::Serialize;
use serde::Serializer;
use serde_json::Value;
/// Keeps ICE credentials out of command diagnostics and session recordings.
/// Convert back to a string only when sending the offer to app-server signaling.
#[derive(Clone, PartialEq, Eq)]
pub(crate) struct RealtimeOfferSdp(String);
impl From<String> for RealtimeOfferSdp {
fn from(value: String) -> Self {
Self(value)
}
}
impl From<RealtimeOfferSdp> for String {
fn from(value: RealtimeOfferSdp) -> Self {
value.0
}
}
impl fmt::Debug for RealtimeOfferSdp {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("<redacted>")
}
}
impl Serialize for RealtimeOfferSdp {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str("<redacted>")
}
}
/// Keeps spoken answers out of command diagnostics and session recordings.
/// Convert back to a string only at the app-server signaling boundary.
#[derive(Clone, PartialEq, Eq)]
pub(crate) struct RealtimeSpeechText(String);
impl From<String> for RealtimeSpeechText {
fn from(value: String) -> Self {
Self(value)
}
}
impl From<RealtimeSpeechText> for String {
fn from(value: RealtimeSpeechText) -> Self {
value.0
}
}
impl RealtimeSpeechText {
pub(crate) fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Debug for RealtimeSpeechText {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("<redacted>")
}
}
impl Serialize for RealtimeSpeechText {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: Serializer,
{
serializer.serialize_str("<redacted>")
}
}
#[allow(clippy::large_enum_variant)]
#[derive(Debug, Clone, PartialEq, Serialize)]
pub(crate) enum AppCommand {
Interrupt,
CleanBackgroundTerminals,
RealtimeConversationStart {
thread_id: ThreadId,
offer_sdp: RealtimeOfferSdp,
},
RealtimeConversationStop {
thread_id: ThreadId,
},
RealtimeConversationSpeech {
thread_id: ThreadId,
attempt_id: u64,
input_generation: u64,
delivery_id: u64,
text: RealtimeSpeechText,
},
RunUserShellCommand {
command: String,
},
@@ -252,3 +339,7 @@ impl From<&AppCommand> for AppCommand {
value.clone()
}
}
#[cfg(test)]
#[path = "app_command_tests.rs"]
mod tests;
+70
View File
@@ -0,0 +1,70 @@
//! Protect connection credentials in diagnostics without changing signaling data.
use super::AppCommand;
use codex_protocol::ThreadId;
use pretty_assertions::assert_eq;
use serde_json::json;
#[test]
fn realtime_offer_is_redacted_in_diagnostics_and_preserved_for_signaling() {
let thread_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000001").expect("valid thread id");
let original_offer =
"v=0\r\na=ice-ufrag:private-user\r\na=ice-pwd:private-credential\r\n".to_string();
let command = AppCommand::RealtimeConversationStart {
thread_id,
offer_sdp: original_offer.clone().into(),
};
assert_eq!(
format!("{command:?}"),
format!("RealtimeConversationStart {{ thread_id: {thread_id:?}, offer_sdp: <redacted> }}")
);
assert_eq!(
serde_json::to_value(&command).expect("command diagnostics serialize"),
json!({
"RealtimeConversationStart": {
"thread_id": thread_id,
"offer_sdp": "<redacted>"
}
})
);
let AppCommand::RealtimeConversationStart { offer_sdp, .. } = command else {
panic!("expected the start command");
};
assert_eq!(String::from(offer_sdp), original_offer);
}
#[test]
fn realtime_speech_is_redacted_in_diagnostics_and_preserved_for_signaling() {
let thread_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000001").expect("valid thread id");
let original_text = "Private spoken answer".to_string();
let command = AppCommand::RealtimeConversationSpeech {
thread_id,
attempt_id: 1,
input_generation: 2,
delivery_id: 3,
text: original_text.clone().into(),
};
assert!(!format!("{command:?}").contains(&original_text));
assert_eq!(
serde_json::to_value(&command).expect("command diagnostics serialize"),
json!({
"RealtimeConversationSpeech": {
"thread_id": thread_id,
"attempt_id": 1,
"input_generation": 2,
"delivery_id": 3,
"text": "<redacted>"
}
})
);
let AppCommand::RealtimeConversationSpeech { text, .. } = command else {
panic!("expected the speech command");
};
assert_eq!(String::from(text), original_text);
}
+20
View File
@@ -63,6 +63,7 @@ use codex_plugin::PluginCapabilitySummary;
use codex_protocol::config_types::CollaborationModeMask;
use codex_protocol::config_types::Personality;
use codex_protocol::models::ActivePermissionProfile;
use codex_realtime_webrtc::StartedRealtimeWebrtcSession;
use crate::history_cell::HistoryCell;
@@ -1053,6 +1054,25 @@ pub(crate) enum AppEvent {
/// Update the current personality in the running app and widget.
UpdatePersonality(Personality),
/// Result of creating a TUI-owned WebRTC offer for an active thread.
RealtimeWebrtcOfferCreated {
thread_id: ThreadId,
attempt_id: u64,
result: Result<StartedRealtimeWebrtcSession, String>,
},
/// Result of establishing the WebRTC connection for an active voice attempt.
RealtimeWebrtcConnected {
thread_id: ThreadId,
attempt_id: u64,
result: Result<(), codex_realtime_webrtc::ConnectionError>,
},
/// Stop voice on its original thread after its chat widget is replaced.
StopRealtimeConversation {
thread_id: ThreadId,
},
/// Finish a settings selection after its preceding update events have been applied.
SettingsSelectionClosed,
/// Run after any nested settings events emitted while handling the close event.
+1
View File
@@ -6,6 +6,7 @@
mod fs;
mod history;
mod models;
mod realtime;
mod rollout_history;
mod thread_list;
@@ -0,0 +1,91 @@
//! App-server signaling for TUI-owned realtime WebRTC sessions.
use super::AppServerSession;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ThreadRealtimeAppendSpeechParams;
use codex_app_server_protocol::ThreadRealtimeAppendSpeechResponse;
use codex_app_server_protocol::ThreadRealtimeStartParams;
use codex_app_server_protocol::ThreadRealtimeStartResponse;
use codex_app_server_protocol::ThreadRealtimeStartTransport;
use codex_app_server_protocol::ThreadRealtimeStopParams;
use codex_app_server_protocol::ThreadRealtimeStopResponse;
use codex_protocol::ThreadId;
use codex_protocol::protocol::RealtimeConversationVersion;
use codex_protocol::protocol::RealtimeOutputModality;
use color_eyre::eyre::Result;
use color_eyre::eyre::WrapErr;
impl AppServerSession {
pub(crate) async fn thread_realtime_start(
&mut self,
thread_id: ThreadId,
offer_sdp: String,
model: Option<String>,
) -> Result<()> {
let request_id = self.next_request_id();
let _: ThreadRealtimeStartResponse = self
.client
.request_typed(ClientRequest::ThreadRealtimeStart {
request_id,
params: ThreadRealtimeStartParams {
thread_id: thread_id.to_string(),
client_managed_handoffs: Some(true),
delegation_ack_filler: None,
flush_transcript_tail_on_session_end: None,
codex_responses_as_items: None,
codex_response_item_prefix: None,
codex_response_handoff_mode: None,
codex_response_handoff_channel_prefixes: None,
model,
output_modality: RealtimeOutputModality::Audio,
include_startup_context: Some(false),
initial_items: None,
realtime_start_instructions: None,
realtime_end_instructions: None,
prompt: None,
realtime_session_id: None,
transport: Some(ThreadRealtimeStartTransport::Webrtc { sdp: offer_sdp }),
version: Some(RealtimeConversationVersion::V3),
voice: None,
},
})
.await
.wrap_err("thread/realtime/start failed in TUI")?;
Ok(())
}
pub(crate) async fn thread_realtime_stop(&mut self, thread_id: ThreadId) -> Result<()> {
let request_id = self.next_request_id();
let _: ThreadRealtimeStopResponse = self
.client
.request_typed(ClientRequest::ThreadRealtimeStop {
request_id,
params: ThreadRealtimeStopParams {
thread_id: thread_id.to_string(),
},
})
.await
.wrap_err("thread/realtime/stop failed in TUI")?;
Ok(())
}
pub(crate) async fn thread_realtime_append_speech(
&mut self,
thread_id: ThreadId,
text: String,
) -> Result<()> {
let request_id = self.next_request_id();
let _: ThreadRealtimeAppendSpeechResponse = self
.client
.request_typed(ClientRequest::ThreadRealtimeAppendSpeech {
request_id,
params: ThreadRealtimeAppendSpeechParams {
thread_id: thread_id.to_string(),
text,
},
})
.await
.wrap_err("thread/realtime/appendSpeech failed in TUI")?;
Ok(())
}
}
+15 -20
View File
@@ -554,6 +554,7 @@ pub(crate) struct ChatComposer {
mentions_v2_enabled: bool,
goal_command_enabled: bool,
personality_command_enabled: bool,
voice_command_enabled: bool,
worktrees_enabled: bool,
windows_degraded_sandbox_active: bool,
side_conversation_active: bool,
@@ -718,6 +719,7 @@ impl ChatComposer {
mentions_v2_enabled: false,
goal_command_enabled: false,
personality_command_enabled: false,
voice_command_enabled: false,
worktrees_enabled: false,
windows_degraded_sandbox_active: false,
side_conversation_active: false,
@@ -924,6 +926,10 @@ impl ChatComposer {
self.goal_command_enabled = enabled;
}
pub fn set_voice_command_enabled(&mut self, enabled: bool) {
self.voice_command_enabled = enabled;
}
/// Replace composer, editor, and footer-hint key bindings from one runtime snapshot.
///
/// Submit and queue bindings are cached here because composer dispatch must
@@ -1035,19 +1041,11 @@ impl ChatComposer {
.unwrap_or_else(|| footer_height(&footer_props));
let footer_spacing = Self::footer_spacing(footer_hint_height);
let footer_total_height = footer_hint_height + footer_spacing;
let popup_constraint = match &self.popups.active {
ActivePopup::Command(popup) => {
Constraint::Max(popup.calculate_required_height(area.width))
}
ActivePopup::File(popup) => Constraint::Max(popup.calculate_required_height()),
ActivePopup::Skill(popup) => {
Constraint::Max(popup.calculate_required_height(area.width))
}
ActivePopup::MentionV2(popup) => {
Constraint::Max(popup.calculate_required_height(area.width))
}
ActivePopup::None => Constraint::Max(footer_total_height),
};
let popup_constraint = Constraint::Max(
self.popups
.active
.required_height(area.width, footer_total_height),
);
let [composer_rect, popup_rect] =
Layout::vertical([Constraint::Min(3), popup_constraint]).areas(area);
let mut textarea_rect = composer_rect.inset(Insets::tlbr(
@@ -4656,13 +4654,10 @@ impl ChatComposer {
+ remote_images_height
+ remote_images_separator
+ 2
+ match &self.popups.active {
ActivePopup::None => footer_total_height,
ActivePopup::Command(c) => c.calculate_required_height(width),
ActivePopup::File(c) => c.calculate_required_height(),
ActivePopup::Skill(c) => c.calculate_required_height(width),
ActivePopup::MentionV2(c) => c.calculate_required_height(width),
}
+ self
.popups
.active
.required_height(width, footer_total_height)
}
}
@@ -112,6 +112,18 @@ pub(super) enum ActivePopup {
MentionV2(MentionV2Popup),
}
impl ActivePopup {
pub(super) fn required_height(&self, width: u16, footer_total_height: u16) -> u16 {
match self {
Self::None => footer_total_height,
Self::Command(popup) => popup.calculate_required_height(width),
Self::File(popup) => popup.calculate_required_height(),
Self::Skill(popup) => popup.calculate_required_height(width),
Self::MentionV2(popup) => popup.calculate_required_height(width),
}
}
}
#[cfg(test)]
#[path = "popup_state_tests.rs"]
mod tests;
@@ -178,6 +178,7 @@ impl<'a> SlashInput<'a> {
service_tier_commands_enabled: self.command_flags.service_tier_commands_enabled,
goal_command_enabled: self.command_flags.goal_command_enabled,
personality_command_enabled: self.command_flags.personality_command_enabled,
voice_command_enabled: self.command_flags.voice_command_enabled,
worktrees_enabled: self.command_flags.worktrees_enabled,
windows_degraded_sandbox_active: self.command_flags.allow_elevate_sandbox,
side_conversation_active: self.command_flags.side_conversation_active,
@@ -216,6 +217,7 @@ impl ChatComposer {
service_tier_commands_enabled: self.service_tier_commands_enabled,
goal_command_enabled: self.goal_command_enabled,
personality_command_enabled: self.personality_command_enabled,
voice_command_enabled: self.voice_command_enabled,
worktrees_enabled: self.worktrees_enabled,
allow_elevate_sandbox: self.windows_degraded_sandbox_active,
side_conversation_active: self.side_conversation_active,
@@ -48,6 +48,7 @@ pub(crate) struct CommandPopupFlags {
pub(crate) service_tier_commands_enabled: bool,
pub(crate) goal_command_enabled: bool,
pub(crate) personality_command_enabled: bool,
pub(crate) voice_command_enabled: bool,
pub(crate) worktrees_enabled: bool,
pub(crate) windows_degraded_sandbox_active: bool,
pub(crate) side_conversation_active: bool,
@@ -63,6 +64,7 @@ impl From<CommandPopupFlags> for BuiltinCommandFlags {
service_tier_commands_enabled: value.service_tier_commands_enabled,
goal_command_enabled: value.goal_command_enabled,
personality_command_enabled: value.personality_command_enabled,
voice_command_enabled: value.voice_command_enabled,
worktrees_enabled: value.worktrees_enabled,
allow_elevate_sandbox: value.windows_degraded_sandbox_active,
side_conversation_active: value.side_conversation_active,
@@ -412,6 +414,31 @@ mod tests {
insta::assert_snapshot!("command_popup_app", format!("{buf:?}"));
}
#[cfg(target_os = "macos")]
#[test]
fn voice_command_popup_snapshot() {
let mut popup = CommandPopup::new(
CommandPopupFlags {
voice_command_enabled: true,
..CommandPopupFlags::default()
},
Vec::new(),
);
popup.on_composer_text_change("/voice".to_string());
let width = 72;
let area = Rect::new(
/*x*/ 0,
/*y*/ 0,
width,
popup.calculate_required_height(width),
);
let mut buf = Buffer::empty(area);
popup.render_ref(area, &mut buf);
insta::assert_snapshot!("command_popup_voice", format!("{buf:?}"));
}
#[cfg(target_os = "macos")]
#[test]
fn default_command_popup_items_snapshot() {
@@ -537,6 +564,7 @@ mod tests {
service_tier_commands_enabled: false,
goal_command_enabled: false,
personality_command_enabled: true,
voice_command_enabled: false,
worktrees_enabled: true,
windows_degraded_sandbox_active: false,
side_conversation_active: false,
@@ -565,6 +593,7 @@ mod tests {
service_tier_commands_enabled: false,
goal_command_enabled: false,
personality_command_enabled: false,
voice_command_enabled: false,
worktrees_enabled: false,
windows_degraded_sandbox_active: false,
side_conversation_active: false,
@@ -598,6 +627,7 @@ mod tests {
service_tier_commands_enabled: false,
goal_command_enabled: false,
personality_command_enabled: true,
voice_command_enabled: false,
worktrees_enabled: true,
windows_degraded_sandbox_active: false,
side_conversation_active: false,
+47 -2
View File
@@ -875,9 +875,17 @@ pub(crate) fn footer_hint_items_line(items: &[(String, String)]) -> Line<'static
for (idx, (key, label)) in items.iter().enumerate() {
spans.push(" ".into());
spans.push(key.clone().bold());
spans.push(format!(" {label}").into());
if idx == 0
&& key == "voice"
&& let Some(label) = label.strip_prefix("● ")
{
spans.push(" ●".red());
spans.push(format!(" {label}").into());
} else {
spans.push(format!(" {label}").into());
}
if idx + 1 != items.len() {
spans.push(" ".into());
spans.push((if items[0].0 == "voice" { " " } else { " " }).into());
}
}
Line::from(spans)
@@ -1313,6 +1321,43 @@ mod tests {
use ratatui::backend::Backend;
use ratatui::backend::TestBackend;
#[test]
fn voice_live_microphone_indicator_is_red() {
let line = footer_hint_items_line(&[("voice".into(), "● listen".into())]);
assert_eq!(line.spans[2].style.fg, Some(ratatui::style::Color::Red));
}
#[test]
fn voice_footer_rendering_preserves_text_and_styles() {
let items = [
("voice".into(), "● listen".into()),
("ctrl+m".into(), "mute".into()),
];
let mut terminal =
Terminal::new(TestBackend::new(/*width*/ 40, /*height*/ 1)).expect("create terminal");
terminal
.draw(|frame| render_footer_hint_items(frame.area(), frame.buffer_mut(), &items))
.expect("render voice footer");
let backend = terminal.backend();
let mut previous_style = None;
let style_runs = (0..40)
.filter_map(|x| {
let cell = backend.buffer().cell((x, 0))?;
let style = cell.style();
if previous_style == Some(style) {
None
} else {
previous_style = Some(style);
Some((x, style))
}
})
.collect::<Vec<_>>();
insta::assert_debug_snapshot!(
"voice_footer_rendered_styles",
(backend.to_string(), style_runs)
);
}
fn snapshot_footer(name: &str, props: FooterProps) {
snapshot_footer_with_mode_indicator(
name, /*width*/ 80, &props, /*collaboration_mode_indicator*/ None,
+5
View File
@@ -546,6 +546,11 @@ impl BottomPane {
self.request_redraw();
}
pub fn set_voice_command_enabled(&mut self, enabled: bool) {
self.composer.set_voice_command_enabled(enabled);
self.request_redraw();
}
pub(crate) fn set_side_conversation_active(&mut self, active: bool) {
self.composer.set_side_conversation_active(active);
self.request_redraw();
@@ -62,6 +62,7 @@ pub(crate) struct BuiltinCommandFlags {
pub(crate) service_tier_commands_enabled: bool,
pub(crate) goal_command_enabled: bool,
pub(crate) personality_command_enabled: bool,
pub(crate) voice_command_enabled: bool,
pub(crate) worktrees_enabled: bool,
pub(crate) allow_elevate_sandbox: bool,
pub(crate) side_conversation_active: bool,
@@ -79,6 +80,7 @@ pub(crate) fn builtins_for_input(flags: BuiltinCommandFlags) -> Vec<(&'static st
.filter(|(_, cmd)| flags.goal_command_enabled || *cmd != SlashCommand::Goal)
.filter(|(_, cmd)| flags.worktrees_enabled || *cmd != SlashCommand::Worktree)
.filter(|(_, cmd)| flags.personality_command_enabled || *cmd != SlashCommand::Personality)
.filter(|(_, cmd)| flags.voice_command_enabled || *cmd != SlashCommand::Voice)
.filter(|(_, cmd)| !flags.side_conversation_active || cmd.available_in_side_conversation())
.collect()
}
@@ -173,6 +175,7 @@ mod tests {
service_tier_commands_enabled: true,
goal_command_enabled: true,
personality_command_enabled: true,
voice_command_enabled: true,
worktrees_enabled: true,
allow_elevate_sandbox: true,
side_conversation_active: false,
@@ -282,6 +285,13 @@ mod tests {
assert_eq!(find_builtin_command("goal", flags), None);
}
#[test]
fn voice_command_is_hidden_when_disabled() {
let mut flags = all_enabled_flags();
flags.voice_command_enabled = false;
assert_eq!(find_builtin_command("voice", flags), None);
}
#[test]
fn usage_command_is_hidden_from_input_when_account_token_activity_is_disabled() {
let mut flags = all_enabled_flags();
@@ -0,0 +1,15 @@
---
source: tui/src/bottom_pane/command_popup.rs
expression: "format!(\"{buf:?}\")"
---
Buffer {
area: Rect { x: 0, y: 0, width: 72, height: 1 },
content: [
" /voice start or stop a live voice conversation ",
],
styles: [
x: 0, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
x: 2, y: 0, fg: Cyan, bg: Reset, underline: Reset, modifier: BOLD,
x: 49, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: NONE,
]
}
@@ -0,0 +1,34 @@
---
source: tui/src/bottom_pane/footer.rs
assertion_line: 1355
expression: "(backend.to_string(), style_runs)"
---
(
"\" voice ● listen ctrl+m mute \"\n",
[
(
0,
Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset),
),
(
3,
Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold(),
),
(
8,
Style::new().red().bg(Color::Reset).underline_color(Color::Reset),
),
(
10,
Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset),
),
(
20,
Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold(),
),
(
26,
Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset),
),
],
)
+48 -1
View File
@@ -357,6 +357,7 @@ mod mcp_startup;
use self::mcp_startup::McpStartupStatus;
mod misalignment_policy;
pub(crate) use misalignment_policy::MisalignmentReview;
pub(crate) use misalignment_policy::MisalignmentTurnSource;
mod pets;
mod session_flow;
mod session_header;
@@ -406,7 +407,17 @@ mod recap;
mod reset_credits;
pub(crate) use self::rate_limits::limit_label_for_window;
mod completion;
mod realtime;
pub(crate) use realtime::MAX_REPLAY_TRANSCRIPT_CELLS;
pub(crate) use realtime::MAX_TRANSCRIPT_BYTES;
pub(crate) use realtime::RealtimeTranscriptRecord;
pub(crate) use realtime::is_private_realtime_agent_item;
pub(crate) use realtime::realtime_delegation_display_text;
pub(crate) use realtime::realtime_delegation_input;
#[cfg(test)]
pub(crate) use realtime::tests::activate_voice_for_thread;
mod reasoning_shortcuts;
use self::realtime::RealtimeConversationUiState;
mod rendering;
mod replay;
mod review;
@@ -631,6 +642,8 @@ pub(crate) struct ChatWidget {
last_unified_wait: Option<UnifiedExecWaitState>,
unified_exec_wait_streak: Option<UnifiedExecWaitStreak>,
turn_lifecycle: TurnLifecycleState,
realtime_conversation: RealtimeConversationUiState,
realtime_conversation_available_for_thread: bool,
safety_buffering: SafetyBufferingState,
task_complete_pending: bool,
unified_exec_processes: Vec<UnifiedExecProcessSummary>,
@@ -1179,6 +1192,8 @@ impl ChatWidget {
self.update_due_hook_visibility();
self.schedule_hook_timer_if_needed();
self.bottom_pane.pre_draw_tick();
self.flush_realtime_transcript_history();
self.refresh_realtime_microphone_level();
if let Some(pet) = self.ambient_pet.as_ref() {
pet.schedule_next_frame();
}
@@ -1295,7 +1310,20 @@ impl ChatWidget {
items: &[UserInput],
client_id: Option<&str>,
from_replay: bool,
turn_id: &str,
) {
if let Some(input) = realtime::realtime_delegation_input(items) {
if !from_replay && self.should_hide_realtime_delegation(turn_id) {
return;
}
let input = realtime::realtime_delegation_display_text(input);
let projected = [UserInput::Text {
text: input,
text_elements: Vec::new(),
}];
self.on_committed_user_message(&projected, client_id, from_replay, turn_id);
return;
}
let display = Self::user_message_display_from_inputs(items);
if from_replay {
if self.review.is_review_mode {
@@ -1920,9 +1948,14 @@ impl ChatWidget {
/// the main viewport updates.
pub(crate) fn active_cell_transcript_key(&self) -> Option<ActiveCellTranscriptKey> {
let cell = self.transcript.active_cell.as_ref();
let realtime_cell = self.realtime_conversation.live_transcript_cell.as_ref();
let token_activity_cell = self.pending_token_activity_output();
let rate_limit_reset_hint = self.pending_rate_limit_reset_hint();
if cell.is_none() && token_activity_cell.is_none() && rate_limit_reset_hint.is_none() {
if cell.is_none()
&& realtime_cell.is_none()
&& token_activity_cell.is_none()
&& rate_limit_reset_hint.is_none()
{
return None;
}
Some(ActiveCellTranscriptKey {
@@ -1948,6 +1981,13 @@ impl ChatWidget {
if let Some(cell) = self.transcript.active_cell.as_ref() {
lines.extend(cell.transcript_hyperlink_lines(width));
}
if let Some(cell) = self.realtime_conversation.live_transcript_cell.as_ref() {
let realtime_lines = cell.transcript_hyperlink_lines(width);
if !realtime_lines.is_empty() && !lines.is_empty() {
lines.push(HyperlinkLine::from(""));
}
lines.extend(realtime_lines);
}
if let Some(token_activity_cell) = self.pending_token_activity_output() {
let token_activity_lines = token_activity_cell.transcript_hyperlink_lines(width);
if !token_activity_lines.is_empty() && !lines.is_empty() {
@@ -1998,6 +2038,13 @@ fn has_websocket_timing_metrics(summary: RuntimeMetricsSummary) -> bool {
impl Drop for ChatWidget {
fn drop(&mut self) {
if self.realtime_conversation.handle.is_some()
&& let Some(thread_id) = self.thread_id
{
self.app_event_tx
.send(AppEvent::StopRealtimeConversation { thread_id });
}
self.reset_realtime_conversation();
self.stop_rate_limit_poller();
}
}
@@ -175,6 +175,8 @@ impl ChatWidget {
last_unified_wait: None,
unified_exec_wait_streak: None,
turn_lifecycle: TurnLifecycleState::new(prevent_idle_sleep),
realtime_conversation: RealtimeConversationUiState::default(),
realtime_conversation_available_for_thread: false,
safety_buffering: SafetyBufferingState::default(),
task_complete_pending: false,
unified_exec_processes: Vec::new(),
@@ -291,6 +293,9 @@ impl ChatWidget {
widget.sync_worktrees_enabled();
widget.sync_plugins_command_enabled();
widget.sync_goal_command_enabled();
widget
.bottom_pane
.set_voice_command_enabled(/*enabled*/ false);
widget.sync_mentions_v2_enabled();
widget
.bottom_pane
@@ -410,6 +410,7 @@ impl ChatWidget {
return (false, None);
}
self.dismiss_backend_banner_for_new_turn();
self.note_realtime_typed_input(&submitted_message.text);
if render_in_history {
self.input_queue.user_turn_pending_start = true;
}
@@ -26,6 +26,13 @@ impl MisalignmentReview {
pub(super) struct MisalignmentViolation {
turn_id: Option<String>,
review: Option<Arc<MisalignmentReview>>,
retired_voice: bool,
}
#[derive(Clone, Copy, Eq, PartialEq)]
pub(crate) enum MisalignmentTurnSource {
ServerNotification,
AcknowledgedContinuation,
}
const MISALIGNMENT_POLICY_TITLE: &str = "Chat stopped as a precaution";
@@ -41,6 +48,7 @@ impl ChatWidget {
&& !matches!(
op,
AppCommand::Interrupt
| AppCommand::RealtimeConversationStop { .. }
| AppCommand::CleanBackgroundTerminals
| AppCommand::OverrideTurnContext { .. }
| AppCommand::ReloadUserConfig
@@ -80,6 +88,10 @@ impl ChatWidget {
{
return;
}
let retired_voice = self.realtime_conversation_is_running();
if retired_voice {
self.stop_realtime_conversation();
}
let review = self.thread_id.zip(turn_id.clone()).zip(details).map(
|((thread_id, turn_id), details)| {
Arc::new(MisalignmentReview {
@@ -89,7 +101,11 @@ impl ChatWidget {
})
},
);
self.misalignment_policy_violation = Some(MisalignmentViolation { turn_id, review });
self.misalignment_policy_violation = Some(MisalignmentViolation {
turn_id,
review,
retired_voice,
});
self.input_queue.clear();
self.finalize_turn();
self.refresh_pending_input_preview();
@@ -211,11 +227,19 @@ impl ChatWidget {
});
}
pub(crate) fn clear_misalignment_for_new_turn(&mut self, turn_id: &str) {
pub(crate) fn clear_misalignment_for_new_turn(
&mut self,
turn_id: &str,
source: MisalignmentTurnSource,
) {
if self
.misalignment_policy_violation
.as_ref()
.is_some_and(|violation| violation.turn_id.as_ref().is_some_and(|id| id != turn_id))
.is_some_and(|violation| {
violation.turn_id.as_ref().is_some_and(|id| id != turn_id)
&& (!violation.retired_voice
|| source == MisalignmentTurnSource::AcknowledgedContinuation)
})
{
self.misalignment_policy_violation = None;
self.bottom_pane.dismiss_view_by_id(PRECAUTION_VIEW);
+116 -15
View File
@@ -61,7 +61,10 @@ impl ChatWidget {
}
ServerNotification::TurnStarted(notification) => {
if replay_kind.is_none() {
self.clear_misalignment_for_new_turn(&notification.turn.id);
self.clear_misalignment_for_new_turn(
&notification.turn.id,
MisalignmentTurnSource::ServerNotification,
);
}
self.turn_lifecycle.last_turn_id = Some(notification.turn.id);
self.last_non_retry_error = None;
@@ -80,18 +83,43 @@ impl ChatWidget {
self.handle_item_completed_notification(notification, replay_kind);
}
ServerNotification::AgentMessageDelta(notification) => {
self.on_agent_message_delta(notification.delta);
if !self.is_realtime_delegated_reasoning_turn(&notification.turn_id)
&& (from_replay
|| !self.is_realtime_delegated_agent_item(
&notification.turn_id,
&notification.item_id,
))
{
self.on_agent_message_delta(notification.delta);
}
}
ServerNotification::PlanDelta(notification) => self.on_plan_delta(notification.delta),
ServerNotification::ReasoningSummaryTextDelta(notification) => {
self.on_agent_reasoning_delta(notification.delta);
}
ServerNotification::ReasoningTextDelta(notification) => {
if self.config.show_raw_agent_reasoning {
if !self.is_realtime_delegated_reasoning_item(
&notification.turn_id,
&notification.item_id,
) {
self.on_agent_reasoning_delta(notification.delta);
}
}
ServerNotification::ReasoningSummaryPartAdded(_) => self.on_reasoning_section_break(),
ServerNotification::ReasoningTextDelta(notification) => {
if self.config.show_raw_agent_reasoning
&& !self.is_realtime_delegated_reasoning_item(
&notification.turn_id,
&notification.item_id,
)
{
self.on_agent_reasoning_delta(notification.delta);
}
}
ServerNotification::ReasoningSummaryPartAdded(notification) => {
if !self.is_realtime_delegated_reasoning_item(
&notification.turn_id,
&notification.item_id,
) {
self.on_reasoning_section_break();
}
}
ServerNotification::TerminalInteraction(notification) => {
self.on_terminal_interaction(notification.process_id, notification.stdin)
}
@@ -235,6 +263,36 @@ impl ChatWidget {
self.on_shutdown_complete();
}
}
ServerNotification::ThreadRealtimeSdp(notification) => {
if !from_replay {
self.on_realtime_conversation_sdp(notification.sdp);
}
}
ServerNotification::ThreadRealtimeStarted(_) => {
if !from_replay {
self.on_realtime_conversation_started();
}
}
ServerNotification::ThreadRealtimeTranscriptDelta(notification) => {
if !from_replay {
self.on_realtime_transcript_delta(notification.role, notification.delta);
}
}
ServerNotification::ThreadRealtimeTranscriptDone(notification) => {
if !from_replay {
self.on_realtime_transcript_done(notification.role, notification.text);
}
}
ServerNotification::ThreadRealtimeError(notification) => {
if !from_replay {
self.on_realtime_error(notification.message);
}
}
ServerNotification::ThreadRealtimeClosed(notification) => {
if !from_replay {
self.on_realtime_conversation_closed(notification.reason);
}
}
ServerNotification::ServerRequestResolved(_)
| ServerNotification::AccountUpdated(_)
| ServerNotification::AccountRateLimitsUpdated(_)
@@ -264,17 +322,11 @@ impl ChatWidget {
| ServerNotification::TurnModerationMetadata(_)
| ServerNotification::FuzzyFileSearchSessionUpdated(_)
| ServerNotification::FuzzyFileSearchSessionCompleted(_)
| ServerNotification::ThreadRealtimeStarted(_)
| ServerNotification::ThreadRealtimeItemAdded(_)
| ServerNotification::ThreadRealtimeItemStarted(_)
| ServerNotification::ThreadRealtimeItemTranscriptDelta(_)
| ServerNotification::ThreadRealtimeItemCompleted(_)
| ServerNotification::ThreadRealtimeOutputAudioDelta(_)
| ServerNotification::ThreadRealtimeError(_)
| ServerNotification::ThreadRealtimeClosed(_)
| ServerNotification::ThreadRealtimeSdp(_)
| ServerNotification::ThreadRealtimeTranscriptDelta(_)
| ServerNotification::ThreadRealtimeTranscriptDone(_)
| ServerNotification::WindowsWorldWritableWarning(_)
| ServerNotification::WindowsSandboxSetupCompleted(_)
| ServerNotification::AccountLoginCompleted(_)
@@ -310,7 +362,12 @@ impl ChatWidget {
text,
phase: Some(MessagePhase::FinalAnswer) | None,
..
} => Some((item.clone(), id.clone(), text.clone())),
} if !self
.is_realtime_delegated_reasoning_turn(&notification.turn.id)
|| !realtime::is_private_realtime_agent_item(item) =>
{
Some((item.clone(), id.clone(), text.clone()))
}
_ => None,
});
if let Some((item, id, _)) = &last_agent_message
@@ -329,6 +386,11 @@ impl ChatWidget {
.map_or(ThreadItemRenderSource::Live, ThreadItemRenderSource::Replay),
);
}
if replay_kind.is_none()
&& let Some((item, _, _)) = &last_agent_message
{
self.speak_completed_realtime_delegation(&notification.turn.id, item);
}
self.last_non_retry_error = None;
let completion = self.completion_cell(&notification.turn, replay_kind);
self.on_task_complete(
@@ -355,7 +417,10 @@ impl ChatWidget {
&& error.codex_error_info
== Some(AppServerCodexErrorInfo::MisalignmentPolicyViolation)
{
self.on_misalignment_error(Some(notification.turn.id), error.misalignment);
self.on_misalignment_error(
Some(notification.turn.id.clone()),
error.misalignment,
);
self.last_non_retry_error = None;
} else if self.last_non_retry_error.as_ref()
== Some(&(notification.turn.id.clone(), error.message.clone()))
@@ -373,6 +438,9 @@ impl ChatWidget {
}
TurnStatus::InProgress => {}
}
if replay_kind.is_none() {
self.finish_realtime_turn(&notification.turn.id);
}
self.thread_usage.replaying_turn_completion = was_replaying_turn_completion;
}
@@ -382,6 +450,28 @@ impl ChatWidget {
replay_kind: Option<ReplayKind>,
) {
match notification.item {
ThreadItem::UserMessage { content, .. } if replay_kind.is_none() => {
self.note_realtime_user_item_started(&notification.turn_id, &content);
}
ThreadItem::UserMessage { content, .. }
if realtime::realtime_delegation_input(&content).is_some() =>
{
self.remember_realtime_delegated_reasoning_turn(&notification.turn_id);
}
ThreadItem::AgentMessage { id, .. } if replay_kind.is_none() => {
self.is_realtime_delegated_agent_item(&notification.turn_id, &id);
}
ThreadItem::Reasoning { id, .. }
if replay_kind.is_none()
&& !self.is_realtime_delegated_reasoning_turn(&notification.turn_id) =>
{
// A later voice handoff can steer this turn without making an
// already-started typed reasoning item private.
self.realtime_conversation
.agent_items
.entry((notification.turn_id.clone(), id))
.or_insert(realtime::RealtimeAgentItemOrigin::Typed);
}
ThreadItem::ContextCompaction { id }
if !matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages)) =>
{
@@ -441,6 +531,17 @@ impl ChatWidget {
notification: ItemCompletedNotification,
replay_kind: Option<ReplayKind>,
) {
if replay_kind.is_none()
&& self.is_realtime_delegated_reasoning_turn(&notification.turn_id)
&& realtime::is_private_realtime_agent_item(&notification.item)
&& !matches!(&notification.item, ThreadItem::AgentMessage { id, .. } | ThreadItem::Reasoning { id, .. }
if matches!(
self.realtime_conversation.agent_items.get(&(notification.turn_id.clone(), id.clone())),
Some(realtime::RealtimeAgentItemOrigin::Typed)
))
{
return;
}
// Buffered live notifications can introduce questions; historical turn replay cannot.
if replay_kind == Some(ReplayKind::ThreadSnapshot)
&& let ThreadItem::AgentMessage {
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,107 @@
//! Focused visual coverage and shared setup for voice behavior tests.
use super::RealtimeConversationPhase;
use crate::app_event::AppEvent;
use crate::chatwidget::ChatWidget;
use crate::chatwidget::tests::make_chatwidget_manual_with_sender;
use crate::chatwidget::tests::render_bottom_popup;
use codex_protocol::ThreadId;
pub(crate) fn activate_voice_for_thread(chat: &mut ChatWidget, thread_id: ThreadId) {
chat.thread_id = Some(thread_id);
chat.realtime_conversation.phase = RealtimeConversationPhase::Active;
chat.realtime_conversation.thread_id = Some(thread_id);
chat.realtime_conversation.backend_started = true;
chat.realtime_conversation.latest_input_was_voice = true;
}
#[tokio::test]
async fn enabling_voice_on_an_open_thread_snapshots_the_new_thread_notice() {
let (mut chat, _sender, mut events, _ops) = make_chatwidget_manual_with_sender().await;
chat.set_feature_enabled(
codex_features::Feature::RealtimeConversation,
/*enabled*/ true,
);
let rendered = std::iter::from_fn(|| events.try_recv().ok())
.filter_map(|event| match event {
AppEvent::InsertHistoryCell(cell) => Some(
cell.display_lines(/*width*/ 80)
.into_iter()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n"),
),
_ => None,
})
.collect::<Vec<_>>()
.join("\n");
insta::assert_snapshot!(rendered, @"• Voice conversations will be available in new threads.");
}
#[tokio::test]
async fn voice_footer_renders_the_main_conversation_states() {
let (mut chat, _sender, _events, _ops) = make_chatwidget_manual_with_sender().await;
let mut states = Vec::new();
for (label, phase, muted, level, speaker_level, role, transcript) in [
(
"connecting",
RealtimeConversationPhase::Starting,
false,
0,
0,
None,
"",
),
(
"listening",
RealtimeConversationPhase::Active,
false,
0,
0,
None,
"",
),
(
"speaking",
RealtimeConversationPhase::Active,
false,
4,
5,
None,
"",
),
(
"muted",
RealtimeConversationPhase::Active,
true,
4,
4,
None,
"",
),
(
"transcript",
RealtimeConversationPhase::Active,
false,
2,
2,
Some("assistant"),
"Hello there",
),
] {
chat.realtime_conversation.phase = phase;
chat.realtime_conversation.microphone_muted = muted;
chat.realtime_conversation.microphone_level = level;
chat.realtime_conversation.speaker_level = speaker_level;
chat.realtime_conversation.transcript_role = role.map(str::to_string);
chat.realtime_conversation.transcript = transcript.to_string();
chat.update_realtime_footer();
states.push(format!(
"{label}:\n{}",
render_bottom_popup(&chat, /*width*/ 80)
));
}
insta::assert_snapshot!(states.join("\n\n"));
}
+3
View File
@@ -4,6 +4,9 @@ use super::*;
impl ChatWidget {
pub(crate) fn pause_for_disconnect(&mut self) {
// The app-server transport can fail while the separate WebRTC helper
// still sends microphone audio. Retire local media before showing offline UI.
let _ = self.reset_realtime_conversation();
if let Some(questions) = &mut self.bottom_pane.questions {
questions.delivery_enabled = false;
}
+11
View File
@@ -158,6 +158,17 @@ impl ChatWidget {
};
let mut flex = FlexRenderable::new();
flex.push(/*flex*/ 1, active_cell_renderable);
if let Some(cell) = self.realtime_conversation.live_transcript_cell.as_ref() {
flex.push(
/*flex*/ 1,
RenderableItem::Owned(Box::new(TranscriptAreaRenderable {
child: cell.as_ref(),
top: 1,
right: active_cell_right_reserve,
persistent_layout: None,
})),
);
}
if let Some(cell) = self.pending_token_activity_output() {
flex.push(
/*flex*/ 1,
+38 -1
View File
@@ -46,13 +46,31 @@ impl ChatWidget {
completed_at,
duration_ms,
} = turn;
let delegated = items.iter().any(|item| {
matches!(item, ThreadItem::UserMessage { content, .. }
if realtime::realtime_delegation_input(content).is_some())
});
if matches!(status, TurnStatus::InProgress) {
if delegated {
self.remember_realtime_delegated_reasoning_turn(&turn_id);
}
self.warning_display_state.startup_complete = true;
self.turn_lifecycle.last_turn_id = Some(turn_id.clone());
self.last_non_retry_error = None;
self.on_task_started();
}
let mut replaying_delegation = false;
for item in items {
if matches!(&item, ThreadItem::UserMessage { content, .. }
if realtime::realtime_delegation_input(content).is_some())
{
replaying_delegation = true;
}
// Voice can steer a typed turn already in progress. Its earlier
// commentary and reasoning still belong to the typed request.
if replaying_delegation && realtime::is_private_realtime_agent_item(&item) {
continue;
}
if hidden_nested_review_turn && matches!(item, ThreadItem::UserMessage { .. }) {
continue;
}
@@ -122,7 +140,12 @@ impl ChatWidget {
ThreadItem::UserMessage {
content, client_id, ..
} => {
self.on_committed_user_message(&content, client_id.as_deref(), from_replay);
self.on_committed_user_message(
&content,
client_id.as_deref(),
from_replay,
&turn_id,
);
}
ThreadItem::AgentMessage {
id,
@@ -133,6 +156,20 @@ impl ChatWidget {
questions,
..
} => {
if self.complete_realtime_delegated_agent_item(
&turn_id,
&ThreadItem::AgentMessage {
id: id.clone(),
text: text.clone(),
phase: phase.clone(),
memory_citation: memory_citation.clone(),
delivery,
questions: questions.clone(),
},
from_replay,
) {
return;
}
self.on_agent_message_item_completed(
AgentMessageItem {
id,
@@ -24,6 +24,11 @@ impl ChatWidget {
let connector_scope_changed = previous_thread_id != Some(session.thread_id)
|| self.config.cwd.as_path() != session.cwd.as_path();
self.thread_id = Some(session.thread_id);
self.realtime_conversation_available_for_thread =
self.config.features.enabled(Feature::RealtimeConversation)
&& codex_realtime_webrtc::RealtimeWebrtcSession::is_supported();
self.bottom_pane
.set_voice_command_enabled(self.realtime_conversation_available_for_thread);
self.bottom_pane
.set_queue_submissions(/*queue_submissions*/ false);
if previous_thread_id != self.thread_id {
+15
View File
@@ -100,6 +100,21 @@ impl ChatWidget {
self.update_collaboration_mode_indicator();
}
}
if feature == Feature::RealtimeConversation && !enabled {
self.realtime_conversation_available_for_thread = false;
self.bottom_pane
.set_voice_command_enabled(/*enabled*/ false);
self.stop_realtime_conversation();
}
if feature == Feature::RealtimeConversation
&& enabled
&& !self.realtime_conversation_available_for_thread
{
self.add_info_message(
"Voice conversations will be available in new threads.".into(),
/*hint*/ None,
);
}
if feature == Feature::MentionsV2 {
self.sync_mentions_v2_enabled();
}
@@ -333,6 +333,9 @@ impl ChatWidget {
);
}
}
SlashCommand::Voice => {
self.toggle_realtime_conversation();
}
SlashCommand::Side | SlashCommand::Btw => {
self.request_empty_side_conversation(cmd);
}
@@ -757,6 +760,11 @@ impl ChatWidget {
}
}
}
SlashCommand::Voice => match trimmed.to_ascii_lowercase().as_str() {
"mute" => self.toggle_realtime_microphone(),
"stop" => self.stop_realtime_conversation(),
_ => self.add_error_message("Usage: /voice [mute|stop]".to_string()),
},
SlashCommand::Ide => {
self.handle_ide_command_args(trimmed);
}
@@ -1144,6 +1152,7 @@ impl ChatWidget {
goal_command_enabled: self.config.features.enabled(Feature::Goals),
service_tier_commands_enabled: self.fast_mode_enabled(),
personality_command_enabled: self.config.features.enabled(Feature::Personality),
voice_command_enabled: self.realtime_conversation_available_for_thread,
worktrees_enabled: self.config.features.enabled(Feature::Worktrees)
&& self.local_worktree_operations,
allow_elevate_sandbox,
@@ -1183,6 +1192,7 @@ impl ChatWidget {
| SlashCommand::Diff
| SlashCommand::App
| SlashCommand::Rename
| SlashCommand::Voice
| SlashCommand::Recap
| SlashCommand::TestApproval => QueueDrain::Continue,
SlashCommand::Cd => match self.thread_id {
@@ -0,0 +1,29 @@
---
source: tui/src/chatwidget/realtime_tests.rs
assertion_line: 176
expression: "states.join(\"\\n\\n\")"
---
connecting:
› Ask Codex to do anything
voice ◌ connecting /voice stop /voice mute
listening:
› Ask Codex to do anything
voice ● listening /voice stop mic ░░░░░ codex ░░░░░ /voice mute
speaking:
› Ask Codex to do anything
voice ● speaking /voice stop mic ████░ codex █████ /voice mute
muted:
› Ask Codex to do anything
voice ◌ muted /voice stop mic off codex ████░ /voice mute
transcript:
› Ask Codex to do anything
voice ● speaking /voice stop mic ██░░░ codex ██░░░ /voice mute
@@ -0,0 +1,16 @@
---
source: tui/src/chatwidget/tests/popups_and_settings.rs
assertion_line: 3142
expression: popup
---
Experimental features
Checked features are configured on. Some experimental features take effect
only in new tasks or after restarting the Codex server.
› [ ] Network proxy Apply network proxy restrictions to
sandboxed sessions that already have
network access.
[ ] Prevent sleep while running Keep your computer awake while Codex is
running a thread.
Press space to select or enter to save
@@ -0,0 +1,16 @@
---
source: tui/src/chatwidget/tests/status_and_layout.rs
assertion_line: 34
expression: rendered
---
" "
"• Agent answer arriving "
" "
" "
"› pick a number "
" "
" "
" "
"› Ask Codex to do anything "
" "
" voice ● listening /voice stop mic ░░░░░ codex ░░░░░ "
@@ -1,8 +1,51 @@
use super::*;
use crate::chatwidget::realtime::tests::activate_voice_for_thread;
use pretty_assertions::assert_eq;
const ERROR_MESSAGE: &str = "Responses API returned misalignment_policy_violation";
#[tokio::test]
async fn misalignment_precaution_retires_voice_and_keeps_late_turn_blocked() {
let (mut chat, _events, mut ops) = make_chatwidget_manual(/*model_override*/ None).await;
let thread_id = ThreadId::new();
chat.thread_id = Some(thread_id);
handle_turn_started(&mut chat, "unsafe-turn");
activate_voice_for_thread(&mut chat, thread_id);
handle_error(
&mut chat,
ERROR_MESSAGE,
Some(CodexErrorInfo::MisalignmentPolicyViolation),
);
assert!(!chat.realtime_conversation_is_running());
assert_matches!(
ops.try_recv(),
Ok(Op::RealtimeConversationStop { thread_id: stopped }) if stopped == thread_id
);
assert!(ops.try_recv().is_err());
assert!(chat.has_misalignment_policy_violation());
assert!(!chat.bottom_pane.composer_input_enabled());
assert!(chat.submit_op(Op::RealtimeConversationStop { thread_id }));
assert_matches!(ops.try_recv(), Ok(Op::RealtimeConversationStop { .. }));
// A handoff already in flight may still produce a TurnStarted notification.
// It must not dismiss the precaution or restart local capture.
handle_turn_started(&mut chat, "late-voice-turn");
assert!(chat.has_misalignment_policy_violation());
assert!(!chat.bottom_pane.composer_input_enabled());
chat.toggle_realtime_conversation();
assert!(!chat.realtime_conversation_is_running());
assert!(ops.try_recv().is_err());
chat.clear_misalignment_for_new_turn(
"acknowledged-turn",
MisalignmentTurnSource::AcknowledgedContinuation,
);
assert!(!chat.has_misalignment_policy_violation());
assert!(chat.bottom_pane.composer_input_enabled());
}
#[tokio::test]
async fn misalignment_policy_failure_stops_the_thread_and_renders_once() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
@@ -3175,6 +3175,35 @@ async fn experimental_popup_loading_snapshot() {
assert!(rx.try_recv().is_err());
}
#[tokio::test]
async fn experimental_popup_available_snapshot() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.open_experimental_popup();
let AppEvent::FetchExperimentalFeatures { response_tx, .. } = rx.try_recv().unwrap() else {
panic!("expected experimental discovery request");
};
let features = [
("network_proxy", "Network proxy", "Apply network proxy restrictions to sandboxed sessions that already have network access."),
("prevent_idle_sleep", "Prevent sleep while running", "Keep your computer awake while Codex is running a thread."),
]
.into_iter()
.map(|(name, display_name, description)| codex_app_server_protocol::ExperimentalFeature {
name: name.to_string(),
stage: codex_app_server_protocol::ExperimentalFeatureStage::Beta,
display_name: Some(display_name.to_string()),
description: Some(description.to_string()),
announcement: None,
enabled: false,
default_enabled: false,
})
.collect();
response_tx.send(Ok(features)).unwrap();
chat.pre_draw_tick();
let popup = render_bottom_popup(&chat, /*width*/ 80);
assert_chatwidget_snapshot!("experimental_features_available_popup", popup);
}
#[tokio::test]
async fn feature_enable_prompts_snapshot() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
@@ -3,12 +3,37 @@ use crate::bottom_pane::goal_status_indicator_line;
use crate::chatwidget::ThreadUsageOutcome;
use crate::chatwidget::rate_limits::NUDGE_MODEL_SLUG;
use crate::chatwidget::rate_limits::get_limits_duration;
use crate::chatwidget::realtime::tests::activate_voice_for_thread;
use codex_app_server_protocol::SpendControlLimitSnapshot;
use codex_app_server_protocol::ThreadUsage;
use pretty_assertions::assert_eq;
use ratatui::Terminal;
use ratatui::backend::TestBackend;
use serial_test::serial;
#[tokio::test]
async fn voice_live_transcript_renders_beside_the_streamed_cell() {
let (mut chat, _rx, _ops) = make_chatwidget_manual(/*model_override*/ None).await;
chat.config.animations = false;
activate_voice_for_thread(&mut chat, ThreadId::new());
chat.transcript.active_cell = Some(Box::new(history_cell::StreamingAgentTailCell::new(
vec![Line::from("Agent answer arriving").into()],
/*is_first_line*/ true,
)));
chat.on_realtime_transcript_delta("user".into(), "pick a number".into());
let width = 60;
let height = chat.desired_height(width);
let mut terminal = Terminal::new(TestBackend::new(width, height)).expect("create terminal");
terminal
.draw(|frame| chat.render(frame.area(), frame.buffer_mut()))
.expect("render live voice transcript");
let rendered = normalized_backend_snapshot(terminal.backend());
assert!(rendered.contains("Agent answer arriving"), "{rendered}");
assert!(rendered.contains("pick a number"), "{rendered}");
assert_chatwidget_snapshot!("voice_live_transcript_and_stream", rendered);
}
fn enable_test_ambient_pet(chat: &mut ChatWidget) {
chat.set_pet_image_support_for_tests(crate::pets::PetImageSupport::Supported(
crate::pets::ImageProtocol::Kitty,
+8 -1
View File
@@ -208,7 +208,14 @@ impl ChatWidget {
.current_goal_status
.as_ref()
.is_some_and(GoalStatusState::is_active);
if !follow_up_started && !active_goal_continuing {
if !follow_up_started
&& !active_goal_continuing
&& !self
.turn_lifecycle
.last_turn_id
.as_deref()
.is_some_and(|turn_id| self.should_hide_realtime_delegation(turn_id))
{
self.notify(Notification::AgentTurnComplete {
response: notification_response,
});
+5
View File
@@ -41,6 +41,7 @@ pub enum SlashCommand {
Compact,
Recap,
Plan,
Voice,
Goal,
Agents,
Side,
@@ -130,6 +131,7 @@ impl SlashCommand {
}
SlashCommand::Personality => "choose a communication style for Codex",
SlashCommand::Plan => "switch to Plan mode",
SlashCommand::Voice => "start or stop a live voice conversation",
SlashCommand::Goal => "set or view the goal for a long-running task",
SlashCommand::Agents => "view and switch between all active agent sessions",
SlashCommand::MultiAgents => "switch between this session's subagents",
@@ -172,6 +174,7 @@ impl SlashCommand {
| SlashCommand::Fork
| SlashCommand::Plan
| SlashCommand::Goal
| SlashCommand::Voice
| SlashCommand::Ide
| SlashCommand::Keymap
| SlashCommand::Mcp
@@ -250,6 +253,7 @@ impl SlashCommand {
| SlashCommand::Stop
| SlashCommand::App
| SlashCommand::Goal
| SlashCommand::Voice
| SlashCommand::Mcp
| SlashCommand::Apps
| SlashCommand::Plugins
@@ -274,6 +278,7 @@ impl SlashCommand {
SlashCommand::SandboxReadRoot => cfg!(target_os = "windows"),
SlashCommand::Copy => !cfg!(target_os = "android"),
SlashCommand::App => cfg!(any(target_os = "macos", target_os = "windows")),
SlashCommand::Voice => true,
SlashCommand::Rollout | SlashCommand::TestApproval => cfg!(debug_assertions),
_ => true,
}
+8 -1
View File
@@ -176,9 +176,15 @@ fn run(
let Some(peer) = transport.as_ref() else {
return Err(io::Error::other("voice transport not started"));
};
executor
let outcome = executor
.block_on(peer.apply_answer(sdp.into_sdp()))
.map_err(io::Error::other)?;
if outcome == transport::AnswerOutcome::TimedOut {
// The client reaps this helper before considering a fresh negotiation.
output.write_all(&encode_frame(&Message::TransportTimedOut {})?)?;
output.flush()?;
return Ok(());
}
answered = true;
Message::TransportReady {}
}
@@ -207,6 +213,7 @@ fn run(
| Message::ApplyAnswer { .. }
| Message::Offer { .. }
| Message::TransportReady {}
| Message::TransportTimedOut {}
| Message::OpenDevices {}
| Message::DevicesOpened {}
| Message::AudioControlsApplied {}
+10 -2
View File
@@ -59,6 +59,12 @@ impl PeerConnectionEventHandler for Events {
}
}
#[derive(Debug, PartialEq)]
pub(crate) enum AnswerOutcome {
Ready,
TimedOut,
}
pub(crate) struct Transport {
pub(crate) incoming: crate::incoming::Incoming,
pub(crate) audio: crate::audio_track::AudioTrack,
@@ -162,7 +168,7 @@ impl Transport {
.map_err(|_| "timed out gathering voice candidates")?
}
pub(crate) async fn apply_answer(&self, sdp: String) -> Result<()> {
pub(crate) async fn apply_answer(&self, sdp: String) -> Result<AnswerOutcome> {
let answer = RTCSessionDescription::answer(sdp).map_err(|_| "invalid voice answer")?;
let parsed = answer.unmarshal().map_err(|_| "invalid voice answer")?;
let mut seen = HashSet::new();
@@ -213,7 +219,9 @@ impl Transport {
.map_err(|_| "voice event channel closed")
})
.await
.map_err(|_| "timed out connecting voice peer")?
.map_or(Ok(AnswerOutcome::TimedOut), |result| {
result.map(|()| AnswerOutcome::Ready)
})
}
pub(crate) async fn close(&mut self) -> Result<()> {
+2 -1
View File
@@ -86,10 +86,11 @@ async fn check_negotiation(runtime: Arc<dyn webrtc::runtime::Runtime>) {
for _ in candidates.len()..MAX_REMOTE_CANDIDATES {
answer.push_str(&format!("{}\r\n", candidates[0]));
}
local
let outcome = local
.apply_answer(answer)
.await
.unwrap_or_else(|error| panic!("tcp={tcp}: {error}"));
assert_eq!(outcome, super::AnswerOutcome::Ready);
let channel = channels.recv().await.unwrap();
assert_eq!(
(