openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-17 12:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 8b1b3f1356 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2025-08-15 13:55:09 -07:00
committed by GitHub
parent 54c598f8fe 8b1b3f1356
commit 57de0099bc
14 changed files with 212 additions and 1521 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
AGENTS.md
View File

@@ -12,6 +12,15 @@ Before finalizing a change to `codex-rs`, run `just fmt` (in `codex-rs` director
1. Run the test for the specific project that was changed. For example, if changes were made in `codex-rs/tui`, run `cargo test -p codex-tui`.
2. Once those pass, if any changes were made in common, core, or protocol, run the complete test suite with `cargo test --all-features`.
## TUI code conventions
- Use concise styling helpers from ratatui’s Stylize trait.
- Basic spans: use "text".into()
- Styled spans: use "text".red(), "text".green(), "text".magenta(), "text".dim(), etc.
- Prefer these over constructing styles with `Span::styled` and `Style` directly.
- Example: patch summary file lines
- Desired: vec![" └ ".into(), "M".red(), " ".dim(), "tui/src/app.rs".dim()]
## Snapshot tests
This repo uses snapshot tests (via `insta`), especially in `codex-rs/tui`, to validate rendered output. When UI or text output changes intentionally, update the snapshots as follows:

5
codex-rs/cli/src/login.rs
View File

@@ -21,10 +21,7 @@ pub async fn login_with_chatgpt(codex_home: PathBuf) -> std::io::Result<()> {
server.actual_port, server.auth_url,
);
server.block_until_done()?;
eprintln!("Successfully logged in");
Ok(())
server.block_until_done()
}
pub async fn run_login_with_chatgpt(cli_config_overrides: CliConfigOverrides) -> ! {

2
codex-rs/core/src/client_common.rs
View File

@@ -94,6 +94,7 @@ pub(crate) struct Reasoning {
#[derive(Debug, Serialize, Default, Clone, Copy)]
#[serde(rename_all = "lowercase")]
pub(crate) enum OpenAiReasoningEffort {
Minimal,
Low,
#[default]
Medium,
@@ -103,6 +104,7 @@ pub(crate) enum OpenAiReasoningEffort {
impl From<ReasoningEffortConfig> for Option<OpenAiReasoningEffort> {
fn from(effort: ReasoningEffortConfig) -> Self {
match effort {
ReasoningEffortConfig::Minimal => Some(OpenAiReasoningEffort::Minimal),
ReasoningEffortConfig::Low => Some(OpenAiReasoningEffort::Low),
ReasoningEffortConfig::Medium => Some(OpenAiReasoningEffort::Medium),
ReasoningEffortConfig::High => Some(OpenAiReasoningEffort::High),

1
codex-rs/core/src/config_types.rs
View File

@@ -206,6 +206,7 @@ impl From<ShellEnvironmentPolicyToml> for ShellEnvironmentPolicy {
#[serde(rename_all = "lowercase")]
#[strum(serialize_all = "lowercase")]
pub enum ReasoningEffort {
Minimal,
Low,
#[default]
Medium,

3
codex-rs/mcp-server/src/lib.rs
View File

@@ -18,15 +18,12 @@ use tracing_subscriber::EnvFilter;
mod codex_message_processor;
mod codex_tool_config;
mod codex_tool_runner;
mod conversation_loop;
mod error_code;
mod exec_approval;
mod json_to_toml;
pub mod mcp_protocol;
pub(crate) mod message_processor;
mod outgoing_message;
mod patch_approval;
pub(crate) mod tool_handlers;
pub mod wire_format;
use crate::message_processor::MessageProcessor;

1054
codex-rs/mcp-server/src/mcp_protocol.rs
View File

File diff suppressed because it is too large Load Diff

70
codex-rs/mcp-server/src/message_processor.rs
View File

@@ -1,5 +1,4 @@
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::Arc;
@@ -9,18 +8,12 @@ use crate::codex_tool_config::CodexToolCallReplyParam;
use crate::codex_tool_config::create_tool_for_codex_tool_call_param;
use crate::codex_tool_config::create_tool_for_codex_tool_call_reply_param;
use crate::error_code::INVALID_REQUEST_ERROR_CODE;
use crate::mcp_protocol::ToolCallRequestParams;
use crate::mcp_protocol::ToolCallResponse;
use crate::mcp_protocol::ToolCallResponseResult;
use crate::outgoing_message::OutgoingMessageSender;
use crate::tool_handlers::create_conversation::handle_create_conversation;
use crate::tool_handlers::send_message::handle_send_message;
use crate::wire_format::ClientRequest;
use codex_core::ConversationManager;
use codex_core::config::Config as CodexConfig;
use codex_core::protocol::Submission;
use mcp_types::CallToolRequest;
use mcp_types::CallToolRequestParams;
use mcp_types::CallToolResult;
use mcp_types::ClientRequest as McpClientRequest;
@@ -48,7 +41,7 @@ pub(crate) struct MessageProcessor {
codex_linux_sandbox_exe: Option<PathBuf>,
conversation_manager: Arc<ConversationManager>,
running_requests_id_to_codex_uuid: Arc<Mutex<HashMap<RequestId, Uuid>>>,
running_session_ids: Arc<Mutex<HashSet<Uuid>>>,
// Tracks in-flight MCP request IDs to Codex session UUIDs.
}
impl MessageProcessor {
@@ -72,22 +65,9 @@ impl MessageProcessor {
codex_linux_sandbox_exe,
conversation_manager,
running_requests_id_to_codex_uuid: Arc::new(Mutex::new(HashMap::new())),
running_session_ids: Arc::new(Mutex::new(HashSet::new())),
}
}
pub(crate) fn get_conversation_manager(&self) -> &ConversationManager {
&self.conversation_manager
}
pub(crate) fn outgoing(&self) -> Arc<OutgoingMessageSender> {
self.outgoing.clone()
}
pub(crate) fn running_session_ids(&self) -> Arc<Mutex<HashSet<Uuid>>> {
self.running_session_ids.clone()
}
pub(crate) async fn process_request(&mut self, request: JSONRPCRequest) {
if let Ok(request_json) = serde_json::to_value(request.clone())
&& let Ok(codex_request) = serde_json::from_value::<ClientRequest>(request_json)
@@ -341,14 +321,6 @@ impl MessageProcessor {
params: <mcp_types::CallToolRequest as mcp_types::ModelContextProtocolRequest>::Params,
) {
tracing::info!("tools/call -> params: {:?}", params);
// Serialize params into JSON and try to parse as new type
if let Ok(new_params) =
serde_json::to_value(&params).and_then(serde_json::from_value::<ToolCallRequestParams>)
{
// New tool call matched → forward
self.handle_new_tool_calls(id, new_params).await;
return;
}
let CallToolRequestParams { name, arguments } = params;
match name.as_str() {
@@ -372,30 +344,6 @@ impl MessageProcessor {
}
}
}
async fn handle_new_tool_calls(&self, request_id: RequestId, params: ToolCallRequestParams) {
match params {
ToolCallRequestParams::ConversationCreate(args) => {
handle_create_conversation(self, request_id, args).await;
}
ToolCallRequestParams::ConversationSendMessage(args) => {
handle_send_message(self, request_id, args).await;
}
_ => {
let result = CallToolResult {
content: vec![ContentBlock::TextContent(TextContent {
r#type: "text".to_string(),
text: "Unknown tool".to_string(),
annotations: None,
})],
is_error: Some(true),
structured_content: None,
};
self.send_response::<CallToolRequest>(request_id, result)
.await;
}
}
}
async fn handle_tool_call_codex(&self, id: RequestId, arguments: Option<serde_json::Value>) {
let (initial_prompt, config): (String, CodexConfig) = match arguments {
Some(json_val) => match serde_json::from_value::<CodexToolCallParam>(json_val) {
@@ -692,20 +640,4 @@ impl MessageProcessor {
) {
tracing::info!("notifications/message -> params: {:?}", params);
}
pub(crate) async fn send_response_with_optional_error(
&self,
id: RequestId,
message: Option<ToolCallResponseResult>,
error: Option<bool>,
) {
let response = ToolCallResponse {
request_id: id.clone(),
is_error: error,
result: message,
};
let result: CallToolResult = response.into();
self.send_response::<mcp_types::CallToolRequest>(id.clone(), result)
.await;
}
}

127
codex-rs/mcp-server/src/tool_handlers/create_conversation.rs
View File

@@ -1,127 +0,0 @@
use std::path::PathBuf;
use codex_core::NewConversation;
use codex_core::config::Config as CodexConfig;
use codex_core::config::ConfigOverrides;
use mcp_types::RequestId;
use crate::conversation_loop::run_conversation_loop;
use crate::json_to_toml::json_to_toml;
use crate::mcp_protocol::ConversationCreateArgs;
use crate::mcp_protocol::ConversationCreateResult;
use crate::mcp_protocol::ConversationId;
use crate::mcp_protocol::ToolCallResponseResult;
use crate::message_processor::MessageProcessor;
pub(crate) async fn handle_create_conversation(
message_processor: &MessageProcessor,
id: RequestId,
args: ConversationCreateArgs,
) {
// Build ConfigOverrides from args
let ConversationCreateArgs {
prompt: _, // not used here; creation only establishes the session
model,
cwd,
approval_policy,
sandbox,
config,
profile,
base_instructions,
} = args;
// Convert config overrides JSON into CLI-style TOML overrides
let cli_overrides: Vec<(String, toml::Value)> = match config {
Some(v) => match v.as_object() {
Some(map) => map
.into_iter()
.map(|(k, v)| (k.clone(), json_to_toml(v.clone())))
.collect(),
None => Vec::new(),
},
None => Vec::new(),
};
let overrides = ConfigOverrides {
model: Some(model.clone()),
cwd: Some(PathBuf::from(cwd)),
approval_policy,
sandbox_mode: sandbox,
model_provider: None,
config_profile: profile,
codex_linux_sandbox_exe: None,
base_instructions,
include_plan_tool: None,
include_apply_patch_tool: None,
disable_response_storage: None,
show_raw_agent_reasoning: None,
};
let cfg: CodexConfig = match CodexConfig::load_with_cli_overrides(cli_overrides, overrides) {
Ok(cfg) => cfg,
Err(e) => {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationCreate(
ConversationCreateResult::Error {
message: format!("Failed to load config: {e}"),
},
)),
Some(true),
)
.await;
return;
}
};
// Initialize Codex session via server API
let NewConversation {
conversation_id: session_id,
conversation,
session_configured,
} = match message_processor
.get_conversation_manager()
.new_conversation(cfg)
.await
{
Ok(conv) => conv,
Err(e) => {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationCreate(
ConversationCreateResult::Error {
message: format!("Failed to initialize session: {e}"),
},
)),
Some(true),
)
.await;
return;
}
};
let effective_model = session_configured.model.clone();
// Run the conversation loop in the background so this request can return immediately.
let outgoing = message_processor.outgoing();
let spawn_id = id.clone();
tokio::spawn(async move {
run_conversation_loop(conversation.clone(), outgoing, spawn_id).await;
});
// Reply with the new conversation id and effective model
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationCreate(
ConversationCreateResult::Ok {
conversation_id: ConversationId(session_id),
model: effective_model,
},
)),
Some(false),
)
.await;
}

114
codex-rs/mcp-server/src/tool_handlers/send_message.rs
View File

@@ -1,114 +0,0 @@
use codex_core::protocol::Op;
use codex_core::protocol::Submission;
use mcp_types::RequestId;
use crate::mcp_protocol::ConversationSendMessageArgs;
use crate::mcp_protocol::ConversationSendMessageResult;
use crate::mcp_protocol::ToolCallResponseResult;
use crate::message_processor::MessageProcessor;
pub(crate) async fn handle_send_message(
message_processor: &MessageProcessor,
id: RequestId,
arguments: ConversationSendMessageArgs,
) {
let ConversationSendMessageArgs {
conversation_id,
content: items,
parent_message_id: _,
conversation_overrides: _,
} = arguments;
if items.is_empty() {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationSendMessage(
ConversationSendMessageResult::Error {
message: "No content items provided".to_string(),
},
)),
Some(true),
)
.await;
return;
}
let session_id = conversation_id.0;
let Ok(codex) = message_processor
.get_conversation_manager()
.get_conversation(session_id)
.await
else {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationSendMessage(
ConversationSendMessageResult::Error {
message: "Session does not exist".to_string(),
},
)),
Some(true),
)
.await;
return;
};
let running = {
let running_sessions = message_processor.running_session_ids();
let mut running_sessions = running_sessions.lock().await;
!running_sessions.insert(session_id)
};
if running {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationSendMessage(
ConversationSendMessageResult::Error {
message: "Session is already running".to_string(),
},
)),
Some(true),
)
.await;
return;
}
let request_id_string = match &id {
RequestId::String(s) => s.clone(),
RequestId::Integer(i) => i.to_string(),
};
let submit_res = codex
.submit_with_id(Submission {
id: request_id_string,
op: Op::UserInput { items },
})
.await;
if let Err(e) = submit_res {
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationSendMessage(
ConversationSendMessageResult::Error {
message: format!("Failed to submit user input: {e}"),
},
)),
Some(true),
)
.await;
return;
}
message_processor
.send_response_with_optional_error(
id,
Some(ToolCallResponseResult::ConversationSendMessage(
ConversationSendMessageResult::Ok,
)),
Some(false),
)
.await;
}

88
codex-rs/mcp-server/tests/common/mcp_process.rs
View File

@@ -11,13 +11,8 @@ use tokio::process::ChildStdout;
use anyhow::Context;
use assert_cmd::prelude::*;
use codex_core::protocol::InputItem;
use codex_mcp_server::CodexToolCallParam;
use codex_mcp_server::CodexToolCallReplyParam;
use codex_mcp_server::mcp_protocol::ConversationCreateArgs;
use codex_mcp_server::mcp_protocol::ConversationId;
use codex_mcp_server::mcp_protocol::ConversationSendMessageArgs;
use codex_mcp_server::mcp_protocol::ToolCallRequestParams;
use codex_mcp_server::wire_format::AddConversationListenerParams;
use codex_mcp_server::wire_format::NewConversationParams;
use codex_mcp_server::wire_format::RemoveConversationListenerParams;
@@ -40,7 +35,6 @@ use pretty_assertions::assert_eq;
use serde_json::json;
use std::process::Command as StdCommand;
use tokio::process::Command;
use uuid::Uuid;
pub struct McpProcess {
next_request_id: AtomicI64,
@@ -186,60 +180,7 @@ impl McpProcess {
.await
}
pub async fn send_user_message_tool_call(
&mut self,
message: &str,
session_id: &str,
) -> anyhow::Result<i64> {
let params = ToolCallRequestParams::ConversationSendMessage(ConversationSendMessageArgs {
conversation_id: ConversationId(Uuid::parse_str(session_id)?),
content: vec![InputItem::Text {
text: message.to_string(),
}],
parent_message_id: None,
conversation_overrides: None,
});
self.send_request(
mcp_types::CallToolRequest::METHOD,
Some(serde_json::to_value(params)?),
)
.await
}
pub async fn send_conversation_create_tool_call(
&mut self,
prompt: &str,
model: &str,
cwd: &str,
) -> anyhow::Result<i64> {
let params = ToolCallRequestParams::ConversationCreate(ConversationCreateArgs {
prompt: prompt.to_string(),
model: model.to_string(),
cwd: cwd.to_string(),
approval_policy: None,
sandbox: None,
config: None,
profile: None,
base_instructions: None,
});
self.send_request(
mcp_types::CallToolRequest::METHOD,
Some(serde_json::to_value(params)?),
)
.await
}
pub async fn send_conversation_create_with_args(
&mut self,
args: ConversationCreateArgs,
) -> anyhow::Result<i64> {
let params = ToolCallRequestParams::ConversationCreate(args);
self.send_request(
mcp_types::CallToolRequest::METHOD,
Some(serde_json::to_value(params)?),
)
.await
}
// Deprecated tool-call helpers removed. Use newConversation/sendUserMessage APIs instead.
// ---------------------------------------------------------------------
// Codex JSON-RPC (non-tool) helpers
@@ -384,6 +325,33 @@ impl McpProcess {
}
}
pub async fn read_stream_until_error_message(
&mut self,
request_id: RequestId,
) -> anyhow::Result<mcp_types::JSONRPCError> {
loop {
let message = self.read_jsonrpc_message().await?;
eprint!("message: {message:?}");
match message {
JSONRPCMessage::Notification(_) => {
eprintln!("notification: {message:?}");
}
JSONRPCMessage::Request(_) => {
anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}");
}
JSONRPCMessage::Response(_) => {
// Keep scanning; we're waiting for an error with matching id.
}
JSONRPCMessage::Error(err) => {
if err.id == request_id {
return Ok(err);
}
}
}
}
}
pub async fn read_stream_until_notification_message(
&mut self,
method: &str,

84
codex-rs/mcp-server/tests/create_conversation.rs
View File

@@ -1,5 +1,12 @@
use std::path::Path;
use codex_mcp_server::wire_format::AddConversationListenerParams;
use codex_mcp_server::wire_format::AddConversationSubscriptionResponse;
use codex_mcp_server::wire_format::InputItem;
use codex_mcp_server::wire_format::NewConversationParams;
use codex_mcp_server::wire_format::NewConversationResponse;
use codex_mcp_server::wire_format::SendUserMessageParams;
use codex_mcp_server::wire_format::SendUserMessageResponse;
use mcp_test_support::McpProcess;
use mcp_test_support::create_final_assistant_message_sse_response;
use mcp_test_support::create_mock_chat_completions_server;
@@ -33,43 +40,64 @@ async fn test_conversation_create_and_send_message_ok() {
.expect("init timeout")
.expect("init failed");
// Create a conversation via the new tool.
let req_id = mcp
.send_conversation_create_tool_call("", "o3", "/repo")
// Create a conversation via the new JSON-RPC API.
let new_conv_id = mcp
.send_new_conversation_request(NewConversationParams {
model: Some("o3".to_string()),
..Default::default()
})
.await
.expect("send conversationCreate");
let resp: JSONRPCResponse = timeout(
.expect("send newConversation");
let new_conv_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
mcp.read_stream_until_response_message(RequestId::Integer(new_conv_id)),
)
.await
.expect("create response timeout")
.expect("create response error");
.expect("newConversation timeout")
.expect("newConversation resp");
let NewConversationResponse {
conversation_id,
model,
} = to_response::<NewConversationResponse>(new_conv_resp)
.expect("deserialize newConversation response");
assert_eq!(model, "o3");
// Structured content must include status=ok, a UUID conversation_id and the model we passed.
let sc = &resp.result["structuredContent"];
let conv_id = sc["conversation_id"].as_str().expect("uuid string");
assert!(!conv_id.is_empty());
assert_eq!(sc["model"], json!("o3"));
// Now send a message to the created conversation and expect an OK result.
let send_id = mcp
.send_user_message_tool_call("Hello", conv_id)
// Add a listener so we receive notifications for this conversation (not strictly required for this test).
let add_listener_id = mcp
.send_add_conversation_listener_request(AddConversationListenerParams { conversation_id })
.await
.expect("send message");
.expect("send addConversationListener");
let _sub: AddConversationSubscriptionResponse =
to_response::<AddConversationSubscriptionResponse>(
timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(add_listener_id)),
)
.await
.expect("addConversationListener timeout")
.expect("addConversationListener resp"),
)
.expect("deserialize addConversationListener response");
// Now send a user message via the wire API and expect an OK (empty object) result.
let send_id = mcp
.send_send_user_message_request(SendUserMessageParams {
conversation_id,
items: vec![InputItem::Text {
text: "Hello".to_string(),
}],
})
.await
.expect("send sendUserMessage");
let send_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(send_id)),
)
.await
.expect("send response timeout")
.expect("send response error");
assert_eq!(
send_resp.result["structuredContent"],
json!({ "status": "ok" })
);
.expect("sendUserMessage timeout")
.expect("sendUserMessage resp");
let _ok: SendUserMessageResponse = to_response::<SendUserMessageResponse>(send_resp)
.expect("deserialize sendUserMessage response");
// avoid race condition by waiting for the mock server to receive the chat.completions request
let deadline = std::time::Instant::now() + DEFAULT_READ_TIMEOUT;
@@ -101,6 +129,12 @@ async fn test_conversation_create_and_send_message_ok() {
drop(server);
}
fn to_response<T: serde::de::DeserializeOwned>(response: JSONRPCResponse) -> anyhow::Result<T> {
let value = serde_json::to_value(response.result)?;
let codex_response = serde_json::from_value(value)?;
Ok(codex_response)
}
// Helper to create a config.toml pointing at the mock model server.
fn create_config_toml(codex_home: &Path, server_uri: &str) -> std::io::Result<()> {
let config_toml = codex_home.join("config.toml");

110
codex-rs/mcp-server/tests/send_message.rs
View File

@@ -2,15 +2,18 @@ use std::path::Path;
use std::thread::sleep;
use std::time::Duration;
use codex_mcp_server::CodexToolCallParam;
use codex_mcp_server::wire_format::ConversationId;
use codex_mcp_server::wire_format::InputItem;
use codex_mcp_server::wire_format::NewConversationParams;
use codex_mcp_server::wire_format::NewConversationResponse;
use codex_mcp_server::wire_format::SendUserMessageParams;
use codex_mcp_server::wire_format::SendUserMessageResponse;
use mcp_test_support::McpProcess;
use mcp_test_support::create_final_assistant_message_sse_response;
use mcp_test_support::create_mock_chat_completions_server;
use mcp_types::JSONRPC_VERSION;
use mcp_types::JSONRPCResponse;
use mcp_types::RequestId;
use pretty_assertions::assert_eq;
use serde_json::json;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -39,63 +42,44 @@ async fn test_send_message_success() {
.expect("init timed out")
.expect("init failed");
// Kick off a Codex session so we have a valid session_id.
let codex_request_id = mcp_process
.send_codex_tool_call(CodexToolCallParam {
prompt: "Start a session".to_string(),
..Default::default()
})
// Start a conversation using the new wire API.
let new_conv_id = mcp_process
.send_new_conversation_request(NewConversationParams::default())
.await
.expect("send codex tool call");
// Wait for the session_configured event to get the session_id.
let session_id = mcp_process
.read_stream_until_configured_response_message()
.await
.expect("read session_configured");
// The original codex call will finish quickly given our mock; consume its response.
timeout(
.expect("send newConversation");
let new_conv_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_response_message(RequestId::Integer(codex_request_id)),
mcp_process.read_stream_until_response_message(RequestId::Integer(new_conv_id)),
)
.await
.expect("codex response timeout")
.expect("codex response error");
.expect("newConversation timeout")
.expect("newConversation resp");
let NewConversationResponse {
conversation_id, ..
} = to_response::<NewConversationResponse>(new_conv_resp)
.expect("deserialize newConversation response");
// Now exercise the send-user-message tool.
let send_msg_request_id = mcp_process
.send_user_message_tool_call("Hello again", &session_id)
// Now exercise sendUserMessage.
let send_id = mcp_process
.send_send_user_message_request(SendUserMessageParams {
conversation_id,
items: vec![InputItem::Text {
text: "Hello again".to_string(),
}],
})
.await
.expect("send send-message tool call");
.expect("send sendUserMessage");
let response: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp_process.read_stream_until_response_message(RequestId::Integer(send_msg_request_id)),
mcp_process.read_stream_until_response_message(RequestId::Integer(send_id)),
)
.await
.expect("send-user-message response timeout")
.expect("send-user-message response error");
.expect("sendUserMessage response timeout")
.expect("sendUserMessage response error");
assert_eq!(
JSONRPCResponse {
jsonrpc: JSONRPC_VERSION.into(),
id: RequestId::Integer(send_msg_request_id),
result: json!({
"content": [
{
"text": "{\"status\":\"ok\"}",
"type": "text",
}
],
"isError": false,
"structuredContent": {
"status": "ok"
}
}),
},
response
);
let _ok: SendUserMessageResponse = to_response::<SendUserMessageResponse>(response)
.expect("deserialize sendUserMessage response");
// wait for the server to hear the user message
sleep(Duration::from_secs(5));
@@ -113,24 +97,26 @@ async fn test_send_message_session_not_found() {
.expect("timeout")
.expect("init");
let unknown = uuid::Uuid::new_v4().to_string();
let unknown = ConversationId(uuid::Uuid::new_v4());
let req_id = mcp
.send_user_message_tool_call("ping", &unknown)
.send_send_user_message_request(SendUserMessageParams {
conversation_id: unknown,
items: vec![InputItem::Text {
text: "ping".to_string(),
}],
})
.await
.expect("send tool");
.expect("send sendUserMessage");
let resp: JSONRPCResponse = timeout(
// Expect an error response for unknown conversation.
let err = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
mcp.read_stream_until_error_message(RequestId::Integer(req_id)),
)
.await
.expect("timeout")
.expect("resp");
let result = resp.result.clone();
let content = result["content"][0]["text"].as_str().unwrap_or("");
assert!(content.contains("Session does not exist"));
assert_eq!(result["isError"], json!(true));
.expect("error");
assert_eq!(err.id, RequestId::Integer(req_id));
}
// ---------------------------------------------------------------------------
@@ -159,3 +145,9 @@ stream_max_retries = 0
),
)
}
fn to_response<T: serde::de::DeserializeOwned>(response: JSONRPCResponse) -> anyhow::Result<T> {
let value = serde_json::to_value(response.result)?;
let codex_response = serde_json::from_value(value)?;
Ok(codex_response)
}

38
codex-rs/tui/src/bottom_pane/chat_composer_history.rs
View File

@@ -55,11 +55,18 @@ impl ChatComposerHistory {
/// Record a message submitted by the user in the current session so it can
/// be recalled later.
pub fn record_local_submission(&mut self, text: &str) {
if !text.is_empty() {
self.local_history.push(text.to_string());
self.history_cursor = None;
self.last_history_text = None;
if text.is_empty() {
return;
}
// Avoid inserting a duplicate if identical to the previous entry.
if self.local_history.last().is_some_and(|prev| prev == text) {
return;
}
self.local_history.push(text.to_string());
self.history_cursor = None;
self.last_history_text = None;
}
/// Should Up/Down key presses be interpreted as history navigation given
@@ -187,6 +194,29 @@ mod tests {
use codex_core::protocol::Op;
use std::sync::mpsc::channel;
#[test]
fn duplicate_submissions_are_not_recorded() {
let mut history = ChatComposerHistory::new();
// Empty submissions are ignored.
history.record_local_submission("");
assert_eq!(history.local_history.len(), 0);
// First entry is recorded.
history.record_local_submission("hello");
assert_eq!(history.local_history.len(), 1);
assert_eq!(history.local_history.last().unwrap(), "hello");
// Identical consecutive entry is skipped.
history.record_local_submission("hello");
assert_eq!(history.local_history.len(), 1);
// Different entry is recorded.
history.record_local_submission("world");
assert_eq!(history.local_history.len(), 2);
assert_eq!(history.local_history.last().unwrap(), "world");
}
#[test]
fn navigation_with_async_fetch() {
let (tx, rx) = channel::<AppEvent>();

28
codex-rs/tui/src/history_cell.rs
View File

@@ -818,8 +818,32 @@ pub(crate) fn new_patch_apply_success(stdout: String) -> PlainHistoryCell {
let mut iter = stdout.lines();
for (i, raw) in iter.by_ref().take(TOOL_CALL_MAX_LINES).enumerate() {
let prefix = if i == 0 { " └ " } else { " " };
let s = format!("{prefix}{raw}");
lines.push(ansi_escape_line(&s).dim());
// First line is the header; dim it entirely.
if i == 0 {
let s = format!("{prefix}{raw}");
lines.push(ansi_escape_line(&s).dim());
continue;
}
// Subsequent lines should look like: "M path/to/file".
// Colorize the status letter like `git status` (e.g., M red).
let status = raw.chars().next();
let rest = raw.get(1..).unwrap_or("");
let status_span = match status {
Some('M') => "M".red(),
Some('A') => "A".green(),
Some('D') => "D".red(),
Some(other) => other.to_string().into(),
None => "".into(),
};
lines.push(Line::from(vec![
prefix.into(),
status_span,
ansi_escape_line(rest).to_string().into(),
]));
}
let remaining = iter.count();
if remaining > 0 {