From ce6614d28f48ea69a6d391b6152091dfa03fe475 Mon Sep 17 00:00:00 2001 From: Alex Gamble Date: Fri, 12 Jun 2026 15:59:16 -0700 Subject: [PATCH] use developer role for realtime commentary --- codex-rs/core/src/realtime_conversation.rs | 39 +++--- codex-rs/core/src/session/mod.rs | 4 +- codex-rs/core/src/session/turn.rs | 21 +++- codex-rs/core/src/session/turn_tests.rs | 37 ++++++ .../core/tests/suite/realtime_conversation.rs | 118 ++++++++++++++++++ 5 files changed, 198 insertions(+), 21 deletions(-) diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index a1a861c878..9c54624a98 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -113,10 +113,12 @@ struct RealtimeHandoffState { enum HandoffOutput { StandaloneAssistantOutput { output_text: String, + role: ConversationTextRole, }, ProgressUpdate { handoff_id: String, output_text: String, + role: ConversationTextRole, }, FinalUpdate { handoff_id: String, @@ -466,7 +468,7 @@ impl RealtimeConversationManager { Ok(()) } - pub(crate) async fn handoff_out(&self, output_text: String) -> CodexResult<()> { + pub(crate) async fn handoff_out(&self, mut params: ConversationTextParams) -> CodexResult<()> { let handoff = { let guard = self.state.lock().await; let Some(state) = guard.as_ref() else { @@ -480,29 +482,32 @@ 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, + params.text = prefix_realtime_text( + params.text, REALTIME_BACKEND_TEXT_PREFIX, handoff.session_kind, ); - *handoff.last_output_text.lock().await = Some(output_text.clone()); + *handoff.last_output_text.lock().await = Some(params.text.clone()); HandoffOutput::ProgressUpdate { handoff_id, - output_text, + output_text: params.text, + role: params.role, } } - None if output_text.trim().is_empty() => return Ok(()), + None if params.text.trim().is_empty() => return Ok(()), None => { - let output_text = prefix_realtime_text( - output_text, + params.text = prefix_realtime_text( + params.text, REALTIME_BACKEND_TEXT_PREFIX, handoff.session_kind, ); + params.text = truncate_realtime_text_to_token_budget( + ¶ms.text, + REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET, + ); HandoffOutput::StandaloneAssistantOutput { - output_text: truncate_realtime_text_to_token_budget( - &output_text, - REALTIME_ASSISTANT_OUTPUT_TOKEN_BUDGET, - ), + output_text: params.text, + role: params.role, } } }; @@ -1267,7 +1272,7 @@ async fn handle_handoff_output( let result = match event_parser { RealtimeEventParser::V1 => match handoff_output { - HandoffOutput::StandaloneAssistantOutput { output_text } => { + HandoffOutput::StandaloneAssistantOutput { output_text, .. } => { // TODO(guinness): Use the new client event for standalone handoffs once the API changes are complete. writer .send_conversation_handoff_append( @@ -1279,6 +1284,7 @@ async fn handle_handoff_output( HandoffOutput::ProgressUpdate { handoff_id, output_text, + .. } | HandoffOutput::FinalUpdate { handoff_id, @@ -1290,9 +1296,9 @@ async fn handle_handoff_output( } }, RealtimeEventParser::RealtimeV2 => match handoff_output { - HandoffOutput::StandaloneAssistantOutput { output_text } => { + HandoffOutput::StandaloneAssistantOutput { output_text, role } => { if let Err(err) = writer - .send_conversation_item_create(output_text, ConversationTextRole::User) + .send_conversation_item_create(output_text, role) .await { Err(err) @@ -1305,6 +1311,7 @@ async fn handle_handoff_output( HandoffOutput::ProgressUpdate { handoff_id, output_text, + role, } => { let active_handoff = handoff_state.active_handoff.lock().await.clone(); match active_handoff { @@ -1315,7 +1322,7 @@ async fn handle_handoff_output( } } writer - .send_conversation_item_create(output_text, ConversationTextRole::User) + .send_conversation_item_create(output_text, role) .await } HandoffOutput::FinalUpdate { diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index cc40c6d47d..a3d00d481f 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1824,13 +1824,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(params) = 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(params).await { debug!("failed to mirror event text to realtime conversation: {err}"); } } diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 6244274a40..dcddb7261a 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -92,6 +92,8 @@ use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AgentMessageContentDeltaEvent; use codex_protocol::protocol::AgentReasoningSectionBreakEvent; use codex_protocol::protocol::CodexErrorInfo; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::ConversationTextRole; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::PlanDeltaEvent; @@ -1402,11 +1404,17 @@ fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String .collect() } -pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option { +pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option { match msg { - EventMsg::AgentMessage(event) => Some(event.message.clone()), + EventMsg::AgentMessage(event) => Some(ConversationTextParams { + text: event.message.clone(), + role: realtime_role_for_message_phase(event.phase.as_ref()), + }), EventMsg::ItemCompleted(event) => match &event.item { - TurnItem::AgentMessage(item) => Some(agent_message_text(item)), + TurnItem::AgentMessage(item) => Some(ConversationTextParams { + text: agent_message_text(item), + role: realtime_role_for_message_phase(item.phase.as_ref()), + }), _ => None, }, EventMsg::Error(_) @@ -1486,6 +1494,13 @@ pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option { } } +fn realtime_role_for_message_phase(phase: Option<&MessagePhase>) -> ConversationTextRole { + match phase { + Some(MessagePhase::Commentary) => ConversationTextRole::Developer, + Some(MessagePhase::FinalAnswer) | None => ConversationTextRole::User, + } +} + /// Split the stream into normal assistant text vs. proposed plan content. /// Normal text becomes AgentMessage deltas; plan content becomes PlanDelta + /// TurnItem::Plan. diff --git a/codex-rs/core/src/session/turn_tests.rs b/codex-rs/core/src/session/turn_tests.rs index 3511c03bda..1d76663860 100644 --- a/codex-rs/core/src/session/turn_tests.rs +++ b/codex-rs/core/src/session/turn_tests.rs @@ -1,7 +1,12 @@ use super::*; use codex_extension_api::ExtensionData; use codex_extension_api::TurnItemContributor; +use codex_protocol::ThreadId; use codex_protocol::items::AgentMessageContent; +use codex_protocol::items::AgentMessageItem; +use codex_protocol::protocol::ConversationTextParams; +use codex_protocol::protocol::ConversationTextRole; +use codex_protocol::protocol::ItemCompletedEvent; use pretty_assertions::assert_eq; use std::sync::Arc; @@ -36,6 +41,38 @@ fn assistant_output_text(text: &str) -> ResponseItem { } } +#[test] +fn realtime_agent_message_role_follows_message_phase() { + for (phase, role) in [ + ( + Some(MessagePhase::Commentary), + ConversationTextRole::Developer, + ), + (Some(MessagePhase::FinalAnswer), ConversationTextRole::User), + (None, ConversationTextRole::User), + ] { + assert_eq!( + realtime_text_for_event(&EventMsg::ItemCompleted(ItemCompletedEvent { + thread_id: ThreadId::new(), + turn_id: "turn-1".to_string(), + item: TurnItem::AgentMessage(AgentMessageItem { + id: "message-1".to_string(), + content: vec![AgentMessageContent::Text { + text: "Status update".to_string(), + }], + phase, + memory_citation: None, + }), + completed_at_ms: 0, + })), + Some(ConversationTextParams { + text: "Status update".to_string(), + role, + }) + ); + } +} + #[tokio::test] async fn plan_mode_uses_contributed_turn_item_for_last_agent_message() { let (mut session, turn_context) = crate::session::tests::make_session_and_context().await; diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 85bc1827f0..c09593ed2b 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -2670,6 +2670,124 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn realtime_v2_mirrors_commentary_as_developer_text() -> Result<()> { + skip_if_no_network!(Ok(())); + + let api_server = start_mock_server().await; + let _response_mock = responses::mount_sse_once( + &api_server, + responses::sse(vec![ + responses::ev_response_created("resp_roles"), + json!({ + "type": "response.output_item.done", + "item": { + "type": "message", + "role": "assistant", + "id": "commentary_roles", + "content": [{"type": "output_text", "text": "Still working"}], + "phase": "commentary" + } + }), + json!({ + "type": "response.output_item.done", + "item": { + "type": "message", + "role": "assistant", + "id": "final_roles", + "content": [{"type": "output_text", "text": "Finished"}], + "phase": "final_answer" + } + }), + responses::ev_completed("resp_roles"), + ]), + ) + .await; + + let realtime_server = start_websocket_server(vec![vec![ + vec![ + json!({ + "type": "session.updated", + "session": { "id": "sess_roles", "instructions": "backend prompt" } + }), + json!({ + "type": "conversation.item.done", + "item": { + "id": "item_roles", + "type": "function_call", + "name": "background_agent", + "call_id": "handoff_roles", + "arguments": "{\"prompt\":\"Do the task\"}" + } + }), + ], + vec![], + vec![], + vec![], + vec![], + ]]) + .await; + + let mut builder = test_codex().with_config({ + let realtime_base_url = realtime_server.uri().to_string(); + move |config| { + config.experimental_realtime_ws_base_url = Some(realtime_base_url); + config.realtime.version = RealtimeWsVersion::V2; + } + }); + let test = builder.build(&api_server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + architecture: None, + model: None, + output_modality: RealtimeOutputModality::Audio, + prompt: Some(Some("backend prompt".to_string())), + realtime_session_id: None, + transport: None, + version: None, + voice: None, + })) + .await?; + + let _ = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HandoffRequested(handoff), + }) if handoff.handoff_id == "handoff_roles" => Some(()), + _ => None, + }) + .await; + + let commentary = wait_for_websocket_request(&realtime_server, 0, 1).await?; + assert_eq!( + commentary.body_json(), + json!({ + "type": "conversation.item.create", + "item": { + "type": "message", + "role": "developer", + "content": [{"type": "input_text", "text": "[BACKEND] Still working"}] + } + }) + ); + + let final_answer = wait_for_websocket_request(&realtime_server, 0, 2).await?; + assert_eq!( + final_answer.body_json(), + json!({ + "type": "conversation.item.create", + "item": { + "type": "message", + "role": "user", + "content": [{"type": "input_text", "text": "[BACKEND] Finished"}] + } + }) + ); + + realtime_server.shutdown().await; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> Result<()> { skip_if_no_network!(Ok(()));