openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-05 15:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

use developer role for realtime commentary

Browse Source
This commit is contained in:
Alex Gamble
2026-06-12 15:59:16 -07:00
parent 216dee1189
commit ce6614d28f
5 changed files with 198 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
codex-rs/core/src/realtime_conversation.rs
View File

@@ -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(
&params.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 {

4
codex-rs/core/src/session/mod.rs
View File

@@ -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}");
}
}

21
codex-rs/core/src/session/turn.rs
View File

@@ -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<String> {
pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<ConversationTextParams> {
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<String> {
}
}
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.

37
codex-rs/core/src/session/turn_tests.rs
View File

@@ -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;

118
codex-rs/core/tests/suite/realtime_conversation.rs
View File

@@ -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(()));