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

[codex] control automatic realtime handoff delivery (#27986)

Browse Source 
## What

Built on the realtime speech-control plumbing merged in #27917.

- Add optional `codexResponseHandoffPrefix` to `thread/realtime/start`.
- Apply that prefix only to automatic V1 commentary sent through
`conversation.handoff.append`; final answers remain unprefixed.
- Add opt-in `clientManagedHandoffs`. When true, core suppresses
automatic response handoffs and completion output so delivery is
controlled by explicit client append APIs.
- Preserve existing automatic behavior by default.
`codexResponsesAsItems: true` continues to select item routing when
client-managed mode is disabled.

## Why

Voice clients need two delivery policies: automatic background context
with silent commentary instructions and fully client-owned handoffs.
Phase-aware prefixing keeps routine commentary silent without
suppressing the final answer, while client-managed mode lets an app
decide exactly which updates to append.

## Validation

- `just fmt`
- `cargo test -p codex-app-server-protocol
serialize_thread_realtime_start`
- `RUST_MIN_STACK=16777216 cargo test -p codex-core --test all
conversation_handoff_persists_across_item_done_until_turn_complete`
- `RUST_MIN_STACK=16777216 cargo test -p codex-app-server --test all
webrtc_v1_client_managed_handoffs_disable_automatic_output`
- `RUST_MIN_STACK=16777216 cargo test -p codex-app-server --test all
webrtc_v1_final_automatic_handoff_omits_silent_prefix`
- `cargo build -p codex-cli --bin codex`
- Local Codex Apps compatibility check: 43 focused webview tests passed,
and a live voice session routed through the source-built app-server.

The explicit `RUST_MIN_STACK` avoids a macOS Tokio test-worker stack
overflow seen with the default test environment.
This commit is contained in:
jiayuhuang-openai
2026-06-17 19:22:29 -07:00
committed by GitHub
Unverified
parent a306ac4ee3
commit 683bd170dc
13 changed files with 355 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/realtime_conversation.rs
+61 -3
View File
@@ -30,6 +30,7 @@ use codex_login::read_openai_api_key_from_env;
use codex_model_provider_info::ModelProviderInfo;
use codex_protocol::error::CodexErr;
use codex_protocol::error::Result as CodexResult;
use codex_protocol::models::MessagePhase;
use codex_protocol::protocol::CodexErrorInfo;
use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationSpeechParams;
@@ -107,8 +108,10 @@ struct RealtimeHandoffState {
output_tx: Sender<RealtimeOutbound>,
active_handoff: Arc<Mutex<Option<String>>>,
last_output_text: Arc<Mutex<Option<String>>>,
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
session_kind: RealtimeSessionKind,
}
@@ -116,6 +119,7 @@ struct RealtimeHandoffState {
enum RealtimeOutbound {
StandaloneHandoff { text: String },
HandoffUpdate { handoff_id: String, text: String },
HandoffAppend { handoff_id: String, text: String },
CompletedHandoff { handoff_id: String, text: String },
ConversationItem { text: String },
HandoffCompleteAck { handoff_id: String },
@@ -210,16 +214,20 @@ struct RealtimeInputChannels {
impl RealtimeHandoffState {
fn new(
output_tx: Sender<RealtimeOutbound>,
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_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)),
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
session_kind,
}
}
@@ -240,8 +248,10 @@ struct RealtimeStart {
api_provider: ApiProvider,
architecture: RealtimeConversationArchitecture,
extra_headers: Option<HeaderMap>,
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
realtime_call_api_provider: Option<ApiProvider>,
session_config: RealtimeSessionConfig,
model_client: ModelClient,
@@ -296,8 +306,10 @@ impl RealtimeConversationManager {
api_provider,
architecture,
extra_headers,
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
realtime_call_api_provider,
session_config,
model_client,
@@ -321,8 +333,10 @@ impl RealtimeConversationManager {
let realtime_active = Arc::new(AtomicBool::new(true));
let handoff = RealtimeHandoffState::new(
handoff_output_tx,
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
session_kind,
);
let input_channels = RealtimeInputChannels {
@@ -479,7 +493,11 @@ impl RealtimeConversationManager {
Ok(())
}
pub(crate) async fn handoff_out(&self, output_text: String) -> CodexResult<()> {
pub(crate) async fn handoff_out(
&self,
output_text: String,
phase: Option<MessagePhase>,
) -> CodexResult<()> {
let handoff = {
let guard = self.state.lock().await;
let Some(state) = guard.as_ref() else {
@@ -490,6 +508,13 @@ impl RealtimeConversationManager {
state.handoff.clone()
};
if handoff.client_managed_handoffs {
return Ok(());
}
let response_handoff_prefix = match phase {
Some(MessagePhase::Commentary) => handoff.codex_response_handoff_prefix.clone(),
Some(MessagePhase::FinalAnswer) | None => None,
};
let active_handoff = handoff.active_handoff.lock().await.clone();
let output = match active_handoff {
Some(handoff_id) => {
@@ -502,6 +527,16 @@ impl RealtimeConversationManager {
handoff.codex_response_item_prefix.as_deref(),
),
}
} else if handoff.session_kind == RealtimeSessionKind::V1
&& handoff.codex_response_handoff_prefix.is_some()
{
RealtimeOutbound::HandoffAppend {
handoff_id,
text: realtime_backend_item(
output_text,
response_handoff_prefix.as_deref(),
),
}
} else {
RealtimeOutbound::HandoffUpdate {
handoff_id,
@@ -520,7 +555,13 @@ impl RealtimeConversationManager {
),
}
} else {
RealtimeOutbound::StandaloneHandoff { text: output_text }
RealtimeOutbound::StandaloneHandoff {
text: if handoff.session_kind == RealtimeSessionKind::V1 {
realtime_backend_item(output_text, response_handoff_prefix.as_deref())
} else {
output_text
},
}
}
}
};
@@ -565,6 +606,9 @@ impl RealtimeConversationManager {
let Some(handoff) = handoff else {
return Ok(());
};
if handoff.client_managed_handoffs {
return Ok(());
}
match handoff.session_kind {
RealtimeSessionKind::V1 => return Ok(()),
RealtimeSessionKind::V2 => {}
@@ -675,8 +719,10 @@ struct PreparedRealtimeConversationStart {
api_provider: ApiProvider,
architecture: RealtimeConversationArchitecture,
extra_headers: Option<HeaderMap>,
client_managed_handoffs: bool,
codex_responses_as_items: bool,
codex_response_item_prefix: Option<String>,
codex_response_handoff_prefix: Option<String>,
realtime_call_api_provider: Option<ApiProvider>,
requested_realtime_session_id: Option<String>,
version: RealtimeWsVersion,
@@ -744,8 +790,10 @@ async fn prepare_realtime_start(
api_provider,
architecture,
extra_headers,
client_managed_handoffs: params.client_managed_handoffs,
codex_responses_as_items: params.codex_responses_as_items,
codex_response_item_prefix: params.codex_response_item_prefix,
codex_response_handoff_prefix: params.codex_response_handoff_prefix,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -914,8 +962,10 @@ async fn handle_start_inner(
api_provider,
architecture,
extra_headers,
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
realtime_call_api_provider,
requested_realtime_session_id,
version,
@@ -931,8 +981,10 @@ async fn handle_start_inner(
api_provider,
architecture,
extra_headers,
client_managed_handoffs,
codex_responses_as_items,
codex_response_item_prefix,
codex_response_handoff_prefix,
realtime_call_api_provider,
session_config,
model_client: sess.services.model_client.clone(),
@@ -1368,6 +1420,11 @@ async fn handle_handoff_output(
.send_conversation_function_call_output(handoff_id, text)
.await
}
RealtimeOutbound::HandoffAppend { handoff_id, text } => {
writer
.send_conversation_handoff_append(handoff_id, text)
.await
}
RealtimeOutbound::ConversationItem { text } => {
writer
.send_conversation_item_create(text, ConversationTextRole::Developer)
@@ -1388,7 +1445,8 @@ async fn handle_handoff_output(
.await;
}
}
RealtimeOutbound::HandoffUpdate { handoff_id, text } => {
RealtimeOutbound::HandoffUpdate { handoff_id, text }
| RealtimeOutbound::HandoffAppend { handoff_id, text } => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
match active_handoff {
Some(active_handoff) if active_handoff == handoff_id => {}
codex-rs/core/src/realtime_conversation_tests.rs
+2
View File
@@ -130,8 +130,10 @@ async fn clears_active_handoff_explicitly() {
let (tx, _rx) = bounded(1);
let state = RealtimeHandoffState::new(
tx,
/*client_managed_handoffs*/ false,
/*codex_responses_as_items*/ false,
/*codex_response_item_prefix*/ None,
/*codex_response_handoff_prefix*/ None,
RealtimeSessionKind::V1,
);
codex-rs/core/src/session/mod.rs
+2 -2
View File
@@ -1809,13 +1809,13 @@ impl Session {
}
async fn maybe_mirror_event_text_to_realtime(&self, msg: &EventMsg) {
let Some(text) = realtime_text_for_event(msg) else {
let Some((text, phase)) = realtime_text_for_event(msg) else {
return;
};
if self.conversation.running_state().await.is_none() {
return;
}
if let Err(err) = self.conversation.handoff_out(text).await {
if let Err(err) = self.conversation.handoff_out(text, phase).await {
debug!("failed to mirror event text to realtime conversation: {err}");
}
}
codex-rs/core/src/session/turn.rs
+3 -3
View File
@@ -1460,11 +1460,11 @@ fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String
.collect()
}
pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<(String, Option<MessagePhase>)> {
match msg {
EventMsg::AgentMessage(event) => Some(event.message.clone()),
EventMsg::AgentMessage(event) => Some((event.message.clone(), event.phase.clone())),
EventMsg::ItemCompleted(event) => match &event.item {
TurnItem::AgentMessage(item) => Some(agent_message_text(item)),
TurnItem::AgentMessage(item) => Some((agent_message_text(item), item.phase.clone())),
_ => None,
},
EventMsg::Error(_)