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] add roles to realtime append text (#27936)

Browse Source 
## Summary

Add an explicit `user` or `developer` role to
`thread/realtime/appendText` and propagate it through the realtime input
queue into `conversation.item.create`. Older JSON clients that omit the
field continue to default to `user`.

This lets app-provided context such as memory retain developer authority
without bypassing app-server through a renderer-owned data channel. The
app-server schemas, API documentation, and focused protocol and
websocket coverage are updated with the new contract.

The Codex Apps consumer is tracked in
[openai/openai#1025261](https://github.com/openai/openai/pull/1025261).
This commit is contained in:
Alex Gamble
2026-06-12 15:05:37 -07:00
committed by GitHub
parent 9915d34684
commit 216dee1189
18 changed files with 176 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/realtime_conversation.rs
+38 -26
View File
@@ -35,6 +35,7 @@ use codex_protocol::protocol::ConversationAudioParams;
use codex_protocol::protocol::ConversationStartParams;
use codex_protocol::protocol::ConversationStartTransport;
use codex_protocol::protocol::ConversationTextParams;
use codex_protocol::protocol::ConversationTextRole;
use codex_protocol::protocol::ErrorEvent;
use codex_protocol::protocol::Event;
use codex_protocol::protocol::EventMsg;
@@ -62,7 +63,7 @@ use tracing::info;
use tracing::warn;
const AUDIO_IN_QUEUE_CAPACITY: usize = 256;
const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64;
const TEXT_IN_QUEUE_CAPACITY: usize = 64;
const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64;
const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256;
const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300;
@@ -194,7 +195,7 @@ impl RealtimeResponseCreateQueue {
struct RealtimeInputTask {
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
user_text_rx: Receiver<String>,
text_rx: Receiver<ConversationTextParams>,
handoff_output_rx: Receiver<HandoffOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
@@ -204,7 +205,7 @@ struct RealtimeInputTask {
}
struct RealtimeInputChannels {
user_text_rx: Receiver<String>,
text_rx: Receiver<ConversationTextParams>,
handoff_output_rx: Receiver<HandoffOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
}
@@ -223,7 +224,7 @@ impl RealtimeHandoffState {
#[allow(dead_code)]
struct ConversationState {
audio_tx: Sender<RealtimeAudioFrame>,
user_text_tx: Sender<String>,
text_tx: Sender<ConversationTextParams>,
session_kind: RealtimeSessionKind,
handoff: RealtimeHandoffState,
input_task: JoinHandle<()>,
@@ -302,8 +303,8 @@ impl RealtimeConversationManager {
let (audio_tx, audio_rx) =
async_channel::bounded::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
let (user_text_tx, user_text_rx) =
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
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);
let (events_tx, events_rx) =
@@ -312,7 +313,7 @@ impl RealtimeConversationManager {
let realtime_active = Arc::new(AtomicBool::new(true));
let handoff = RealtimeHandoffState::new(handoff_output_tx, session_kind);
let input_channels = RealtimeInputChannels {
user_text_rx,
text_rx,
handoff_output_rx,
audio_rx,
};
@@ -353,7 +354,7 @@ impl RealtimeConversationManager {
let task = spawn_realtime_input_task(RealtimeInputTask {
writer: connection.writer(),
events: connection.events(),
user_text_rx: input_channels.user_text_rx,
text_rx: input_channels.text_rx,
handoff_output_rx: input_channels.handoff_output_rx,
audio_rx: input_channels.audio_rx,
events_tx,
@@ -367,7 +368,7 @@ impl RealtimeConversationManager {
let mut guard = self.state.lock().await;
*guard = Some(ConversationState {
audio_tx,
user_text_tx,
text_tx,
session_kind,
handoff,
input_task: task,
@@ -440,12 +441,12 @@ impl RealtimeConversationManager {
}
}
pub(crate) async fn text_in(&self, text: String) -> CodexResult<()> {
pub(crate) async fn text_in(&self, mut params: ConversationTextParams) -> CodexResult<()> {
let sender = {
let guard = self.state.lock().await;
guard
.as_ref()
.map(|state| (state.user_text_tx.clone(), state.session_kind))
.map(|state| (state.text_tx.clone(), state.session_kind))
};
let Some((sender, session_kind)) = sender else {
@@ -454,9 +455,12 @@ impl RealtimeConversationManager {
));
};
let text = prefix_realtime_text(text, REALTIME_USER_TEXT_PREFIX, session_kind);
if params.role == ConversationTextRole::User {
params.text =
prefix_realtime_text(params.text, REALTIME_USER_TEXT_PREFIX, session_kind);
}
sender
.send(text)
.send(params)
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
Ok(())
@@ -1074,7 +1078,7 @@ pub(crate) async fn handle_text(
params: ConversationTextParams,
) {
debug!(text = %params.text, "[realtime-text] appending realtime conversation text input");
if let Err(err) = sess.conversation.text_in(params.text).await {
if let Err(err) = sess.conversation.text_in(params).await {
error!("failed to append realtime text: {err}");
if sess.conversation.running_state().await.is_some() {
warn!("realtime text input failed while the session was already ending");
@@ -1154,7 +1158,7 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J
run_realtime_input_task(RealtimeInputTask {
writer: connection.writer(),
events: connection.events(),
user_text_rx: input_channels.user_text_rx,
text_rx: input_channels.text_rx,
handoff_output_rx: input_channels.handoff_output_rx,
audio_rx: input_channels.audio_rx,
events_tx,
@@ -1170,7 +1174,7 @@ async fn run_realtime_input_task(input: RealtimeInputTask) {
let RealtimeInputTask {
writer,
events,
user_text_rx,
text_rx,
handoff_output_rx,
audio_rx,
events_tx,
@@ -1184,10 +1188,10 @@ async fn run_realtime_input_task(input: RealtimeInputTask) {
loop {
let result = tokio::select! {
// Text typed by the user that should be sent into realtime.
user_text = user_text_rx.recv() => {
handle_user_text_input(
user_text,
// Text input that should be sent into realtime.
text = text_rx.recv() => {
handle_text_input(
text,
&writer,
&events_tx,
)
@@ -1230,14 +1234,17 @@ async fn run_realtime_input_task(input: RealtimeInputTask) {
}
}
async fn handle_user_text_input(
text: Result<String, RecvError>,
async fn handle_text_input(
params: Result<ConversationTextParams, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
) -> anyhow::Result<()> {
let text = text.context("user text input channel closed")?;
let params = params.context("text input channel closed")?;
if let Err(err) = writer.send_conversation_item_create(text).await {
if let Err(err) = writer
.send_conversation_item_create(params.text, params.role)
.await
{
let mapped_error = map_api_error(err);
warn!("failed to send input text: {mapped_error}");
let _ = events_tx
@@ -1284,7 +1291,10 @@ async fn handle_handoff_output(
},
RealtimeEventParser::RealtimeV2 => match handoff_output {
HandoffOutput::StandaloneAssistantOutput { output_text } => {
if let Err(err) = writer.send_conversation_item_create(output_text).await {
if let Err(err) = writer
.send_conversation_item_create(output_text, ConversationTextRole::User)
.await
{
Err(err)
} else {
return response_create_queue
@@ -1304,7 +1314,9 @@ async fn handle_handoff_output(
return Ok(());
}
}
writer.send_conversation_item_create(output_text).await
writer
.send_conversation_item_create(output_text, ConversationTextRole::User)
.await
}
HandoffOutput::FinalUpdate {
handoff_id,