From 16236b699ef4aec8a320f2d4712a946856fdb67f Mon Sep 17 00:00:00 2001 From: jimmyfraiture Date: Sun, 7 Sep 2025 16:53:29 -0700 Subject: [PATCH] V4 --- codex-rs/core/src/codex.rs | 153 +++++++++++++++++----------- codex-rs/core/src/ishell.rs | 160 +++++++++++++++++++++++++----- codex-rs/core/src/openai_tools.rs | 38 ++++--- codex-rs/protocol/src/models.rs | 2 +- 4 files changed, 252 insertions(+), 101 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index efa089cdee..b69c74ae6d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -21,6 +21,7 @@ use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; use futures::prelude::*; use mcp_types::CallToolResult; +use serde::Deserialize; use serde::Serialize; use serde_json; use tokio::sync::oneshot; @@ -1997,62 +1998,10 @@ async fn handle_response_item( .clone() .or_else(|| session_id.clone()) .unwrap_or_else(|| format!("ishell:{}", Uuid::new_v4())); - let parsed_session_id = match session_id { - Some(value) => match value.parse::() { - Ok(parsed) => Some(parsed), - Err(_) => { - return Ok(Some(ResponseInputItem::FunctionCallOutput { - call_id, - output: FunctionCallOutputPayload { - content: format!("invalid session_id: {value}"), - success: Some(false), - }, - })); - } - }, - None => None, - }; - - let request = crate::ishell::InteractiveShellRequest { - session_id: parsed_session_id, - input: &arguments, - timeout_ms, - }; - - let result = sess.ishell_manager.handle_request(request).await; - - let output_payload = match result { - Ok(value) => { - #[derive(serde::Serialize)] - struct SerializedIshellResult<'a> { - session_id: Option, - output: &'a str, - } - - match serde_json::to_string(&SerializedIshellResult { - session_id: value.session_id, - output: &value.output, - }) { - Ok(serialized) => FunctionCallOutputPayload { - content: serialized, - success: Some(true), - }, - Err(err) => FunctionCallOutputPayload { - content: format!("failed to serialize interactive shell output: {err}"), - success: Some(false), - }, - } - } - Err(err) => FunctionCallOutputPayload { - content: err.to_string(), - success: Some(false), - }, - }; - - Some(ResponseInputItem::FunctionCallOutput { - call_id, - output: output_payload, - }) + Some( + handle_ishell_tool_call(sess, call_id, session_id.clone(), arguments, timeout_ms) + .await, + ) } ResponseItem::CustomToolCall { id: _, @@ -2098,6 +2047,72 @@ async fn handle_response_item( Ok(output) } + +async fn handle_ishell_tool_call( + sess: &Session, + call_id: String, + session_id: Option, + arguments: Vec, + timeout_ms: Option, +) -> ResponseInputItem { + + let parsed_session_id = if let Some(session_id) = session_id { + match session_id.parse::() { + Ok(parsed) => Some(parsed), + Err(output) => return ResponseInputItem::FunctionCallOutput { + call_id: call_id.to_string(), + output: FunctionCallOutputPayload { + content: format!("invalid session_id: {session_id} due to error {output}"), + success: Some(false), + }, + } + } + } else { + None + }; + + let request = crate::ishell::InteractiveShellRequest { + session_id: parsed_session_id, + input_chunks: &arguments, + timeout_ms, + }; + + let result = sess.ishell_manager.handle_request(request).await; + + let output_payload = match result { + Ok(value) => { + #[derive(Serialize)] + struct SerializedIshellResult<'a> { + session_id: Option, + output: &'a str, + } + + match serde_json::to_string(&SerializedIshellResult { + session_id: value.session_id, + output: &value.output, + }) { + Ok(serialized) => FunctionCallOutputPayload { + content: serialized, + success: Some(true), + }, + Err(err) => FunctionCallOutputPayload { + content: format!("failed to serialize interactive shell output: {err}"), + success: Some(false), + }, + } + } + Err(err) => FunctionCallOutputPayload { + content: err.to_string(), + success: Some(false), + }, + }; + + ResponseInputItem::FunctionCallOutput { + call_id, + output: output_payload, + } +} + async fn handle_function_call( sess: &Session, turn_context: &TurnContext, @@ -2125,6 +2140,32 @@ async fn handle_function_call( ) .await } + "unified_exec" => { + #[derive(Deserialize)] + struct UnifiedExecArgs { + input: Vec, + #[serde(default)] + session_id: Option, + #[serde(default)] + timeout_ms: Option, + } + + let args = match serde_json::from_str::(&arguments) { + Ok(args) => args, + Err(err) => { + return ResponseInputItem::FunctionCallOutput { + call_id, + output: FunctionCallOutputPayload { + content: format!("failed to parse function arguments: {err}"), + success: Some(false), + }, + }; + } + }; + + handle_ishell_tool_call(sess, call_id, args.session_id, args.input, args.timeout_ms) + .await + } "view_image" => { #[derive(serde::Deserialize)] struct SeeImageArgs { diff --git a/codex-rs/core/src/ishell.rs b/codex-rs/core/src/ishell.rs index d04e125460..5290c27c57 100644 --- a/codex-rs/core/src/ishell.rs +++ b/codex-rs/core/src/ishell.rs @@ -3,8 +3,11 @@ use portable_pty::PtySize; use portable_pty::native_pty_system; use std::collections::HashMap; use std::collections::VecDeque; +use std::env; use std::io::ErrorKind; use std::io::Read; +use std::path::Path; +use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex as StdMutex; use std::sync::atomic::AtomicBool; @@ -28,7 +31,7 @@ const ISHELL_OUTPUT_MAX_BYTES: usize = 16 * 1024; // 16 KiB #[derive(Debug)] pub(crate) struct InteractiveShellRequest<'a> { pub session_id: Option, - pub input: &'a str, + pub input_chunks: &'a [String], pub timeout_ms: Option, } @@ -106,6 +109,7 @@ impl InteractiveShellSessionManager { &self, request: InteractiveShellRequest<'_>, ) -> Result { + tracing::error!("In the exec"); // todo update the errors let timeout_ms = request.timeout_ms.unwrap_or(DEFAULT_TIMEOUT_MS); @@ -132,7 +136,7 @@ impl InteractiveShellSessionManager { } } } else { - let command = parse_command_line(request.input)?; + let command = command_from_chunks(request.input_chunks)?; let new_id = self.next_session_id.fetch_add(1, Ordering::SeqCst); let session = create_shell_session(&command).await?; let managed_session = ManagedInteractiveSession::new(session); @@ -144,14 +148,12 @@ impl InteractiveShellSessionManager { new_session = Some(managed_session); }; - if request.session_id.is_some() - && !request.input.is_empty() - && writer_tx - .send(request.input.as_bytes().to_vec()) - .await - .is_err() - { - return Err(error::InteractiveShellError::WriteToStdin); + if request.session_id.is_some() { + let joined_input = join_input_chunks(request.input_chunks); + if !joined_input.is_empty() && writer_tx.send(joined_input.into_bytes()).await.is_err() + { + return Err(error::InteractiveShellError::WriteToStdin); + } } let mut collected: Vec = Vec::with_capacity(4096); @@ -249,7 +251,8 @@ async fn create_shell_session( }) .map_err(error::InteractiveShellError::create_session)?; - let mut command_builder = CommandBuilder::new(&command[0]); + let resolved_command = resolve_command_path(&command[0])?; + let mut command_builder = CommandBuilder::new(&resolved_command); for arg in &command[1..] { command_builder.arg(arg); } @@ -326,6 +329,106 @@ async fn create_shell_session( )) } +fn resolve_command_path(command: &str) -> Result { + if command.is_empty() { + return Err(error::InteractiveShellError::MissingCommandLine); + } + + if is_explicit_path(command) { + return ensure_executable(Path::new(command)) + .then_some(command.to_string()) + .ok_or_else(|| error::InteractiveShellError::CommandNotFound { + command: command.to_string(), + }); + } + + if let Some(resolved) = find_in_path(command) { + return Ok(resolved.to_string_lossy().to_string()); + } + + Err(error::InteractiveShellError::CommandNotFound { + command: command.to_string(), + }) +} + +fn command_from_chunks(chunks: &[String]) -> Result, error::InteractiveShellError> { + match chunks { + [] => Err(error::InteractiveShellError::MissingCommandLine), + [single] => parse_command_line(single), + _ => Ok(chunks.to_vec()), + } +} + +fn join_input_chunks(chunks: &[String]) -> String { + match chunks { + [] => String::new(), + [single] => single.clone(), + _ => chunks.concat(), + } +} + +fn is_explicit_path(command: &str) -> bool { + let path = Path::new(command); + path.is_absolute() || path.components().count() > 1 +} + +fn find_in_path(command: &str) -> Option { + let path_var = env::var_os("PATH")?; + env::split_paths(&path_var) + .flat_map(|dir| candidate_paths(dir, command)) + .find(|candidate| ensure_executable(candidate)) +} + +fn candidate_paths(dir: PathBuf, command: &str) -> Vec { + build_platform_candidates(dir.join(command)) +} + +#[cfg(unix)] +fn build_platform_candidates(candidate: PathBuf) -> Vec { + vec![candidate] +} + +#[cfg(windows)] +fn build_platform_candidates(candidate: PathBuf) -> Vec { + if candidate.extension().is_some() { + return vec![candidate]; + } + + let pathext = env::var("PATHEXT").unwrap_or_else(|_| ".COM;.EXE;.BAT;.CMD".to_string()); + let mut candidates = Vec::new(); + for ext in pathext.split(';') { + if ext.is_empty() { + continue; + } + let mut path_with_ext = candidate.clone(); + let new_ext = ext.trim_start_matches('.'); + path_with_ext.set_extension(new_ext); + candidates.push(path_with_ext); + } + if candidates.is_empty() { + candidates.push(candidate); + } + candidates +} + +fn ensure_executable(path: &Path) -> bool { + match path.metadata() { + Ok(metadata) => metadata.is_file() && is_executable(&metadata), + Err(_) => false, + } +} + +#[cfg(unix)] +fn is_executable(metadata: &std::fs::Metadata) -> bool { + use std::os::unix::fs::PermissionsExt; + metadata.permissions().mode() & 0o111 != 0 +} + +#[cfg(not(unix))] +fn is_executable(metadata: &std::fs::Metadata) -> bool { + metadata.is_file() +} + /// Truncate the middle of a UTF-8 string to at most `max_bytes` bytes, /// preserving the beginning and the end. Returns the possibly truncated /// string and `Some(original_token_count)` (estimated at 4 bytes/token) @@ -437,6 +540,8 @@ mod error { MissingCommandLine, #[error("invalid command line: {command_line}")] InvalidCommandLine { command_line: String }, + #[error("command not found: {command}")] + CommandNotFound { command: String }, } impl InteractiveShellError { @@ -486,7 +591,7 @@ mod tests { let open_shell = manager .handle_request(InteractiveShellRequest { session_id: None, - input: "/bin/bash -i", + input_chunks: &["/bin/bash".to_string(), "-i".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -495,7 +600,7 @@ mod tests { manager .handle_request(InteractiveShellRequest { session_id: Some(session_id), - input: "export CODEX_INTERACTIVE_SHELL_VAR=codex\n", + input_chunks: &["export CODEX_INTERACTIVE_SHELL_VAR=codex\n".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -503,7 +608,7 @@ mod tests { let out_2 = manager .handle_request(InteractiveShellRequest { session_id: Some(session_id), - input: "echo $CODEX_INTERACTIVE_SHELL_VAR\n", + input_chunks: &["echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -523,7 +628,7 @@ mod tests { let shell_a = manager .handle_request(InteractiveShellRequest { session_id: None, - input: "/bin/bash -i", + input_chunks: &["/bin/bash".to_string(), "-i".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -532,7 +637,7 @@ mod tests { manager .handle_request(InteractiveShellRequest { session_id: Some(session_a), - input: "export CODEX_INTERACTIVE_SHELL_VAR=codex\n", + input_chunks: &["export CODEX_INTERACTIVE_SHELL_VAR=codex\n".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -540,7 +645,7 @@ mod tests { let out_2 = manager .handle_request(InteractiveShellRequest { session_id: None, - input: "/bin/echo $CODEX_INTERACTIVE_SHELL_VAR\n", + input_chunks: &["/bin/echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -549,7 +654,7 @@ mod tests { let out_3 = manager .handle_request(InteractiveShellRequest { session_id: Some(session_a), - input: "echo $CODEX_INTERACTIVE_SHELL_VAR\n", + input_chunks: &["echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -558,6 +663,8 @@ mod tests { Ok(()) } + // todo add a test for the path + #[cfg(unix)] /// Confirms that output emitted after an initial request times out can be /// collected by a follow-up request against the same session. @@ -568,7 +675,7 @@ mod tests { let open_shell = manager .handle_request(InteractiveShellRequest { session_id: None, - input: "/bin/bash -i", + input_chunks: &["/bin/bash".to_string(), "-i".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -577,7 +684,7 @@ mod tests { manager .handle_request(InteractiveShellRequest { session_id: Some(session_id), - input: "export CODEX_INTERACTIVE_SHELL_VAR=codex\n", + input_chunks: &["export CODEX_INTERACTIVE_SHELL_VAR=codex\n".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -585,7 +692,7 @@ mod tests { let out_2 = manager .handle_request(InteractiveShellRequest { session_id: Some(session_id), - input: "sleep 5 && echo $CODEX_INTERACTIVE_SHELL_VAR\n", + input_chunks: &["sleep 5 && echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()], timeout_ms: Some(10), }) .await?; @@ -594,10 +701,11 @@ mod tests { // Wait for the end of the bash sleep (preventing the usage of tokio controlled clock). tokio::time::sleep(Duration::from_secs(7)).await; + let empty = Vec::new(); let out_3 = manager .handle_request(InteractiveShellRequest { session_id: Some(session_id), - input: "", + input_chunks: &empty, timeout_ms: Some(100), }) .await?; @@ -609,13 +717,16 @@ mod tests { #[cfg(unix)] #[tokio::test] + /// Ensure that commands which immediately complete do not create persistent + /// interactive shell sessions, preventing leftover empty sessions from + /// accumulating. async fn completed_commands_do_not_persist_sessions() -> Result<(), error::InteractiveShellError> { let manager = InteractiveShellSessionManager::default(); let result = manager .handle_request(InteractiveShellRequest { session_id: None, - input: "/bin/echo codex", + input_chunks: &["/bin/echo".to_string(), "codex".to_string()], timeout_ms: Some(1_500), }) .await?; @@ -630,12 +741,13 @@ mod tests { #[test] fn truncate_middle_no_newlines_fallback() { + // todo double check the truncation logic // Long string without newlines forces a pure byte/char-boundary truncation. let s = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ"; let max_bytes = 16; // force truncation let (out, original) = truncate_middle(s, max_bytes); // For very small caps, we return a full, untruncated marker that may exceed the cap. - assert_eq!(out, "…16 tokens truncated…"); + assert_eq!(out, "…16 tokens truncated…"); // todo rename the token word // Original is ceil(62/4) = 16 tokens. assert_eq!(original, Some(16)); } diff --git a/codex-rs/core/src/openai_tools.rs b/codex-rs/core/src/openai_tools.rs index 7cd18f0ad2..13b4e6d4be 100644 --- a/codex-rs/core/src/openai_tools.rs +++ b/codex-rs/core/src/openai_tools.rs @@ -204,11 +204,12 @@ fn create_unified_exec_tool() -> OpenAiTool { let mut properties = BTreeMap::new(); properties.insert( "input".to_string(), - JsonSchema::String { + JsonSchema::Array { + items: Box::new(JsonSchema::String { description: None }), description: Some( - "When no session_id is provided, the entire input is treated as the command \ - to launch. When a session_id is present, this string is written to the \ - session's stdin." + "When no session_id is provided, treat the array as the command and arguments \ + to launch. When session_id is set, concatenate the strings (in order) and write \ + them to the session's stdin." .to_string(), ), }, @@ -625,22 +626,19 @@ pub(crate) fn get_openai_tools( if config.include_view_image_tool { tools.push(create_view_image_tool()); } - // - // if let Some(mcp_tools) = mcp_tools { - // // Ensure deterministic ordering to maximize prompt cache hits. - // // HashMap iteration order is non-deterministic, so sort by fully-qualified tool name. - // let mut entries: Vec<(String, mcp_types::Tool)> = mcp_tools.into_iter().collect(); - // entries.sort_by(|a, b| a.0.cmp(&b.0)); - // - // for (name, tool) in entries.into_iter() { - // match mcp_tool_to_openai_tool(name.clone(), tool.clone()) { - // Ok(converted_tool) => tools.push(OpenAiTool::Function(converted_tool)), - // Err(e) => { - // tracing::error!("Failed to convert {name:?} MCP tool to OpenAI tool: {e:?}"); - // } - // } - // } - // } + if let Some(mcp_tools) = mcp_tools { + let mut entries: Vec<(String, mcp_types::Tool)> = mcp_tools.into_iter().collect(); + entries.sort_by(|a, b| a.0.cmp(&b.0)); + + for (name, tool) in entries.into_iter() { + match mcp_tool_to_openai_tool(name.clone(), tool.clone()) { + Ok(converted_tool) => tools.push(OpenAiTool::Function(converted_tool)), + Err(e) => { + tracing::error!("Failed to convert {name:?} MCP tool to OpenAI tool: {e:?}"); + } + } + } + } tools } diff --git a/codex-rs/protocol/src/models.rs b/codex-rs/protocol/src/models.rs index 4dabee1fcf..9e3f4ec87f 100644 --- a/codex-rs/protocol/src/models.rs +++ b/codex-rs/protocol/src/models.rs @@ -122,7 +122,7 @@ pub enum ResponseItem { session_id: Option, /// Characters that should be written to the interactive shell's /// standard input. - arguments: String, + arguments: Vec, /// Maximum amount of time to wait for additional output after writing /// to stdin. timeout_ms: Option,