openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-01 19:33:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add realtime queue and runtime control tools

Browse Source
This commit is contained in:
Peter Bakkum
2026-03-10 21:59:00 -07:00
parent 103d384781
commit 29e856b588
12 changed files with 2108 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+15
View File
@@ -334,6 +334,21 @@ pub(crate) async fn apply_bespoke_event_handling(
.await;
}
RealtimeEvent::CloseRequested(_) => {}
RealtimeEvent::ToolActionRequested(request) => {
let notification = ThreadRealtimeItemAddedNotification {
thread_id: conversation_id.to_string(),
item: serde_json::json!({
"type": "tool_action_request",
"call_id": request.call_id,
"action": request.action,
}),
};
outgoing
.send_server_notification(ServerNotification::ThreadRealtimeItemAdded(
notification,
))
.await;
}
RealtimeEvent::Error(message) => {
let notification = ThreadRealtimeErrorNotification {
thread_id: conversation_id.to_string(),
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+276 -45
View File
@@ -369,7 +369,7 @@ impl RealtimeWebsocketWriter {
pub async fn send_session_update(&self, instructions: String) -> Result<(), ApiError> {
self.send_json(RealtimeOutboundMessage::SessionUpdate {
session: SessionUpdateSession {
session: Box::new(SessionUpdateSession {
kind: "realtime".to_string(),
instructions,
output_modalities: vec!["audio".to_string()],
@@ -396,54 +396,161 @@ impl RealtimeWebsocketWriter {
voice: "marin".to_string(),
},
},
tools: vec![SessionTool {
kind: "function".to_string(),
name: "codex".to_string(),
description:
"Delegate a request to Codex and return the final result to the user."
.to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::from([
(
"prompt".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "The user request to delegate to Codex."
.to_string(),
},
),
(
"send_immediately".to_string(),
SessionToolProperty {
kind: "boolean".to_string(),
description: "When true, send this to Codex immediately, steering any running turn. When false or omitted, queue it for the next turn.".to_string(),
},
),
]),
required: vec!["prompt".to_string()],
tools: vec![
SessionTool {
kind: "function".to_string(),
name: "codex".to_string(),
description:
"Delegate a request to Codex and return the final result to the user."
.to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::from([
(
"prompt".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "The user request to delegate to Codex."
.to_string(),
},
),
(
"send_immediately".to_string(),
SessionToolProperty {
kind: "boolean".to_string(),
description: "When true, send this to Codex immediately, steering any running turn. When false or omitted, queue it for the next turn.".to_string(),
},
),
]),
required: vec!["prompt".to_string()],
},
},
}, SessionTool {
kind: "function".to_string(),
name: "cancel_current_operation".to_string(),
description: "Cancel the current Codex operation, equivalent to pressing Ctrl-C without exiting Codex.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::new(),
required: Vec::new(),
SessionTool {
kind: "function".to_string(),
name: "cancel_current_operation".to_string(),
description: "Cancel the current Codex operation, equivalent to pressing Ctrl-C without exiting Codex.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::new(),
required: Vec::new(),
},
},
}, SessionTool {
kind: "function".to_string(),
name: "turn_off_realtime_mode".to_string(),
description: "Turn off realtime voice mode, equivalent to using /realtime to stop live voice.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::new(),
required: Vec::new(),
SessionTool {
kind: "function".to_string(),
name: "turn_off_realtime_mode".to_string(),
description: "Turn off realtime voice mode, equivalent to using /realtime to stop live voice.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::new(),
required: Vec::new(),
},
},
}],
SessionTool {
kind: "function".to_string(),
name: "manage_message_queue".to_string(),
description: "Inspect or edit queued draft messages. Supported actions: list, replace_last, remove_last, clear. Queue editing affects queued draft messages only, not pending steers.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::from([
(
"action".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Queue action to run. Use one of: list, replace_last, remove_last, clear.".to_string(),
},
),
(
"message".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Replacement text for action=replace_last.".to_string(),
},
),
]),
required: vec!["action".to_string()],
},
},
SessionTool {
kind: "function".to_string(),
name: "manage_runtime_settings".to_string(),
description: "Inspect or update runtime settings for future Codex turns. Call with no setting fields to list current settings, possible settings, and allowed values. Supported setting keys: model, working_directory, reasoning_effort, fast_mode, personality, collaboration_mode. Changes are not persisted to disk.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::from([
(
"model".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Optional model slug to use for future Codex turns.".to_string(),
},
),
(
"working_directory".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Optional working directory for future Codex turns. Relative paths are resolved against the current working directory.".to_string(),
},
),
(
"reasoning_effort".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Optional reasoning effort. Supported values: default, minimal, low, medium, high, xhigh.".to_string(),
},
),
(
"fast_mode".to_string(),
SessionToolProperty {
kind: "boolean".to_string(),
description: "Optional Fast mode toggle for future Codex turns.".to_string(),
},
),
(
"personality".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Optional personality. Supported values: none, friendly, pragmatic.".to_string(),
},
),
(
"collaboration_mode".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Optional collaboration mode. Supported values: default, plan.".to_string(),
},
),
]),
required: Vec::new(),
},
},
SessionTool {
kind: "function".to_string(),
name: "run_tui_command".to_string(),
description: "Run a small set of built-in TUI actions. Supported commands: compact, review, plan, diff, agent. review can take an optional prompt for custom review instructions. plan can take an optional prompt to switch to Plan mode and submit work immediately.".to_string(),
parameters: SessionToolParameters {
kind: "object".to_string(),
properties: BTreeMap::from([
(
"command".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Built-in TUI command to run. Use one of: compact, review, plan, diff, agent.".to_string(),
},
),
(
"prompt".to_string(),
SessionToolProperty {
kind: "string".to_string(),
description: "Optional text argument for command=review or command=plan.".to_string(),
},
),
]),
required: vec!["command".to_string()],
},
},
],
tool_choice: "auto".to_string(),
},
}),
})
.await
}
@@ -552,6 +659,7 @@ impl RealtimeWebsocketEvents {
| RealtimeEvent::ResponseCancelled(_)
| RealtimeEvent::ConversationItemAdded(_)
| RealtimeEvent::ConversationItemDone { .. }
| RealtimeEvent::ToolActionRequested(_)
| RealtimeEvent::Error(_) => {}
}
}
@@ -748,6 +856,8 @@ mod tests {
use crate::endpoint::realtime_websocket::protocol::RealtimeInterruptRequested;
use crate::endpoint::realtime_websocket::protocol::RealtimeOutputAudioDelta;
use crate::endpoint::realtime_websocket::protocol::RealtimeResponseCancelled;
use crate::endpoint::realtime_websocket::protocol::RealtimeToolAction;
use crate::endpoint::realtime_websocket::protocol::RealtimeToolActionRequested;
use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptDelta;
use crate::endpoint::realtime_websocket::protocol::RealtimeTranscriptEntry;
use http::HeaderValue;
@@ -989,6 +1099,106 @@ mod tests {
);
}
#[test]
fn parse_manage_message_queue_requested_event() {
let payload = json!({
"type": "response.done",
"response": {
"output": [
{
"id": "item_queue",
"type": "function_call",
"name": "manage_message_queue",
"call_id": "queue_123",
"arguments": "{\"action\":\"list\"}"
}
]
}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
Some(RealtimeEvent::ToolActionRequested(
RealtimeToolActionRequested {
call_id: "queue_123".to_string(),
action: RealtimeToolAction::ManageMessageQueue {
action: "list".to_string(),
message: None,
},
}
))
);
}
#[test]
fn parse_manage_runtime_settings_requested_event() {
let payload = json!({
"type": "response.done",
"response": {
"output": [
{
"id": "item_settings",
"type": "function_call",
"name": "manage_runtime_settings",
"call_id": "settings_123",
"arguments": "{\"model\":\"gpt-5\",\"working_directory\":\"src\",\"reasoning_effort\":\"low\"}"
}
]
}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
Some(RealtimeEvent::ToolActionRequested(
RealtimeToolActionRequested {
call_id: "settings_123".to_string(),
action: RealtimeToolAction::ManageRuntimeSettings {
model: Some("gpt-5".to_string()),
working_directory: Some("src".to_string()),
reasoning_effort: Some("low".to_string()),
fast_mode: None,
personality: None,
collaboration_mode: None,
},
}
))
);
}
#[test]
fn parse_run_tui_command_requested_event() {
let payload = json!({
"type": "response.done",
"response": {
"output": [
{
"id": "item_command",
"type": "function_call",
"name": "run_tui_command",
"call_id": "command_123",
"arguments": "{\"command\":\"plan\",\"prompt\":\"make a plan\"}"
}
]
}
})
.to_string();
assert_eq!(
parse_realtime_event(payload.as_str()),
Some(RealtimeEvent::ToolActionRequested(
RealtimeToolActionRequested {
call_id: "command_123".to_string(),
action: RealtimeToolAction::RunTuiCommand {
command: "plan".to_string(),
prompt: Some("make a plan".to_string()),
},
}
))
);
}
#[test]
fn parse_input_audio_speech_started_event() {
let payload = json!({
@@ -1295,6 +1505,27 @@ mod tests {
first_json["session"]["tools"][2]["parameters"]["properties"],
json!({})
);
assert_eq!(
first_json["session"]["tools"]
.as_array()
.expect("tools array")
.iter()
.map(|tool| tool["name"].as_str().expect("tool name"))
.collect::<Vec<_>>(),
vec![
"codex",
"cancel_current_operation",
"turn_off_realtime_mode",
"manage_message_queue",
"manage_runtime_settings",
"run_tui_command",
]
);
assert_eq!(
first_json["session"]["tools"][4]["parameters"]["properties"]["working_directory"]
["type"],
Value::String("string".to_string())
);
ws.send(Message::Text(
json!({
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
+179 -1
View File
@@ -6,6 +6,8 @@ pub use codex_protocol::protocol::RealtimeInputAudioSpeechStarted;
pub use codex_protocol::protocol::RealtimeInterruptRequested;
pub use codex_protocol::protocol::RealtimeOutputAudioDelta;
pub use codex_protocol::protocol::RealtimeResponseCancelled;
pub use codex_protocol::protocol::RealtimeToolAction;
pub use codex_protocol::protocol::RealtimeToolActionRequested;
pub use codex_protocol::protocol::RealtimeTranscriptDelta;
pub use codex_protocol::protocol::RealtimeTranscriptEntry;
use serde::Deserialize;
@@ -36,7 +38,7 @@ pub(super) enum RealtimeOutboundMessage {
audio_end_ms: u32,
},
#[serde(rename = "session.update")]
SessionUpdate { session: SessionUpdateSession },
SessionUpdate { session: Box<SessionUpdateSession> },
#[serde(rename = "conversation.item.create")]
ConversationItemCreate { item: ConversationItem },
}
@@ -273,6 +275,9 @@ pub(super) fn parse_realtime_event(payload: &str) -> Option<RealtimeEvent> {
if let Some(close) = parse_close_requested(&parsed) {
return Some(RealtimeEvent::CloseRequested(close));
}
if let Some(tool_action) = parse_tool_action_requested(&parsed) {
return Some(RealtimeEvent::ToolActionRequested(tool_action));
}
if let Some(cancelled) = parse_response_cancelled(&parsed) {
return Some(RealtimeEvent::ResponseCancelled(cancelled));
}
@@ -357,6 +362,179 @@ fn parse_close_requested(parsed: &Value) -> Option<RealtimeCloseRequested> {
})
}
fn parse_tool_action_requested(parsed: &Value) -> Option<RealtimeToolActionRequested> {
#[derive(Debug, Deserialize, Default)]
struct ReplaceLastQueuedMessageArguments {
#[serde(default)]
message: String,
}
#[derive(Debug, Deserialize, Default)]
struct ManageMessageQueueArguments {
#[serde(default)]
action: String,
#[serde(default)]
message: Option<String>,
}
#[derive(Debug, Deserialize, Default)]
struct UpdateRuntimeSettingsArguments {
#[serde(default)]
model: Option<String>,
#[serde(default)]
working_directory: Option<String>,
#[serde(default)]
reasoning_effort: Option<String>,
#[serde(default)]
fast_mode: Option<bool>,
#[serde(default)]
personality: Option<String>,
#[serde(default)]
collaboration_mode: Option<String>,
}
#[derive(Debug, Deserialize, Default)]
struct RunTuiCommandArguments {
#[serde(default)]
command: String,
#[serde(default)]
prompt: Option<String>,
}
let parse_call_id = |function_call: &Value| {
function_call
.get("call_id")
.and_then(Value::as_str)
.map(str::to_string)
};
if let Some(function_call) = find_function_call(parsed, "manage_message_queue") {
let arguments = function_call
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
let parsed_arguments =
serde_json::from_str::<ManageMessageQueueArguments>(arguments).unwrap_or_default();
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::ManageMessageQueue {
action: parsed_arguments.action,
message: parsed_arguments.message,
},
});
}
if let Some(function_call) = find_function_call(parsed, "list_message_queue") {
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::ListMessageQueue,
});
}
if let Some(function_call) = find_function_call(parsed, "replace_last_queued_message") {
let arguments = function_call
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
let parsed_arguments = serde_json::from_str::<ReplaceLastQueuedMessageArguments>(arguments)
.unwrap_or_else(|_| ReplaceLastQueuedMessageArguments {
message: arguments.to_string(),
});
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::ReplaceLastQueuedMessage {
message: parsed_arguments.message,
},
});
}
if let Some(function_call) = find_function_call(parsed, "remove_last_queued_message") {
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::RemoveLastQueuedMessage,
});
}
if let Some(function_call) = find_function_call(parsed, "clear_queued_messages") {
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::ClearQueuedMessages,
});
}
if let Some(function_call) = find_function_call(parsed, "list_runtime_settings") {
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::ListRuntimeSettings,
});
}
if let Some(function_call) = find_function_call(parsed, "manage_runtime_settings") {
let arguments = function_call
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
let parsed_arguments =
serde_json::from_str::<UpdateRuntimeSettingsArguments>(arguments).unwrap_or_default();
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::ManageRuntimeSettings {
model: parsed_arguments.model,
working_directory: parsed_arguments.working_directory,
reasoning_effort: parsed_arguments.reasoning_effort,
fast_mode: parsed_arguments.fast_mode,
personality: parsed_arguments.personality,
collaboration_mode: parsed_arguments.collaboration_mode,
},
});
}
if let Some(function_call) = find_function_call(parsed, "update_runtime_settings") {
let arguments = function_call
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
let parsed_arguments =
serde_json::from_str::<UpdateRuntimeSettingsArguments>(arguments).unwrap_or_default();
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::UpdateRuntimeSettings {
model: parsed_arguments.model,
working_directory: parsed_arguments.working_directory,
reasoning_effort: parsed_arguments.reasoning_effort,
fast_mode: parsed_arguments.fast_mode,
personality: parsed_arguments.personality,
collaboration_mode: parsed_arguments.collaboration_mode,
},
});
}
if let Some(function_call) = find_function_call(parsed, "run_tui_command") {
let arguments = function_call
.get("arguments")
.and_then(Value::as_str)
.unwrap_or_default();
let parsed_arguments =
serde_json::from_str::<RunTuiCommandArguments>(arguments).unwrap_or_default();
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::RunTuiCommand {
command: parsed_arguments.command,
prompt: parsed_arguments.prompt,
},
});
}
if let Some(function_call) = find_function_call(parsed, "compact_conversation") {
return Some(RealtimeToolActionRequested {
call_id: parse_call_id(function_call)?,
action: RealtimeToolAction::CompactConversation,
});
}
None
}
fn parse_response_cancelled(parsed: &Value) -> Option<RealtimeResponseCancelled> {
let response = parsed.get("response")?.as_object()?;
let status = response.get("status").and_then(Value::as_str)?;
codex-rs/codex-api/tests/realtime_websocket_e2e.rs
+21
View File
@@ -161,6 +161,27 @@ async fn realtime_ws_e2e_session_create_and_event_flow() {
first_json["session"]["tools"][1]["parameters"]["properties"],
json!({})
);
assert_eq!(
first_json["session"]["tools"]
.as_array()
.expect("tools array")
.iter()
.map(|tool| tool["name"].as_str().expect("tool name"))
.collect::<Vec<_>>(),
vec![
"codex",
"cancel_current_operation",
"turn_off_realtime_mode",
"manage_message_queue",
"manage_runtime_settings",
"run_tui_command",
]
);
assert_eq!(
first_json["session"]["tools"][4]["parameters"]["properties"]["working_directory"]
["type"],
Value::String("string".to_string())
);
ws.send(Message::Text(
json!({
codex-rs/core/src/codex.rs
+30
View File
@@ -2547,6 +2547,19 @@ impl Session {
{
debug!("failed to send final realtime handoff tool output: {err}");
}
if let EventMsg::TurnAborted(event) = msg
&& event.reason == TurnAbortReason::Interrupted
&& let Some(call_id) = self.conversation.active_handoff_id().await
&& let Err(err) = self
.conversation
.tool_call_complete(
call_id,
"Cancelled the current Codex operation.".to_string(),
)
.await
{
debug!("failed to send interrupted realtime handoff tool output: {err}");
}
if matches!(msg, EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_)) {
self.conversation.clear_active_handoff().await;
}
@@ -4072,6 +4085,23 @@ async fn submission_loop(sess: Arc<Session>, config: Arc<Config>, rx_sub: Receiv
handle_realtime_conversation_close(&sess, sub.id.clone()).await;
false
}
Op::RealtimeConversationToolCallComplete(params) => {
if let Err(err) = sess
.conversation
.tool_call_complete(params.call_id, params.output_text)
.await
{
sess.send_event_raw(Event {
id: sub.id.clone(),
msg: EventMsg::Error(ErrorEvent {
message: err.to_string(),
codex_error_info: Some(CodexErrorInfo::Other),
}),
})
.await;
}
false
}
Op::OverrideTurnContext {
cwd,
approval_policy,
codex-rs/core/tests/suite/realtime_conversation.rs
+146
View File
@@ -16,6 +16,7 @@ use codex_protocol::protocol::RealtimeAudioFrame;
use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::TurnAbortReason;
use codex_protocol::user_input::UserInput;
use core_test_support::responses;
use core_test_support::responses::start_mock_server;
@@ -1321,6 +1322,134 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_interrupt_resolves_active_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
let (_gate_completion_tx, gate_completion_rx) = oneshot::channel();
let first_chunks = vec![
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_response_created("resp-1")),
},
StreamingSseChunk {
gate: None,
body: sse_event(responses::ev_assistant_message(
"msg-1",
"assistant message 1",
)),
},
StreamingSseChunk {
gate: Some(gate_completion_rx),
body: sse_event(responses::ev_completed("resp-1")),
},
];
let (api_server, _completions) = start_streaming_sse_server(vec![first_chunks]).await;
let realtime_server = start_websocket_server(vec![vec![
vec![
json!({
"type": "session.updated",
"session": { "id": "sess_interrupt", "instructions": "backend prompt" }
}),
realtime_handoff_requested_event("handoff_interrupt", "item_interrupt", "delegate now"),
],
vec![realtime_interrupt_requested_event("cancel_1")],
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);
}
});
let test = builder.build_with_streaming_server(&api_server).await?;
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
prompt: "backend prompt".to_string(),
session_id: None,
}))
.await?;
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::SessionUpdated { session_id, .. },
}) if session_id == "sess_interrupt" => Some(()),
_ => None,
})
.await;
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::HandoffRequested(handoff),
}) if handoff.handoff_id == "handoff_interrupt" => Some(()),
_ => None,
})
.await;
let first_append = realtime_server.wait_for_request(0, 1).await;
assert_eq!(
first_append.body_json()["type"].as_str(),
Some("conversation.item.create")
);
assert_eq!(
first_append.body_json()["item"]["type"].as_str(),
Some("message")
);
assert_eq!(
first_append.body_json()["item"]["role"].as_str(),
Some("assistant")
);
assert_eq!(
first_append.body_json()["item"]["content"][0]["text"].as_str(),
Some("assistant message 1")
);
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::TurnAborted(event) if event.reason == TurnAbortReason::Interrupted => Some(()),
_ => None,
})
.await;
let cancel_output_request = wait_for_matching_websocket_request(
&realtime_server,
"cancel_current_operation tool output",
|request| {
request.body_json()["item"]["type"].as_str() == Some("function_call_output")
&& request.body_json()["item"]["call_id"].as_str() == Some("cancel_1")
},
)
.await;
assert_eq!(
cancel_output_request.body_json()["item"]["output"].as_str(),
Some("{\"content\":\"Cancelled the current Codex operation.\"}")
);
let interrupted_handoff_request = wait_for_matching_websocket_request(
&realtime_server,
"interrupted realtime handoff tool output",
|request| {
request.body_json()["item"]["type"].as_str() == Some("function_call_output")
&& request.body_json()["item"]["call_id"].as_str() == Some("handoff_interrupt")
},
)
.await;
assert_eq!(
interrupted_handoff_request.body_json()["item"]["output"].as_str(),
Some("{\"content\":\"Cancelled the current Codex operation.\"}")
);
realtime_server.shutdown().await;
api_server.shutdown().await;
Ok(())
}
fn sse_event(event: Value) -> String {
responses::sse(vec![event])
}
@@ -1342,6 +1471,23 @@ fn realtime_handoff_requested_event(handoff_id: &str, item_id: &str, prompt: &st
})
}
fn realtime_interrupt_requested_event(call_id: &str) -> Value {
json!({
"type": "response.done",
"response": {
"output": [
{
"id": "item_cancel",
"type": "function_call",
"name": "cancel_current_operation",
"call_id": call_id,
"arguments": "{}",
}
]
}
})
}
fn message_input_texts(body: &Value, role: &str) -> Vec<String> {
body.get("input")
.and_then(Value::as_array)
codex-rs/protocol/src/protocol.rs
+67
View File
@@ -169,6 +169,63 @@ pub struct RealtimeCloseRequested {
pub call_id: String,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
#[serde(tag = "type", rename_all = "snake_case")]
pub enum RealtimeToolAction {
ManageMessageQueue {
action: String,
#[serde(skip_serializing_if = "Option::is_none")]
message: Option<String>,
},
ListMessageQueue,
ReplaceLastQueuedMessage {
message: String,
},
RemoveLastQueuedMessage,
ClearQueuedMessages,
ManageRuntimeSettings {
#[serde(skip_serializing_if = "Option::is_none")]
model: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
working_directory: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
reasoning_effort: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
fast_mode: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
personality: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
collaboration_mode: Option<String>,
},
ListRuntimeSettings,
UpdateRuntimeSettings {
#[serde(skip_serializing_if = "Option::is_none")]
model: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
working_directory: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
reasoning_effort: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
fast_mode: Option<bool>,
#[serde(skip_serializing_if = "Option::is_none")]
personality: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
collaboration_mode: Option<String>,
},
RunTuiCommand {
command: String,
#[serde(skip_serializing_if = "Option::is_none")]
prompt: Option<String>,
},
CompactConversation,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct RealtimeToolActionRequested {
pub call_id: String,
pub action: RealtimeToolAction,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct RealtimeInputAudioSpeechStarted {
#[serde(skip_serializing_if = "Option::is_none")]
@@ -199,6 +256,7 @@ pub enum RealtimeEvent {
HandoffRequested(RealtimeHandoffRequested),
InterruptRequested(RealtimeInterruptRequested),
CloseRequested(RealtimeCloseRequested),
ToolActionRequested(RealtimeToolActionRequested),
Error(String),
}
@@ -219,6 +277,12 @@ pub struct ConversationTextParams {
pub text: String,
}
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)]
pub struct RealtimeToolCallCompleteParams {
pub call_id: String,
pub output_text: String,
}
/// Submission operation
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, JsonSchema)]
#[serde(tag = "type", rename_all = "snake_case")]
@@ -247,6 +311,9 @@ pub enum Op {
/// Close the running realtime conversation stream.
RealtimeConversationClose,
/// Send a tool-call result back to the running realtime conversation stream.
RealtimeConversationToolCallComplete(RealtimeToolCallCompleteParams),
/// Legacy user input.
///
/// Prefer [`Op::UserTurn`] so the caller provides full turn context
codex-rs/tui/src/app.rs
+5
View File
@@ -2417,6 +2417,11 @@ impl App {
AppEvent::UpdatePersonality(personality) => {
self.on_update_personality(personality);
}
AppEvent::UpdateWorkingDirectory(cwd) => {
self.config.cwd = cwd.clone();
self.file_search.update_search_dir(cwd);
self.refresh_status_line();
}
AppEvent::OpenRealtimeAudioDeviceSelection { kind } => {
self.chat_widget.open_realtime_audio_device_selection(kind);
}
codex-rs/tui/src/app_event.rs
+3
View File
@@ -187,6 +187,9 @@ pub(crate) enum AppEvent {
/// Update the current personality in the running app and widget.
UpdatePersonality(Personality),
/// Update the app-level working directory state after the chat widget already applied it.
UpdateWorkingDirectory(PathBuf),
/// Persist the selected model and reasoning effort to the appropriate config.
PersistModelSelection {
model: String,
codex-rs/tui/src/chatwidget.rs
+16 -4
View File
@@ -1641,7 +1641,9 @@ impl ChatWidget {
self.saw_plan_item_this_turn = false;
}
// If there is a queued user message, send exactly one now to begin the next turn.
self.maybe_send_next_queued_input();
if !self.maybe_close_realtime_conversation_when_idle() {
self.maybe_send_next_queued_input();
}
// Emit a notification when the turn completes (suppressed if focused).
self.notify(Notification::AgentTurnComplete {
response: last_agent_message.unwrap_or_default(),
@@ -2027,7 +2029,9 @@ impl ChatWidget {
self.mcp_startup_status = None;
self.update_task_running_state();
self.maybe_send_next_queued_input();
if !self.maybe_close_realtime_conversation_when_idle() {
self.maybe_send_next_queued_input();
}
self.request_redraw();
}
@@ -2072,6 +2076,7 @@ impl ChatWidget {
self.restore_user_message_to_composer(combined);
}
self.refresh_pending_input_preview();
self.maybe_close_realtime_conversation_when_idle();
self.request_redraw();
}
@@ -7335,10 +7340,9 @@ impl ChatWidget {
self.bottom_pane
.set_audio_device_selection_enabled(self.realtime_audio_device_selection_enabled());
if !realtime_conversation_enabled && self.realtime_conversation.is_live() {
self.request_realtime_conversation_close(Some(
self.force_close_realtime_conversation(Some(
"Realtime voice mode was closed because the feature was disabled.".to_string(),
));
self.reset_realtime_conversation_state();
}
}
if feature == Feature::FastMode {
@@ -7468,6 +7472,14 @@ impl ChatWidget {
}
}
pub(crate) fn set_working_directory(&mut self, cwd: PathBuf) {
self.current_cwd = Some(cwd.clone());
self.config.cwd = cwd;
self.request_status_line_branch_refresh();
self.refresh_status_line();
self.request_redraw();
}
/// Set the syntax theme override in the widget's config copy.
pub(crate) fn set_tui_theme(&mut self, theme: Option<String>) {
self.config.tui_theme = theme;
codex-rs/tui/src/chatwidget/realtime.rs
+900 -2
View File
@@ -8,6 +8,10 @@ use codex_protocol::protocol::RealtimeConversationStartedEvent;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeOutputAudioDelta;
use codex_protocol::protocol::RealtimeToolAction;
use codex_protocol::protocol::RealtimeToolActionRequested;
use codex_protocol::protocol::RealtimeToolCallCompleteParams;
use serde_json::json;
const REALTIME_CONVERSATION_PROMPT: &str = concat!(
"You are in a realtime voice conversation in the Codex TUI. ",
@@ -21,6 +25,10 @@ const REALTIME_CONVERSATION_PROMPT: &str = concat!(
"text output that the user can read in the TUI. ",
"Use codex whenever the user asks for codebase-specific facts, debugging, file changes, ",
"command output, or anything that benefits from tools. ",
"Use manage_message_queue to inspect or edit queued draft work. Supported actions are list, replace_last, remove_last, and clear. ",
"Use manage_runtime_settings to inspect or change runtime settings for future turns, including model, working_directory, reasoning_effort, fast_mode, personality, and collaboration_mode. ",
"If you call manage_runtime_settings without any setting fields, it returns the current settings and the list of possible settings and allowed values. ",
"Use run_tui_command for built-in TUI actions. Supported commands are compact, review, plan, diff, and agent. ",
"When you speak to the user directly, be extremely concise. ",
"Prefer short spoken answers because the user can see the detailed text output themselves. ",
"Do not read long code snippets, long file paths, diffs, or command output aloud unless the ",
@@ -40,6 +48,7 @@ pub(super) enum RealtimeConversationPhase {
pub(super) struct RealtimeConversationUiState {
phase: RealtimeConversationPhase,
requested_close: bool,
close_when_idle: bool,
session_id: Option<String>,
warned_unsupported_composer_submission: bool,
meter_placeholder_id: Option<String>,
@@ -71,6 +80,18 @@ impl RealtimeConversationUiState {
pub(super) fn set_phase_for_test(&mut self, phase: RealtimeConversationPhase) {
self.phase = phase;
}
pub(super) fn phase_for_test(&self) -> RealtimeConversationPhase {
self.phase
}
pub(super) fn requested_close_for_test(&self) -> bool {
self.requested_close
}
pub(super) fn close_when_idle_for_test(&self) -> bool {
self.close_when_idle
}
}
#[derive(Clone, Debug, PartialEq)]
@@ -280,6 +301,7 @@ impl ChatWidget {
pub(super) fn start_realtime_conversation(&mut self) {
self.realtime_conversation.phase = RealtimeConversationPhase::Starting;
self.realtime_conversation.requested_close = false;
self.realtime_conversation.close_when_idle = false;
self.realtime_conversation.session_id = None;
self.realtime_conversation
.warned_unsupported_composer_submission = false;
@@ -301,10 +323,20 @@ impl ChatWidget {
self.realtime_conversation.requested_close = true;
self.realtime_conversation.phase = RealtimeConversationPhase::Stopping;
self.submit_op(Op::RealtimeConversationClose);
self.stop_realtime_local_audio();
self.set_footer_hint_override(None);
if self.bottom_pane.is_task_running() {
self.realtime_conversation.close_when_idle = true;
if !self.submit_op(Op::Interrupt) {
self.realtime_conversation.close_when_idle = false;
self.submit_op(Op::RealtimeConversationClose);
}
} else {
self.realtime_conversation.close_when_idle = false;
self.submit_op(Op::RealtimeConversationClose);
}
if let Some(message) = info_message {
self.add_info_message(message, None);
} else {
@@ -312,11 +344,43 @@ impl ChatWidget {
}
}
pub(super) fn force_close_realtime_conversation(&mut self, info_message: Option<String>) {
if !self.realtime_conversation.is_live() {
if let Some(message) = info_message {
self.add_info_message(message, None);
}
return;
}
self.realtime_conversation.requested_close = true;
self.realtime_conversation.close_when_idle = false;
self.stop_realtime_local_audio();
self.set_footer_hint_override(None);
self.submit_op(Op::RealtimeConversationClose);
self.reset_realtime_conversation_state();
if let Some(message) = info_message {
self.add_info_message(message, None);
} else {
self.request_redraw();
}
}
pub(super) fn maybe_close_realtime_conversation_when_idle(&mut self) -> bool {
if !self.realtime_conversation.close_when_idle || self.bottom_pane.is_task_running() {
return false;
}
self.realtime_conversation.close_when_idle = false;
self.submit_op(Op::RealtimeConversationClose)
}
pub(super) fn reset_realtime_conversation_state(&mut self) {
self.stop_realtime_local_audio();
self.set_footer_hint_override(None);
self.realtime_conversation.phase = RealtimeConversationPhase::Inactive;
self.realtime_conversation.requested_close = false;
self.realtime_conversation.close_when_idle = false;
self.realtime_conversation.session_id = None;
self.realtime_conversation
.warned_unsupported_composer_submission = false;
@@ -365,9 +429,17 @@ impl ChatWidget {
RealtimeEvent::InterruptRequested(_) => {}
RealtimeEvent::CloseRequested(_) => {
self.request_realtime_conversation_close(Some(
"Realtime voice mode turned off.".to_string(),
if self.bottom_pane.is_task_running() {
"Stopping realtime voice mode after current work is cancelled."
} else {
"Realtime voice mode turned off."
}
.to_string(),
));
}
RealtimeEvent::ToolActionRequested(request) => {
self.on_realtime_tool_action_requested(request);
}
RealtimeEvent::Error(message) => {
self.add_error_message(format!("Realtime voice error: {message}"));
self.reset_realtime_conversation_state();
@@ -446,6 +518,832 @@ impl ChatWidget {
}
}
fn on_realtime_tool_action_requested(&mut self, request: RealtimeToolActionRequested) {
let output_text = match request.action {
RealtimeToolAction::ManageMessageQueue { action, message } => {
self.manage_message_queue_via_realtime(action, message)
}
RealtimeToolAction::ListMessageQueue => self.realtime_message_queue_output(),
RealtimeToolAction::ReplaceLastQueuedMessage { message } => {
self.replace_last_queued_message_via_realtime(message)
}
RealtimeToolAction::RemoveLastQueuedMessage => {
self.remove_last_queued_message_via_realtime()
}
RealtimeToolAction::ClearQueuedMessages => self.clear_queued_messages_via_realtime(),
RealtimeToolAction::ManageRuntimeSettings {
model,
working_directory,
reasoning_effort,
fast_mode,
personality,
collaboration_mode,
} => self.update_runtime_settings_via_realtime(
model,
working_directory,
reasoning_effort,
fast_mode,
personality,
collaboration_mode,
),
RealtimeToolAction::ListRuntimeSettings => self.realtime_runtime_settings_output(),
RealtimeToolAction::UpdateRuntimeSettings {
model,
working_directory,
reasoning_effort,
fast_mode,
personality,
collaboration_mode,
} => self.update_runtime_settings_via_realtime(
model,
working_directory,
reasoning_effort,
fast_mode,
personality,
collaboration_mode,
),
RealtimeToolAction::RunTuiCommand { command, prompt } => {
self.run_tui_command_via_realtime(command, prompt)
}
RealtimeToolAction::CompactConversation => self.compact_conversation_via_realtime(),
};
self.complete_realtime_tool_call(request.call_id, output_text);
}
fn complete_realtime_tool_call(&self, call_id: String, output_text: String) {
self.app_event_tx
.send(AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(
RealtimeToolCallCompleteParams {
call_id,
output_text,
},
)));
}
fn realtime_message_queue_output(&self) -> String {
let queued_messages = self
.queued_user_messages
.iter()
.enumerate()
.map(|(index, message)| {
json!({
"position": index + 1,
"text": message.text,
"local_image_count": message.local_images.len(),
"remote_image_count": message.remote_image_urls.len(),
})
})
.collect::<Vec<_>>();
let pending_steers = self
.pending_steers
.iter()
.enumerate()
.map(|(index, pending)| {
json!({
"position": index + 1,
"text": pending.user_message.text,
"local_image_count": pending.user_message.local_images.len(),
"remote_image_count": pending.user_message.remote_image_urls.len(),
})
})
.collect::<Vec<_>>();
json!({
"queued_messages": queued_messages,
"pending_steers": pending_steers,
"supported_actions": ["list", "replace_last", "remove_last", "clear"],
"editable_scope": "Queue editing tools affect queued_messages only.",
})
.to_string()
}
fn manage_message_queue_via_realtime(
&mut self,
action: String,
message: Option<String>,
) -> String {
match action.trim().to_ascii_lowercase().as_str() {
"list" => self.realtime_message_queue_output(),
"replace_last" => self.replace_last_queued_message_via_realtime(
message.unwrap_or_default(),
),
"remove_last" => self.remove_last_queued_message_via_realtime(),
"clear" => self.clear_queued_messages_via_realtime(),
other => json!({
"status": "error",
"message": format!(
"Unsupported queue action `{other}`. Use list, replace_last, remove_last, or clear."
),
"supported_actions": ["list", "replace_last", "remove_last", "clear"],
})
.to_string(),
}
}
fn replace_last_queued_message_via_realtime(&mut self, message: String) -> String {
let replacement = message.trim().to_string();
if replacement.is_empty() {
return json!({
"status": "error",
"message": "The replacement message must not be empty.",
})
.to_string();
}
let Some(last_message) = self.queued_user_messages.back_mut() else {
return json!({
"status": "error",
"message": "There are no queued draft messages to replace.",
})
.to_string();
};
if !last_message.local_images.is_empty() || !last_message.remote_image_urls.is_empty() {
return json!({
"status": "error",
"message": "The last queued draft includes attachments and cannot be replaced via realtime tools.",
})
.to_string();
}
let previous_message = std::mem::replace(&mut last_message.text, replacement.clone());
last_message.text_elements.clear();
last_message.mention_bindings.clear();
self.refresh_pending_input_preview();
self.request_redraw();
json!({
"status": "ok",
"previous_message": previous_message,
"message": replacement,
"queued_message_count": self.queued_user_messages.len(),
})
.to_string()
}
fn remove_last_queued_message_via_realtime(&mut self) -> String {
let Some(removed_message) = self.queued_user_messages.pop_back() else {
return json!({
"status": "error",
"message": "There are no queued draft messages to remove.",
})
.to_string();
};
self.refresh_pending_input_preview();
self.request_redraw();
json!({
"status": "ok",
"removed_message": removed_message.text,
"queued_message_count": self.queued_user_messages.len(),
})
.to_string()
}
fn clear_queued_messages_via_realtime(&mut self) -> String {
let removed_count = self.queued_user_messages.len();
let cleared_messages = self
.queued_user_messages
.drain(..)
.map(|message| message.text)
.collect::<Vec<_>>();
self.refresh_pending_input_preview();
self.request_redraw();
json!({
"status": "ok",
"cleared_count": removed_count,
"cleared_messages": cleared_messages,
})
.to_string()
}
fn realtime_runtime_settings_output(&self) -> String {
json!({
"status": "ok",
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string()
}
fn current_runtime_settings_json(&self) -> serde_json::Value {
json!({
"working_directory": self.config.cwd.display().to_string(),
"model": self.current_model(),
"reasoning_effort": Self::status_line_reasoning_effort_label(
self.effective_reasoning_effort(),
),
"fast_mode": matches!(self.current_service_tier(), Some(ServiceTier::Fast)),
"personality": self.realtime_personality_name(
self.config.personality.unwrap_or(Personality::None),
),
"collaboration_mode": self.realtime_collaboration_mode_name(self.active_mode_kind()),
})
}
fn possible_runtime_settings_json(&self) -> serde_json::Value {
let available_models = self
.models_manager
.try_list_models()
.map(|models| {
models
.into_iter()
.map(|preset| preset.model)
.collect::<Vec<_>>()
})
.unwrap_or_else(|_| Vec::new());
let plan_available = collaboration_modes::plan_mask(self.models_manager.as_ref()).is_some();
json!({
"keys": [
"model",
"working_directory",
"reasoning_effort",
"fast_mode",
"personality",
"collaboration_mode",
],
"model_values": available_models,
"working_directory": {
"description": "Absolute path or path relative to the current working directory.",
},
"reasoning_effort_values": ["default", "none", "minimal", "low", "medium", "high", "xhigh"],
"fast_mode_values": [false, true],
"personality_values": ["none", "friendly", "pragmatic"],
"collaboration_mode_values": if plan_available {
json!(["default", "plan"])
} else {
json!(["default"])
},
})
}
fn update_runtime_settings_via_realtime(
&mut self,
model: Option<String>,
working_directory: Option<String>,
reasoning_effort: Option<String>,
fast_mode: Option<bool>,
personality: Option<String>,
collaboration_mode: Option<String>,
) -> String {
let requested_anything = model.is_some()
|| working_directory.is_some()
|| reasoning_effort.is_some()
|| fast_mode.is_some()
|| personality.is_some()
|| collaboration_mode.is_some();
if !requested_anything {
return self.realtime_runtime_settings_output();
}
let resolved_cwd = match working_directory {
Some(working_directory) => {
match self.resolve_realtime_working_directory(working_directory.as_str()) {
Ok(path) => Some(path),
Err(message) => {
return json!({
"status": "error",
"message": message,
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
}
}
None => None,
};
let parsed_reasoning_effort = match reasoning_effort {
Some(reasoning_effort) => {
match self.parse_realtime_reasoning_effort(reasoning_effort.as_str()) {
Ok(effort) => Some(effort),
Err(message) => {
return json!({
"status": "error",
"message": message,
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
}
}
None => None,
};
let parsed_personality = match personality {
Some(personality) => match self.parse_realtime_personality(personality.as_str()) {
Ok(personality) => Some(personality),
Err(message) => {
return json!({
"status": "error",
"message": message,
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
},
None => None,
};
let parsed_collaboration_mode = match collaboration_mode {
Some(collaboration_mode) => {
match self.parse_realtime_collaboration_mode(collaboration_mode.as_str()) {
Ok(mode) => Some(mode),
Err(message) => {
return json!({
"status": "error",
"message": message,
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
}
}
None => None,
};
let needs_model_lookup =
model.is_some() || parsed_reasoning_effort.is_some() || parsed_personality.is_some();
let target_preset = if needs_model_lookup {
let requested_model = model
.as_deref()
.unwrap_or_else(|| self.current_model())
.to_string();
match self.find_realtime_model_preset(requested_model.as_str()) {
Ok(preset) => Some(preset),
Err(message) => {
return json!({
"status": "error",
"message": message,
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
}
} else {
None
};
let target_model = target_preset
.as_ref()
.map(|preset| preset.model.clone())
.unwrap_or_else(|| self.current_model().to_string());
let target_effort = if let Some(effort) = parsed_reasoning_effort {
effort
} else if model.is_some() && target_model != self.current_model() {
target_preset
.as_ref()
.map(|preset| preset.default_reasoning_effort)
} else {
self.effective_reasoning_effort()
};
if let Some(preset) = &target_preset
&& let Some(effort) = target_effort
&& !preset
.supported_reasoning_efforts
.iter()
.any(|option| option.effort == effort)
{
return json!({
"status": "error",
"message": format!(
"Model {} does not support reasoning effort {}.",
preset.model,
Self::status_line_reasoning_effort_label(Some(effort)),
),
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
if let Some(personality) = parsed_personality
&& personality != Personality::None
&& !self.config.features.enabled(Feature::Personality)
{
return json!({
"status": "error",
"message": "The personality feature is disabled in this session.",
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
if let Some(personality) = parsed_personality
&& personality != Personality::None
&& !target_preset
.as_ref()
.is_some_and(|preset| preset.supports_personality)
{
return json!({
"status": "error",
"message": format!(
"Model {} does not support personalities.",
target_model
),
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
let target_collaboration_mask = match parsed_collaboration_mode {
Some(ModeKind::Default) => {
collaboration_modes::default_mask(self.models_manager.as_ref())
}
Some(ModeKind::Plan) => collaboration_modes::plan_mask(self.models_manager.as_ref()),
Some(_) => None,
None => None,
};
if parsed_collaboration_mode.is_some() && target_collaboration_mask.is_none() {
return json!({
"status": "error",
"message": "The requested collaboration mode is not available right now.",
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string();
}
let mut updated_fields = serde_json::Map::new();
let mut override_cwd = None;
let mut override_model = None;
let mut override_effort = None;
let mut override_service_tier = None;
let mut override_personality = None;
if let Some(path) = resolved_cwd {
self.set_working_directory(path.clone());
self.app_event_tx
.send(AppEvent::UpdateWorkingDirectory(path.clone()));
self.app_event_tx.send(AppEvent::CodexOp(Op::ListSkills {
cwds: Vec::new(),
force_reload: false,
}));
updated_fields.insert(
"working_directory".to_string(),
json!(path.display().to_string()),
);
override_cwd = Some(path);
}
if model.is_some() || parsed_reasoning_effort.is_some() {
if model.is_some() {
updated_fields.insert("model".to_string(), json!(target_model.clone()));
override_model = Some(target_model.clone());
}
updated_fields.insert(
"reasoning_effort".to_string(),
json!(Self::status_line_reasoning_effort_label(target_effort)),
);
override_effort = Some(target_effort);
self.set_model(&target_model);
self.set_reasoning_effort(target_effort);
}
if let Some(enabled) = fast_mode {
let next_service_tier = enabled.then_some(ServiceTier::Fast);
self.set_service_tier(next_service_tier);
updated_fields.insert("fast_mode".to_string(), json!(enabled));
override_service_tier = Some(next_service_tier);
}
if let Some(personality) = parsed_personality {
self.set_personality(personality);
updated_fields.insert(
"personality".to_string(),
json!(self.realtime_personality_name(personality)),
);
override_personality = Some(personality);
}
if let Some(mask) = target_collaboration_mask {
self.set_collaboration_mask(mask);
updated_fields.insert(
"collaboration_mode".to_string(),
json!(self.realtime_collaboration_mode_name(self.active_mode_kind())),
);
}
if override_cwd.is_some()
|| override_model.is_some()
|| override_effort.is_some()
|| override_service_tier.is_some()
|| override_personality.is_some()
{
self.app_event_tx
.send(AppEvent::CodexOp(Op::OverrideTurnContext {
cwd: override_cwd,
approval_policy: None,
sandbox_policy: None,
windows_sandbox_level: None,
model: override_model,
effort: override_effort,
summary: None,
service_tier: override_service_tier,
collaboration_mode: None,
personality: override_personality,
}));
}
self.refresh_status_line();
self.request_redraw();
json!({
"status": "ok",
"updated": updated_fields,
"current_settings": self.current_runtime_settings_json(),
"possible_settings": self.possible_runtime_settings_json(),
})
.to_string()
}
fn compact_conversation_via_realtime(&mut self) -> String {
if self.bottom_pane.is_task_running() {
return json!({
"status": "error",
"message": "Codex must be idle before compacting the conversation.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string();
}
self.clear_token_usage();
self.app_event_tx.send(AppEvent::CodexOp(Op::Compact));
json!({
"status": "ok",
"message": "Started compacting the current conversation.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string()
}
fn run_tui_command_via_realtime(&mut self, command: String, prompt: Option<String>) -> String {
match command.trim().to_ascii_lowercase().as_str() {
"compact" => self.compact_conversation_via_realtime(),
"review" => self.review_via_realtime(prompt),
"plan" => self.plan_via_realtime(prompt),
"diff" => self.diff_via_realtime(),
"agent" => self.agent_via_realtime(),
other => json!({
"status": "error",
"message": format!(
"Unsupported TUI command `{other}`. Use compact, review, plan, diff, or agent."
),
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string(),
}
}
fn review_via_realtime(&mut self, prompt: Option<String>) -> String {
let review_request = match prompt.as_deref().map(str::trim) {
Some(prompt) if !prompt.is_empty() => ReviewRequest {
target: ReviewTarget::Custom {
instructions: prompt.to_string(),
},
user_facing_hint: None,
},
_ => ReviewRequest {
target: ReviewTarget::UncommittedChanges,
user_facing_hint: None,
},
};
if !self.submit_op(Op::Review {
review_request: review_request.clone(),
}) {
return json!({
"status": "error",
"message": "Failed to start review.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string();
}
json!({
"status": "ok",
"command": "review",
"target": review_request.target,
"message": "Started review.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string()
}
fn plan_via_realtime(&mut self, prompt: Option<String>) -> String {
if !self.collaboration_modes_enabled() {
return json!({
"status": "error",
"message": "Collaboration modes are disabled.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string();
}
let Some(mask) = collaboration_modes::plan_mask(self.models_manager.as_ref()) else {
return json!({
"status": "error",
"message": "Plan mode is unavailable right now.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string();
};
self.set_collaboration_mask(mask);
let trimmed_prompt = prompt.unwrap_or_default().trim().to_string();
if trimmed_prompt.is_empty() {
return json!({
"status": "ok",
"command": "plan",
"message": "Switched to Plan mode.",
"current_settings": self.current_runtime_settings_json(),
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string();
}
let user_message = UserMessage {
text: trimmed_prompt,
local_images: Vec::new(),
remote_image_urls: Vec::new(),
text_elements: Vec::new(),
mention_bindings: Vec::new(),
};
if !self.is_session_configured()
|| self.bottom_pane.is_task_running()
|| self.is_review_mode
{
self.queue_user_message(user_message);
json!({
"status": "ok",
"command": "plan",
"message": "Switched to Plan mode and queued the plan request.",
"current_settings": self.current_runtime_settings_json(),
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string()
} else {
self.submit_user_message(user_message);
json!({
"status": "ok",
"command": "plan",
"message": "Switched to Plan mode and submitted the plan request.",
"current_settings": self.current_runtime_settings_json(),
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string()
}
}
fn diff_via_realtime(&mut self) -> String {
self.add_diff_in_progress();
let tx = self.app_event_tx.clone();
tokio::spawn(async move {
let text = match get_git_diff().await {
Ok((is_git_repo, diff_text)) => {
if is_git_repo {
diff_text
} else {
"`/diff` — _not inside a git repository_".to_string()
}
}
Err(err) => format!("Failed to compute diff: {err}"),
};
tx.send(AppEvent::DiffResult(text));
});
json!({
"status": "ok",
"command": "diff",
"message": "Started computing the current git diff.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string()
}
fn agent_via_realtime(&mut self) -> String {
self.app_event_tx.send(AppEvent::OpenAgentPicker);
json!({
"status": "ok",
"command": "agent",
"message": "Opened the agent picker.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
.to_string()
}
fn resolve_realtime_working_directory(
&self,
working_directory: &str,
) -> Result<PathBuf, String> {
let trimmed = working_directory.trim();
if trimmed.is_empty() {
return Err("The working_directory must not be empty.".to_string());
}
let requested_path = PathBuf::from(trimmed);
let resolved_path = if requested_path.is_absolute() {
requested_path
} else {
self.config.cwd.join(requested_path)
};
let canonical_path = std::fs::canonicalize(&resolved_path).map_err(|err| {
format!(
"Failed to resolve working_directory {}: {err}",
resolved_path.display()
)
})?;
if !canonical_path.is_dir() {
return Err(format!(
"The working_directory must point to a directory: {}",
canonical_path.display()
));
}
Ok(canonical_path)
}
fn find_realtime_model_preset(
&self,
model: &str,
) -> Result<codex_protocol::openai_models::ModelPreset, String> {
let models = self.models_manager.try_list_models().map_err(|_| {
"Models are being updated right now. Try again in a moment.".to_string()
})?;
models
.into_iter()
.find(|preset| preset.model == model)
.ok_or_else(|| {
format!(
"Unknown model `{model}`. Call manage_runtime_settings with no setting fields or use a model from /model."
)
})
}
fn parse_realtime_reasoning_effort(
&self,
reasoning_effort: &str,
) -> Result<Option<ReasoningEffortConfig>, String> {
match reasoning_effort.trim().to_ascii_lowercase().as_str() {
"default" => Ok(None),
"none" => Ok(Some(ReasoningEffortConfig::None)),
"minimal" => Ok(Some(ReasoningEffortConfig::Minimal)),
"low" => Ok(Some(ReasoningEffortConfig::Low)),
"medium" => Ok(Some(ReasoningEffortConfig::Medium)),
"high" => Ok(Some(ReasoningEffortConfig::High)),
"xhigh" => Ok(Some(ReasoningEffortConfig::XHigh)),
other => Err(format!(
"Unsupported reasoning_effort `{other}`. Use default, none, minimal, low, medium, high, or xhigh."
)),
}
}
fn parse_realtime_personality(&self, personality: &str) -> Result<Personality, String> {
match personality.trim().to_ascii_lowercase().as_str() {
"none" => Ok(Personality::None),
"friendly" => Ok(Personality::Friendly),
"pragmatic" => Ok(Personality::Pragmatic),
other => Err(format!(
"Unsupported personality `{other}`. Use none, friendly, or pragmatic."
)),
}
}
fn parse_realtime_collaboration_mode(&self, mode: &str) -> Result<ModeKind, String> {
match mode.trim().to_ascii_lowercase().as_str() {
"default" => Ok(ModeKind::Default),
"plan" => Ok(ModeKind::Plan),
other => Err(format!(
"Unsupported collaboration_mode `{other}`. Use default or plan."
)),
}
}
fn realtime_personality_name(&self, personality: Personality) -> &'static str {
match personality {
Personality::None => "none",
Personality::Friendly => "friendly",
Personality::Pragmatic => "pragmatic",
}
}
fn realtime_collaboration_mode_name(&self, mode: ModeKind) -> &'static str {
match mode {
ModeKind::Plan => "plan",
ModeKind::Default | ModeKind::PairProgramming | ModeKind::Execute => "default",
}
}
#[cfg(all(not(target_os = "linux"), feature = "voice-input"))]
fn start_realtime_local_audio(&mut self) {
if self.realtime_conversation.capture_stop_flag.is_some() {
codex-rs/tui/src/chatwidget/tests.rs
+450
View File
@@ -48,6 +48,7 @@ use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::models::MessagePhase;
use codex_protocol::openai_models::ModelPreset;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::openai_models::ReasoningEffortPreset;
use codex_protocol::openai_models::default_input_modalities;
use codex_protocol::parse_command::ParsedCommand;
@@ -85,6 +86,8 @@ use codex_protocol::protocol::RateLimitWindow;
use codex_protocol::protocol::RealtimeConversationRealtimeEvent;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeToolAction;
use codex_protocol::protocol::RealtimeToolActionRequested;
use codex_protocol::protocol::ReviewRequest;
use codex_protocol::protocol::ReviewTarget;
use codex_protocol::protocol::SessionSource;
@@ -95,6 +98,7 @@ use codex_protocol::protocol::ThreadRolledBackEvent;
use codex_protocol::protocol::TokenCountEvent;
use codex_protocol::protocol::TokenUsage;
use codex_protocol::protocol::TokenUsageInfo;
use codex_protocol::protocol::TurnAbortedEvent;
use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnStartedEvent;
use codex_protocol::protocol::UndoCompletedEvent;
@@ -1942,6 +1946,55 @@ fn next_interrupt_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>) {
}
}
fn next_realtime_close_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>) {
loop {
match op_rx.try_recv() {
Ok(Op::RealtimeConversationClose) => return,
Ok(_) => continue,
Err(TryRecvError::Empty) => {
panic!("expected realtime close op but queue was empty")
}
Err(TryRecvError::Disconnected) => {
panic!("expected realtime close op but channel closed")
}
}
}
}
fn next_app_event(rx: &mut tokio::sync::mpsc::UnboundedReceiver<AppEvent>) -> AppEvent {
rx.try_recv().expect("expected app event")
}
fn next_realtime_tool_call_complete(
rx: &mut tokio::sync::mpsc::UnboundedReceiver<AppEvent>,
) -> codex_protocol::protocol::RealtimeToolCallCompleteParams {
loop {
match rx.try_recv() {
Ok(AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(params))) => {
return params;
}
Ok(_) => continue,
Err(TryRecvError::Empty) => {
panic!("expected realtime tool completion but queue was empty")
}
Err(TryRecvError::Disconnected) => {
panic!("expected realtime tool completion but channel closed")
}
}
}
}
fn next_review_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>) -> ReviewRequest {
loop {
match op_rx.try_recv() {
Ok(Op::Review { review_request }) => return review_request,
Ok(_) => continue,
Err(TryRecvError::Empty) => panic!("expected review op but queue was empty"),
Err(TryRecvError::Disconnected) => panic!("expected review op but channel closed"),
}
}
}
fn assert_no_submit_op(op_rx: &mut tokio::sync::mpsc::UnboundedReceiver<Op>) {
while let Ok(op) = op_rx.try_recv() {
assert!(
@@ -10461,6 +10514,403 @@ async fn realtime_handoff_can_send_immediately_while_turn_is_running() {
}
}
#[tokio::test]
async fn realtime_close_request_interrupts_before_disconnect_when_turn_is_running() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
chat.realtime_conversation
.set_phase_for_test(super::realtime::RealtimeConversationPhase::Active);
chat.on_task_started();
chat.handle_codex_event(Event {
id: "rt-close-1".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::CloseRequested(
codex_protocol::protocol::RealtimeCloseRequested {
call_id: "close_1".to_string(),
},
),
}),
});
assert_eq!(
chat.realtime_conversation.phase_for_test(),
super::realtime::RealtimeConversationPhase::Stopping
);
assert!(chat.realtime_conversation.requested_close_for_test());
assert!(chat.realtime_conversation.close_when_idle_for_test());
next_interrupt_op(&mut op_rx);
let inserted = drain_insert_history(&mut rx);
let rendered = inserted
.iter()
.map(|lines| lines_to_single_string(lines))
.find(|text| text.contains("Stopping realtime voice mode after current work is cancelled."))
.expect("expected close deferral info message");
assert!(rendered.contains("Stopping realtime voice mode after current work is cancelled."));
chat.handle_codex_event(Event {
id: "turn-abort".into(),
msg: EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some("turn-1".to_string()),
reason: TurnAbortReason::Interrupted,
}),
});
next_realtime_close_op(&mut op_rx);
assert!(!chat.realtime_conversation.close_when_idle_for_test());
}
#[tokio::test]
async fn realtime_close_request_closes_immediately_when_idle() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
chat.realtime_conversation
.set_phase_for_test(super::realtime::RealtimeConversationPhase::Active);
chat.handle_codex_event(Event {
id: "rt-close-2".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::CloseRequested(
codex_protocol::protocol::RealtimeCloseRequested {
call_id: "close_2".to_string(),
},
),
}),
});
next_realtime_close_op(&mut op_rx);
assert!(!chat.realtime_conversation.close_when_idle_for_test());
let inserted = drain_insert_history(&mut rx);
let rendered = inserted
.iter()
.map(|lines| lines_to_single_string(lines))
.find(|text| text.contains("Realtime voice mode turned off."))
.expect("expected immediate close info message");
assert!(rendered.contains("Realtime voice mode turned off."));
}
#[tokio::test]
async fn realtime_manage_message_queue_updates_queue_and_completes_tool_call() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
chat.queued_user_messages
.push_back(UserMessage::from("old queued message".to_string()));
chat.handle_codex_event(Event {
id: "rt-queue-replace".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "queue-tool-1".to_string(),
action: RealtimeToolAction::ManageMessageQueue {
action: "replace_last".to_string(),
message: Some("new queued message".to_string()),
},
}),
}),
});
assert_eq!(
chat.queued_user_messages
.back()
.map(|message| message.text.as_str()),
Some("new queued message")
);
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(params)) => {
assert_eq!(params.call_id, "queue-tool-1");
assert_eq!(
serde_json::from_str::<serde_json::Value>(&params.output_text)
.expect("valid json tool output"),
serde_json::json!({
"status": "ok",
"previous_message": "old queued message",
"message": "new queued message",
"queued_message_count": 1,
})
);
}
other => panic!("unexpected app event: {other:?}"),
}
}
#[tokio::test]
async fn realtime_manage_runtime_settings_updates_state_and_emits_follow_up_ops() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.2-codex")).await;
let temp = tempdir().expect("tempdir");
let next_cwd = temp.path().join("nested");
std::fs::create_dir_all(&next_cwd).expect("create nested dir");
let canonical_cwd = next_cwd.canonicalize().expect("canonicalize cwd");
chat.handle_codex_event(Event {
id: "rt-settings-1".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "settings-tool-1".to_string(),
action: RealtimeToolAction::ManageRuntimeSettings {
model: Some("gpt-5.3-codex".to_string()),
working_directory: Some(next_cwd.display().to_string()),
reasoning_effort: Some("low".to_string()),
fast_mode: None,
personality: None,
collaboration_mode: None,
},
}),
}),
});
assert_eq!(chat.config.cwd, canonical_cwd);
assert_eq!(chat.current_model(), "gpt-5.3-codex");
assert_eq!(chat.current_reasoning_effort(), Some(ReasoningEffort::Low));
match next_app_event(&mut rx) {
AppEvent::UpdateWorkingDirectory(cwd) => assert_eq!(cwd, canonical_cwd),
other => panic!("unexpected app event: {other:?}"),
}
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::ListSkills { cwds, force_reload }) => {
assert!(cwds.is_empty());
assert!(!force_reload);
}
other => panic!("unexpected app event: {other:?}"),
}
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::OverrideTurnContext {
cwd, model, effort, ..
}) => {
assert_eq!(cwd, Some(canonical_cwd.clone()));
assert_eq!(model.as_deref(), Some("gpt-5.3-codex"));
assert_eq!(effort, Some(Some(ReasoningEffort::Low)));
}
other => panic!("unexpected app event: {other:?}"),
}
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(params)) => {
assert_eq!(params.call_id, "settings-tool-1");
let output =
serde_json::from_str::<serde_json::Value>(&params.output_text).expect("json");
assert_eq!(output["status"], "ok");
assert_eq!(output["updated"]["model"], "gpt-5.3-codex");
assert_eq!(
output["updated"]["working_directory"],
canonical_cwd.display().to_string()
);
assert_eq!(output["updated"]["reasoning_effort"], "low");
assert_eq!(
output["current_settings"]["working_directory"],
canonical_cwd.display().to_string()
);
assert_eq!(output["current_settings"]["model"], "gpt-5.3-codex");
assert_eq!(output["current_settings"]["reasoning_effort"], "low");
assert_eq!(
output["possible_settings"]["keys"],
serde_json::json!([
"model",
"working_directory",
"reasoning_effort",
"fast_mode",
"personality",
"collaboration_mode",
])
);
}
other => panic!("unexpected app event: {other:?}"),
}
}
#[tokio::test]
async fn realtime_manage_runtime_settings_lists_possible_settings() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.2-codex")).await;
chat.handle_codex_event(Event {
id: "rt-settings-list".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "settings-tool-list".to_string(),
action: RealtimeToolAction::ManageRuntimeSettings {
model: None,
working_directory: None,
reasoning_effort: None,
fast_mode: None,
personality: None,
collaboration_mode: None,
},
}),
}),
});
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(params)) => {
assert_eq!(params.call_id, "settings-tool-list");
let output =
serde_json::from_str::<serde_json::Value>(&params.output_text).expect("json");
assert_eq!(output["status"], "ok");
assert_eq!(output["current_settings"]["model"], "gpt-5.2-codex");
assert_eq!(
output["possible_settings"]["keys"],
serde_json::json!([
"model",
"working_directory",
"reasoning_effort",
"fast_mode",
"personality",
"collaboration_mode",
])
);
assert_eq!(
output["possible_settings"]["reasoning_effort_values"],
serde_json::json!([
"default", "none", "minimal", "low", "medium", "high", "xhigh"
])
);
}
other => panic!("unexpected app event: {other:?}"),
}
}
#[tokio::test]
async fn realtime_run_tui_command_compact_emits_compact_and_tool_completion() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
chat.handle_codex_event(Event {
id: "rt-compact-1".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "compact-tool-1".to_string(),
action: RealtimeToolAction::RunTuiCommand {
command: "compact".to_string(),
prompt: None,
},
}),
}),
});
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::Compact) => {}
other => panic!("unexpected app event: {other:?}"),
}
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(params)) => {
assert_eq!(params.call_id, "compact-tool-1");
assert_eq!(
serde_json::from_str::<serde_json::Value>(&params.output_text).expect("json"),
serde_json::json!({
"status": "ok",
"message": "Started compacting the current conversation.",
"possible_commands": ["compact", "review", "plan", "diff", "agent"],
})
);
}
other => panic!("unexpected app event: {other:?}"),
}
}
#[tokio::test]
async fn realtime_run_tui_command_review_emits_review_op_and_tool_completion() {
let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
chat.handle_codex_event(Event {
id: "rt-review-1".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "review-tool-1".to_string(),
action: RealtimeToolAction::RunTuiCommand {
command: "review".to_string(),
prompt: Some("focus on risky changes".to_string()),
},
}),
}),
});
assert_eq!(
next_review_op(&mut op_rx),
ReviewRequest {
target: ReviewTarget::Custom {
instructions: "focus on risky changes".to_string(),
},
user_facing_hint: None,
}
);
let params = next_realtime_tool_call_complete(&mut rx);
assert_eq!(params.call_id, "review-tool-1");
let output = serde_json::from_str::<serde_json::Value>(&params.output_text).expect("json");
assert_eq!(output["status"], "ok");
assert_eq!(output["command"], "review");
assert_eq!(
output["possible_commands"],
serde_json::json!(["compact", "review", "plan", "diff", "agent"])
);
}
#[tokio::test]
async fn realtime_run_tui_command_plan_switches_mode_and_queues_prompt() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.2-codex")).await;
chat.handle_codex_event(Event {
id: "rt-plan-1".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "plan-tool-1".to_string(),
action: RealtimeToolAction::RunTuiCommand {
command: "plan".to_string(),
prompt: Some("plan the migration".to_string()),
},
}),
}),
});
assert_eq!(chat.active_mode_kind(), ModeKind::Plan);
assert_eq!(chat.queued_user_messages.len(), 1);
assert_eq!(
chat.queued_user_messages
.front()
.map(|message| message.text.as_str()),
Some("plan the migration")
);
let params = next_realtime_tool_call_complete(&mut rx);
assert_eq!(params.call_id, "plan-tool-1");
let output = serde_json::from_str::<serde_json::Value>(&params.output_text).expect("json");
assert_eq!(output["status"], "ok");
assert_eq!(output["command"], "plan");
assert_eq!(
output["message"],
"Switched to Plan mode and queued the plan request."
);
}
#[tokio::test]
async fn realtime_run_tui_command_agent_opens_picker_and_completes_tool_call() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
chat.handle_codex_event(Event {
id: "rt-agent-1".into(),
msg: EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ToolActionRequested(RealtimeToolActionRequested {
call_id: "agent-tool-1".to_string(),
action: RealtimeToolAction::RunTuiCommand {
command: "agent".to_string(),
prompt: None,
},
}),
}),
});
match next_app_event(&mut rx) {
AppEvent::OpenAgentPicker => {}
other => panic!("unexpected app event: {other:?}"),
}
match next_app_event(&mut rx) {
AppEvent::CodexOp(Op::RealtimeConversationToolCallComplete(params)) => {
assert_eq!(params.call_id, "agent-tool-1");
let output =
serde_json::from_str::<serde_json::Value>(&params.output_text).expect("json");
assert_eq!(output["status"], "ok");
assert_eq!(output["command"], "agent");
}
other => panic!("unexpected app event: {other:?}"),
}
}
#[tokio::test]
async fn review_queues_user_messages_snapshot() {
let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;