agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Add realtime speech append control (#27917)

Browse Source 
## Why

Realtime voice harness tuning needs app-side control over what backend
Codex text is spoken. Backend orchestrator text is written for a reading
UI, so automatically speaking every preamble, progress update, or final
assistant message can make the realtime voice model too chatty.

For experimentation, clients need two simple controls: keep app/client
text-item injection on the existing item-create path, and add an
explicit speakable path that app code can call only when it wants
realtime to speak. Automatic Codex output also needs an opt-in way to
switch from the protocol's default speakable path to regular realtime
items, with a caller-provided prefix so prompt wording can be tuned
outside core.

The default remains unchanged: if a client omits the new start fields
and never calls `appendSpeech`, automatic backend output continues down
the existing speakable path for the selected realtime protocol.

## What Changed

- Adds experimental `thread/realtime/appendSpeech` for app-provided
speakable text.
- Keeps existing `thread/realtime/appendText` as the item-create API for
app-provided realtime text items.
- Adds `codexResponsesAsItems` / `codex_responses_as_items` on
`thread/realtime/start` to send automatic Codex responses with
`conversation.item.create` instead of the protocol's default speakable
output path.
- Adds `codexResponseItemPrefix` / `codex_response_item_prefix` so
clients can prepend experiment instructions to those automatic Codex
response items.
- Keeps literal `conversation.handoff.append` routing scoped to the v1
speakable path; v2 default speech uses its item/function-output plus
`response.create` behavior.
- Removes the earlier public silent-context API and hardcoded
silent-context prefix.
- Updates realtime tests to cover default automatic speakable behavior,
opt-in automatic item-create behavior, and explicit `appendSpeech`
behavior.

## Validation

- `cargo check -p codex-core -p codex-app-server -p codex-api`
- `just test -p codex-app-server realtime_conversation`
- `just test -p codex-core realtime_conversation` (50/51 passed in the
filtered parallel run; the lone failure passed when rerun in isolation)
- `just test -p codex-core
conversation_mirrors_assistant_message_text_to_realtime_handoff`
- `just test -p codex-api
e2e_connect_and_exchange_events_against_mock_ws_server`
- `just fix -p codex-core`
- `just fix -p codex-app-server`
- `cargo build -p codex-cli`
This commit is contained in:
guinness-oai
2026-06-15 16:15:58 -07:00
committed by GitHub
Unverified
parent 9728992fab
commit 1d8ff89aa3
15 changed files with 783 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/realtime_conversation.rs
+156 -67
View File
@@ -32,6 +32,7 @@ use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationSpeechParams;
use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::ConversationStartTransport;
use codex_protocol::protocol::ConversationTextParams;
@@ -103,25 +104,21 @@ enum RealtimeSessionKind {
#[derive(Clone, Debug)]
struct RealtimeHandoffState {
output_tx: Sender<HandoffOutput>,
output_tx: Sender<RealtimeOutbound>,
active_handoff: Arc<Mutex<Option<String>>>,
last_output_text: Arc<Mutex<Option<String>>>,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
session_kind: RealtimeSessionKind,
}
#[derive(Debug, PartialEq, Eq)]
enum HandoffOutput {
StandaloneAssistantOutput {
output_text: String,
},
ProgressUpdate {
handoff_id: String,
output_text: String,
},
FinalUpdate {
handoff_id: String,
output_text: String,
},
enum RealtimeOutbound {
StandaloneHandoff { text: String },
HandoffUpdate { handoff_id: String, text: String },
CompletedHandoff { handoff_id: String, text: String },
ConversationItem { text: String },
HandoffCompleteAck { handoff_id: String },
}
#[derive(Debug, PartialEq, Eq)]
@@ -196,7 +193,7 @@ struct RealtimeInputTask {
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
text_rx: Receiver<ConversationTextParams>,
handoff_output_rx: Receiver<HandoffOutput>,
handoff_output_rx: Receiver<RealtimeOutbound>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
@@ -206,16 +203,23 @@ struct RealtimeInputTask {
struct RealtimeInputChannels {
text_rx: Receiver<ConversationTextParams>,
handoff_output_rx: Receiver<HandoffOutput>,
handoff_output_rx: Receiver<RealtimeOutbound>,
audio_rx: Receiver<RealtimeAudioFrame>,
}
impl RealtimeHandoffState {
fn new(output_tx: Sender<HandoffOutput>, session_kind: RealtimeSessionKind) -> Self {
fn new(
output_tx: Sender<RealtimeOutbound>,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
session_kind: RealtimeSessionKind,
) -> Self {
Self {
output_tx,
active_handoff: Arc::new(Mutex::new(None)),
last_output_text: Arc::new(Mutex::new(None)),
codex_responses_as_items,
codex_response_item_prefix,
session_kind,
}
}
@@ -236,6 +240,8 @@ struct RealtimeStart {
api_provider: ApiProvider,
architecture: RealtimeConversationArchitecture,
extra_headers: Option<HeaderMap>,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
realtime_call_api_provider: Option<ApiProvider>,
session_config: RealtimeSessionConfig,
model_client: ModelClient,
@@ -290,6 +296,8 @@ impl RealtimeConversationManager {
api_provider,
architecture,
extra_headers,
codex_responses_as_items,
codex_response_item_prefix,
realtime_call_api_provider,
session_config,
model_client,
@@ -306,12 +314,17 @@ impl RealtimeConversationManager {
let (text_tx, text_rx) =
async_channel::bounded::<ConversationTextParams>(TEXT_IN_QUEUE_CAPACITY);
let (handoff_output_tx, handoff_output_rx) =
async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
async_channel::bounded::<RealtimeOutbound>(HANDOFF_OUT_QUEUE_CAPACITY);
let (events_tx, events_rx) =
async_channel::bounded::<RealtimeEvent>(OUTPUT_EVENTS_QUEUE_CAPACITY);
let realtime_active = Arc::new(AtomicBool::new(true));
let handoff = RealtimeHandoffState::new(handoff_output_tx, session_kind);
let handoff = RealtimeHandoffState::new(
handoff_output_tx,
codex_responses_as_items,
codex_response_item_prefix,
session_kind,
);
let input_channels = RealtimeInputChannels {
text_rx,
handoff_output_rx,
@@ -480,29 +493,34 @@ impl RealtimeConversationManager {
let active_handoff = handoff.active_handoff.lock().await.clone();
let output = match active_handoff {
Some(handoff_id) => {
let output_text = prefix_realtime_text(
output_text,
REALTIME_BACKEND_TEXT_PREFIX,
handoff.session_kind,
);
let output_text = realtime_backend_output(output_text, handoff.session_kind);
*handoff.last_output_text.lock().await = Some(output_text.clone());
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
if handoff.codex_responses_as_items {
RealtimeOutbound::ConversationItem {
text: realtime_backend_item(
output_text,
handoff.codex_response_item_prefix.as_deref(),
),
}
} else {
RealtimeOutbound::HandoffUpdate {
handoff_id,
text: output_text,
}
}
}
None if output_text.trim().is_empty() => return Ok(()),
None => {
let output_text = prefix_realtime_text(
output_text,
REALTIME_BACKEND_TEXT_PREFIX,
handoff.session_kind,
);
HandoffOutput::StandaloneAssistantOutput {
output_text: truncate_realtime_text_to_token_budget(
&output_text,
REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET,
),
let output_text = realtime_backend_output(output_text, handoff.session_kind);
if handoff.codex_responses_as_items {
RealtimeOutbound::ConversationItem {
text: realtime_backend_item(
output_text,
handoff.codex_response_item_prefix.as_deref(),
),
}
} else {
RealtimeOutbound::StandaloneHandoff { text: output_text }
}
}
};
@@ -514,6 +532,31 @@ impl RealtimeConversationManager {
Ok(())
}
pub(crate) async fn append_speech(&self, text: String) -> CodexResult<()> {
if text.trim().is_empty() {
return Ok(());
}
let handoff = {
let guard = self.state.lock().await;
let Some(state) = guard.as_ref() else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
state.handoff.clone()
};
handoff
.output_tx
.send(RealtimeOutbound::StandaloneHandoff {
text: realtime_backend_output(text, handoff.session_kind),
})
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
Ok(())
}
pub(crate) async fn handoff_complete(&self) -> CodexResult<()> {
let handoff = {
let guard = self.state.lock().await;
@@ -534,12 +577,18 @@ impl RealtimeConversationManager {
return Ok(());
};
let output = if handoff.codex_responses_as_items {
RealtimeOutbound::HandoffCompleteAck { handoff_id }
} else {
RealtimeOutbound::CompletedHandoff {
handoff_id,
text: output_text,
}
};
handoff
.output_tx
.send(HandoffOutput::FinalUpdate {
handoff_id,
output_text,
})
.send(output)
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))
}
@@ -626,6 +675,8 @@ struct PreparedRealtimeConversationStart {
api_provider: ApiProvider,
architecture: RealtimeConversationArchitecture,
extra_headers: Option<HeaderMap>,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
realtime_call_api_provider: Option<ApiProvider>,
requested_realtime_session_id: Option<String>,
version: RealtimeWsVersion,
@@ -701,6 +752,8 @@ async fn prepare_realtime_start(
api_provider,
architecture,
extra_headers,
codex_responses_as_items: params.codex_responses_as_items,
codex_response_item_prefix: params.codex_response_item_prefix,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -812,6 +865,19 @@ fn prefix_realtime_text(text: String, prefix: &str, session_kind: RealtimeSessio
format!("{prefix}{text}")
}
fn realtime_backend_output(output_text: String, session_kind: RealtimeSessionKind) -> String {
let output_text = prefix_realtime_text(output_text, REALTIME_BACKEND_TEXT_PREFIX, session_kind);
truncate_realtime_text_to_token_budget(&output_text, REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET)
}
fn realtime_backend_item(text: String, prefix: Option<&str>) -> String {
let text = match prefix.filter(|prefix| !prefix.is_empty()) {
Some(prefix) => format!("{prefix}\n\n{text}"),
None => text,
};
truncate_realtime_text_to_token_budget(&text, REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET)
}
fn validate_realtime_voice(version: RealtimeWsVersion, voice: RealtimeVoice) -> CodexResult<()> {
let voices = RealtimeVoicesList::builtin();
let allowed = match version {
@@ -846,6 +912,8 @@ async fn handle_start_inner(
api_provider,
architecture,
extra_headers,
codex_responses_as_items,
codex_response_item_prefix,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -861,6 +929,8 @@ async fn handle_start_inner(
api_provider,
architecture,
extra_headers,
codex_responses_as_items,
codex_response_item_prefix,
realtime_call_api_provider,
session_config,
model_client: sess.services.model_client.clone(),
@@ -1089,6 +1159,23 @@ pub(crate) async fn handle_text(
}
}
pub(crate) async fn handle_speech(
sess: &Arc<Session>,
sub_id: String,
params: ConversationSpeechParams,
) {
debug!(text = %params.text, "[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() {
warn!("realtime speech append failed while the session was already ending");
} else {
send_conversation_error(sess, sub_id, err.to_string(), CodexErrorInfo::BadRequest)
.await;
}
}
}
pub(crate) async fn handle_close(sess: &Arc<Session>, sub_id: String) {
end_realtime_conversation(sess, sub_id, RealtimeConversationEnd::Requested).await;
}
@@ -1256,7 +1343,7 @@ async fn handle_text_input(
}
async fn handle_handoff_output(
handoff_output: Result<HandoffOutput, RecvError>,
handoff_output: Result<RealtimeOutbound, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
handoff_state: &RealtimeHandoffState,
@@ -1267,45 +1354,39 @@ async fn handle_handoff_output(
let result = match event_parser {
RealtimeEventParser::V1 => match handoff_output {
HandoffOutput::StandaloneAssistantOutput { output_text } => {
RealtimeOutbound::StandaloneHandoff { text } => {
// TODO(guinness): Use the new client event for standalone handoffs once the API changes are complete.
writer
.send_conversation_handoff_append(
STANDALONE_HANDOFF_ID.to_string(),
output_text,
)
.send_conversation_handoff_append(STANDALONE_HANDOFF_ID.to_string(), text)
.await
}
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
}
| HandoffOutput::FinalUpdate {
handoff_id,
output_text,
} => {
RealtimeOutbound::HandoffUpdate { handoff_id, text }
| RealtimeOutbound::CompletedHandoff { handoff_id, text } => {
writer
.send_conversation_function_call_output(handoff_id, output_text)
.send_conversation_function_call_output(handoff_id, text)
.await
}
RealtimeOutbound::ConversationItem { text } => {
writer
.send_conversation_item_create(text, ConversationTextRole::Developer)
.await
}
RealtimeOutbound::HandoffCompleteAck { .. } => Ok(()),
},
RealtimeEventParser::RealtimeV2 => match handoff_output {
HandoffOutput::StandaloneAssistantOutput { output_text } => {
RealtimeOutbound::StandaloneHandoff { text } => {
if let Err(err) = writer
.send_conversation_item_create(output_text, ConversationTextRole::User)
.send_conversation_item_create(text, ConversationTextRole::User)
.await
{
Err(err)
} else {
return response_create_queue
.request_create(writer, events_tx, "standalone assistant output")
.request_create(writer, events_tx, "standalone handoff")
.await;
}
}
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
} => {
RealtimeOutbound::HandoffUpdate { handoff_id, text } => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
match active_handoff {
Some(active_handoff) if active_handoff == handoff_id => {}
@@ -1315,12 +1396,12 @@ async fn handle_handoff_output(
}
}
writer
.send_conversation_item_create(output_text, ConversationTextRole::User)
.send_conversation_item_create(text, ConversationTextRole::User)
.await
}
HandoffOutput::FinalUpdate {
RealtimeOutbound::CompletedHandoff {
handoff_id,
output_text: _,
text: _,
} => {
if let Err(err) = writer
.send_conversation_function_call_output(
@@ -1336,6 +1417,16 @@ async fn handle_handoff_output(
.await;
}
}
RealtimeOutbound::ConversationItem { text } => {
writer
.send_conversation_item_create(text, ConversationTextRole::Developer)
.await
}
RealtimeOutbound::HandoffCompleteAck { handoff_id } => {
writer
.send_conversation_function_call_output(handoff_id, String::new())
.await
}
},
};
if let Err(err) = result {
@@ -1449,7 +1540,6 @@ async fn handle_realtime_server_event(
match session_kind {
RealtimeSessionKind::V1 => {
*handoff_state.last_output_text.lock().await = None;
*handoff_state.active_handoff.lock().await = Some(handoff.handoff_id.clone());
}
RealtimeSessionKind::V2 => {
@@ -1477,7 +1567,6 @@ async fn handle_realtime_server_event(
.await?;
}
None => {
*handoff_state.last_output_text.lock().await = None;
*handoff_state.active_handoff.lock().await =
Some(handoff.handoff_id.clone());
}
codex-rs/core/src/realtime_conversation_tests.rs
+6 -1
View File
@@ -128,7 +128,12 @@ fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() {
#[tokio::test]
async fn clears_active_handoff_explicitly() {
let (tx, _rx) = bounded(1);
let state = RealtimeHandoffState::new(tx, RealtimeSessionKind::V1);
let state = RealtimeHandoffState::new(
tx,
/*codex_responses_as_items*/ false,
/*codex_response_item_prefix*/ None,
RealtimeSessionKind::V1,
);
*state.active_handoff.lock().await = Some("handoff_1".to_string());
assert_eq!(
codex-rs/core/src/session/handlers.rs
+5
View File
@@ -1,5 +1,6 @@
use crate::realtime_conversation::handle_audio as handle_realtime_conversation_audio;
use crate::realtime_conversation::handle_close as handle_realtime_conversation_close;
use crate::realtime_conversation::handle_speech as handle_realtime_conversation_speech;
use crate::realtime_conversation::handle_start as handle_realtime_conversation_start;
use crate::realtime_conversation::handle_text as handle_realtime_conversation_text;
use async_channel::Receiver;
@@ -737,6 +738,10 @@ pub(super) async fn submission_loop(
handle_realtime_conversation_text(&sess, sub.id.clone(), params).await;
false
}
Op::RealtimeConversationSpeech(params) => {
handle_realtime_conversation_speech(&sess, sub.id.clone(), params).await;
false
}
Op::RealtimeConversationClose => {
handle_realtime_conversation_close(&sess, sub.id.clone()).await;
false