Drop old shells

codex-rs/core/src/chat_completions.rs

@@ -56,9 +56,9 @@ pub(crate) async fn stream_chat_completions(
     for item in &input {
         match item {
             ResponseItem::Message { role, .. } => last_emitted_role = Some(role.as_str()),
-            ResponseItem::FunctionCall { .. } | ResponseItem::LocalShellCall { .. } => {
-                last_emitted_role = Some("assistant")
-            }
+            ResponseItem::FunctionCall { .. }
+            | ResponseItem::LocalShellCall { .. }
+            | ResponseItem::UnifiedExec { .. } => last_emitted_role = Some("assistant"),
             ResponseItem::FunctionCallOutput { .. } => last_emitted_role = Some("tool"),
             ResponseItem::Reasoning { .. } | ResponseItem::Other => {}
             ResponseItem::CustomToolCall { .. } => {}
@@ -119,7 +119,9 @@ pub(crate) async fn stream_chat_completions(
                 // Otherwise, attach to immediate next assistant anchor (tool-calls or assistant message)
                 if !attached && idx + 1 < input.len() {
                     match &input[idx + 1] {
-                        ResponseItem::FunctionCall { .. } | ResponseItem::LocalShellCall { .. } => {
+                        ResponseItem::FunctionCall { .. }
+                        | ResponseItem::LocalShellCall { .. }
+                        | ResponseItem::UnifiedExec { .. } => {
                             reasoning_by_anchor_index
                                 .entry(idx + 1)
                                 .and_modify(|v| v.push_str(&text))
@@ -224,6 +226,12 @@ pub(crate) async fn stream_chat_completions(
                 }
                 messages.push(msg);
             }
+            ResponseItem::UnifiedExec { .. } => {
+                // The interactive shell tool is currently exposed only via the
+                // Responses API. Ignore these items when using Chat
+                // Completions to maintain feature parity with existing
+                // behaviour.
+            }
             ResponseItem::FunctionCallOutput { call_id, output } => {
                 messages.push(json!({
                     "role": "tool",

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;
@@ -107,6 +108,7 @@ use crate::safety::assess_command_safety;
 use crate::safety::assess_safety_for_untrusted_command;
 use crate::shell;
 use crate::turn_diff_tracker::TurnDiffTracker;
+use crate::unified_exec::UnifiedExecSessionManager;
 use crate::user_instructions::UserInstructions;
 use crate::user_notification::UserNotification;
 use crate::util::backoff;
@@ -273,6 +275,7 @@ pub(crate) struct Session {
     /// Manager for external MCP servers/tools.
     mcp_connection_manager: McpConnectionManager,
     session_manager: ExecSessionManager,
+    unified_exec_manager: UnifiedExecSessionManager,
 
     /// External notifier command (will be passed as args to exec()). When
     /// `None` this feature is disabled.
@@ -463,6 +466,7 @@ impl Session {
             tx_event: tx_event.clone(),
             mcp_connection_manager,
             session_manager: ExecSessionManager::default(),
+            unified_exec_manager: UnifiedExecSessionManager::default(),
             notify,
             state: Mutex::new(state),
             rollout: Mutex::new(Some(rollout_recorder)),
@@ -1984,6 +1988,27 @@ async fn handle_response_item(
                 .await,
             )
         }
+        ResponseItem::UnifiedExec {
+            id,
+            session_id,
+            arguments,
+            timeout_ms,
+        } => {
+            let call_id = id.clone().unwrap_or_else(|| match &session_id {
+                Some(session_id) => format!("unified_exec:{session_id}:{}", Uuid::new_v4()),
+                None => format!("unified_exec:{}", Uuid::new_v4()),
+            });
+            Some(
+                handle_unified_exec_tool_call(
+                    sess,
+                    call_id,
+                    session_id.clone(),
+                    arguments,
+                    timeout_ms,
+                )
+                .await,
+            )
+        }
         ResponseItem::CustomToolCall {
             id: _,
             call_id,
@@ -2028,6 +2053,72 @@ async fn handle_response_item(
     Ok(output)
 }
 
+async fn handle_unified_exec_tool_call(
+    sess: &Session,
+    call_id: String,
+    session_id: Option<String>,
+    arguments: Vec<String>,
+    timeout_ms: Option<u64>,
+) -> ResponseInputItem {
+    let parsed_session_id = if let Some(session_id) = session_id {
+        match session_id.parse::<i32>() {
+            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::unified_exec::UnifiedExecRequest {
+        session_id: parsed_session_id,
+        input_chunks: &arguments,
+        timeout_ms,
+    };
+
+    let result = sess.unified_exec_manager.handle_request(request).await;
+
+    let output_payload = match result {
+        Ok(value) => {
+            #[derive(Serialize)]
+            struct SerializedUnifiedExecResult<'a> {
+                session_id: Option<String>,
+                output: &'a str,
+            }
+
+            match serde_json::to_string(&SerializedUnifiedExecResult {
+                session_id: value.session_id.map(|id| id.to_string()),
+                output: &value.output,
+            }) {
+                Ok(serialized) => FunctionCallOutputPayload {
+                    content: serialized,
+                    success: Some(true),
+                },
+                Err(err) => FunctionCallOutputPayload {
+                    content: format!("failed to serialize unified exec output: {err}"),
+                    success: Some(false),
+                },
+            }
+        }
+        Err(err) => FunctionCallOutputPayload {
+            content: format!("unified exec failed: {err}"),
+            success: Some(false),
+        },
+    };
+
+    ResponseInputItem::FunctionCallOutput {
+        call_id,
+        output: output_payload,
+    }
+}
+
 async fn handle_function_call(
     sess: &Session,
     turn_context: &TurnContext,
@@ -2055,6 +2146,38 @@ async fn handle_function_call(
             )
             .await
         }
+        "unified_exec" => {
+            #[derive(Deserialize)]
+            struct UnifiedExecArgs {
+                input: Vec<String>,
+                #[serde(default)]
+                session_id: Option<String>,
+                #[serde(default)]
+                timeout_ms: Option<u64>,
+            }
+
+            let args = match serde_json::from_str::<UnifiedExecArgs>(&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_unified_exec_tool_call(
+                sess,
+                call_id,
+                args.session_id,
+                args.input,
+                args.timeout_ms,
+            )
+            .await
+        }
         "view_image" => {
             #[derive(serde::Deserialize)]
             struct SeeImageArgs {

codex-rs/core/src/conversation_history.rs

@@ -71,6 +71,7 @@ fn is_api_message(message: &ResponseItem) -> bool {
         | ResponseItem::CustomToolCall { .. }
         | ResponseItem::CustomToolCallOutput { .. }
         | ResponseItem::LocalShellCall { .. }
+        | ResponseItem::UnifiedExec { .. }
         | ResponseItem::Reasoning { .. } => true,
         ResponseItem::WebSearchCall { .. } | ResponseItem::Other => false,
     }

codex-rs/core/src/event_mapping.rs

@@ -91,6 +91,7 @@ pub(crate) fn map_response_item_to_event_messages(
         ResponseItem::FunctionCall { .. }
         | ResponseItem::FunctionCallOutput { .. }
         | ResponseItem::LocalShellCall { .. }
+        | ResponseItem::UnifiedExec { .. }
         | ResponseItem::CustomToolCall { .. }
         | ResponseItem::CustomToolCallOutput { .. }
         | ResponseItem::Other => Vec::new(),

codex-rs/core/src/exec_command/exec_command_session.rs

@@ -24,6 +24,9 @@ pub(crate) struct ExecCommandSession {
 
     /// JoinHandle for the child wait task.
     wait_handle: StdMutex<Option<JoinHandle<()>>>,
+
+    /// Tracks whether the underlying process has exited.
+    exit_status: std::sync::Arc<std::sync::atomic::AtomicBool>,
 }
 
 impl ExecCommandSession {
@@ -34,6 +37,7 @@ impl ExecCommandSession {
         reader_handle: JoinHandle<()>,
         writer_handle: JoinHandle<()>,
         wait_handle: JoinHandle<()>,
+        exit_status: std::sync::Arc<std::sync::atomic::AtomicBool>,
     ) -> Self {
         Self {
             writer_tx,
@@ -42,6 +46,7 @@ impl ExecCommandSession {
             reader_handle: StdMutex::new(Some(reader_handle)),
             writer_handle: StdMutex::new(Some(writer_handle)),
             wait_handle: StdMutex::new(Some(wait_handle)),
+            exit_status,
         }
     }
 
@@ -52,6 +57,10 @@ impl ExecCommandSession {
     pub(crate) fn output_receiver(&self) -> broadcast::Receiver<Vec<u8>> {
         self.output_tx.subscribe()
     }
+
+    pub(crate) fn has_exited(&self) -> bool {
+        self.exit_status.load(std::sync::atomic::Ordering::SeqCst)
+    }
 }
 
 impl Drop for ExecCommandSession {

codex-rs/core/src/exec_command/mod.rs

@@ -6,6 +6,7 @@ mod session_manager;
 
 pub use exec_command_params::ExecCommandParams;
 pub use exec_command_params::WriteStdinParams;
+pub(crate) use exec_command_session::ExecCommandSession;
 pub use responses_api::EXEC_COMMAND_TOOL_NAME;
 pub use responses_api::WRITE_STDIN_TOOL_NAME;
 pub use responses_api::create_exec_command_tool_for_responses_api;

codex-rs/core/src/exec_command/session_manager.rs

@@ -19,6 +19,7 @@ use crate::exec_command::exec_command_params::ExecCommandParams;
 use crate::exec_command::exec_command_params::WriteStdinParams;
 use crate::exec_command::exec_command_session::ExecCommandSession;
 use crate::exec_command::session_id::SessionId;
+use crate::truncate::truncate_middle;
 use codex_protocol::models::FunctionCallOutputPayload;
 
 #[derive(Debug, Default)]
@@ -327,11 +328,14 @@ async fn create_exec_command_session(
 
     // Keep the child alive until it exits, then signal exit code.
     let (exit_tx, exit_rx) = oneshot::channel::<i32>();
+    let exit_status = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false));
+    let wait_exit_status = std::sync::Arc::clone(&exit_status);
     let wait_handle = tokio::task::spawn_blocking(move || {
         let code = match child.wait() {
             Ok(status) => status.exit_code() as i32,
             Err(_) => -1,
         };
+        wait_exit_status.store(true, std::sync::atomic::Ordering::SeqCst);
         let _ = exit_tx.send(code);
     });
 
@@ -343,116 +347,11 @@ async fn create_exec_command_session(
         reader_handle,
         writer_handle,
         wait_handle,
+        exit_status,
     );
     Ok((session, exit_rx))
 }
 
-/// 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)
-/// if truncation occurred; otherwise returns the original string and `None`.
-fn truncate_middle(s: &str, max_bytes: usize) -> (String, Option<u64>) {
-    // No truncation needed
-    if s.len() <= max_bytes {
-        return (s.to_string(), None);
-    }
-    let est_tokens = (s.len() as u64).div_ceil(4);
-    if max_bytes == 0 {
-        // Cannot keep any content; still return a full marker (never truncated).
-        return (format!("…{est_tokens} tokens truncated…"), Some(est_tokens));
-    }
-
-    // Helper to truncate a string to a given byte length on a char boundary.
-    fn truncate_on_boundary(input: &str, max_len: usize) -> &str {
-        if input.len() <= max_len {
-            return input;
-        }
-        let mut end = max_len;
-        while end > 0 && !input.is_char_boundary(end) {
-            end -= 1;
-        }
-        &input[..end]
-    }
-
-    // Given a left/right budget, prefer newline boundaries; otherwise fall back
-    // to UTF-8 char boundaries.
-    fn pick_prefix_end(s: &str, left_budget: usize) -> usize {
-        if let Some(head) = s.get(..left_budget)
-            && let Some(i) = head.rfind('\n')
-        {
-            return i + 1; // keep the newline so suffix starts on a fresh line
-        }
-        truncate_on_boundary(s, left_budget).len()
-    }
-
-    fn pick_suffix_start(s: &str, right_budget: usize) -> usize {
-        let start_tail = s.len().saturating_sub(right_budget);
-        if let Some(tail) = s.get(start_tail..)
-            && let Some(i) = tail.find('\n')
-        {
-            return start_tail + i + 1; // start after newline
-        }
-        // Fall back to a char boundary at or after start_tail.
-        let mut idx = start_tail.min(s.len());
-        while idx < s.len() && !s.is_char_boundary(idx) {
-            idx += 1;
-        }
-        idx
-    }
-
-    // Refine marker length and budgets until stable. Marker is never truncated.
-    let mut guess_tokens = est_tokens; // worst-case: everything truncated
-    for _ in 0..4 {
-        let marker = format!("…{guess_tokens} tokens truncated…");
-        let marker_len = marker.len();
-        let keep_budget = max_bytes.saturating_sub(marker_len);
-        if keep_budget == 0 {
-            // No room for any content within the cap; return a full, untruncated marker
-            // that reflects the entire truncated content.
-            return (format!("…{est_tokens} tokens truncated…"), Some(est_tokens));
-        }
-
-        let left_budget = keep_budget / 2;
-        let right_budget = keep_budget - left_budget;
-        let prefix_end = pick_prefix_end(s, left_budget);
-        let mut suffix_start = pick_suffix_start(s, right_budget);
-        if suffix_start < prefix_end {
-            suffix_start = prefix_end;
-        }
-        let kept_content_bytes = prefix_end + (s.len() - suffix_start);
-        let truncated_content_bytes = s.len().saturating_sub(kept_content_bytes);
-        let new_tokens = (truncated_content_bytes as u64).div_ceil(4);
-        if new_tokens == guess_tokens {
-            let mut out = String::with_capacity(marker_len + kept_content_bytes + 1);
-            out.push_str(&s[..prefix_end]);
-            out.push_str(&marker);
-            // Place marker on its own line for symmetry when we keep line boundaries.
-            out.push('\n');
-            out.push_str(&s[suffix_start..]);
-            return (out, Some(est_tokens));
-        }
-        guess_tokens = new_tokens;
-    }
-
-    // Fallback: use last guess to build output.
-    let marker = format!("…{guess_tokens} tokens truncated…");
-    let marker_len = marker.len();
-    let keep_budget = max_bytes.saturating_sub(marker_len);
-    if keep_budget == 0 {
-        return (format!("…{est_tokens} tokens truncated…"), Some(est_tokens));
-    }
-    let left_budget = keep_budget / 2;
-    let right_budget = keep_budget - left_budget;
-    let prefix_end = pick_prefix_end(s, left_budget);
-    let suffix_start = pick_suffix_start(s, right_budget);
-    let mut out = String::with_capacity(marker_len + prefix_end + (s.len() - suffix_start) + 1);
-    out.push_str(&s[..prefix_end]);
-    out.push_str(&marker);
-    out.push('\n');
-    out.push_str(&s[suffix_start..]);
-    (out, Some(est_tokens))
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -616,50 +515,4 @@ Output:
 abc"#;
         assert_eq!(expected, text);
     }
-
-    #[test]
-    fn truncate_middle_no_newlines_fallback() {
-        // A long string with no newlines that exceeds the cap.
-        let s = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ";
-        let max_bytes = 16; // force truncation
-        let (out, original) = truncate_middle(s, max_bytes);
-        // For very small caps, we return the full, untruncated marker,
-        // even if it exceeds the cap.
-        assert_eq!(out, "…16 tokens truncated…");
-        // Original string length is 62 bytes => ceil(62/4) = 16 tokens.
-        assert_eq!(original, Some(16));
-    }
-
-    #[test]
-    fn truncate_middle_prefers_newline_boundaries() {
-        // Build a multi-line string of 20 numbered lines (each "NNN\n").
-        let mut s = String::new();
-        for i in 1..=20 {
-            s.push_str(&format!("{i:03}\n"));
-        }
-        // Total length: 20 lines * 4 bytes per line = 80 bytes.
-        assert_eq!(s.len(), 80);
-
-        // Choose a cap that forces truncation while leaving room for
-        // a few lines on each side after accounting for the marker.
-        let max_bytes = 64;
-        // Expect exact output: first 4 lines, marker, last 4 lines, and correct token estimate (80/4 = 20).
-        assert_eq!(
-            truncate_middle(&s, max_bytes),
-            (
-                r#"001
-002
-003
-004
-…12 tokens truncated…
-017
-018
-019
-020
-"#
-                .to_string(),
-                Some(20)
-            )
-        );
-    }
 }

codex-rs/core/src/lib.rs

@@ -34,6 +34,8 @@ mod mcp_tool_call;
 mod message_history;
 mod model_provider_info;
 pub mod parse_command;
+mod truncate;
+mod unified_exec;
 mod user_instructions;
 pub use model_provider_info::BUILT_IN_OSS_MODEL_PROVIDER_ID;
 pub use model_provider_info::ModelProviderInfo;

codex-rs/core/src/openai_tools.rs

@@ -200,6 +200,53 @@ fn create_shell_tool() -> OpenAiTool {
     })
 }
 
+fn create_unified_exec_tool() -> OpenAiTool {
+    let mut properties = BTreeMap::new();
+    properties.insert(
+        "input".to_string(),
+        JsonSchema::Array {
+            items: Box::new(JsonSchema::String { description: None }),
+            description: Some(
+                "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(),
+            ),
+        },
+    );
+    properties.insert(
+        "session_id".to_string(),
+        JsonSchema::String {
+            description: Some(
+                "Identifier for an existing interactive session. If omitted, a new command \
+                 is spawned."
+                    .to_string(),
+            ),
+        },
+    );
+    properties.insert(
+        "timeout_ms".to_string(),
+        JsonSchema::Number {
+            description: Some(
+                "Maximum time in milliseconds to wait for output after writing the input."
+                    .to_string(),
+            ),
+        },
+    );
+
+    OpenAiTool::Function(ResponsesApiTool {
+        name: "unified_exec".to_string(),
+        description:
+            "Runs a command in a PTY. Provide a session_id to reuse an existing interactive session.".to_string(),
+        strict: false,
+        parameters: JsonSchema::Object {
+            properties,
+            required: Some(vec!["input".to_string()]),
+            additional_properties: Some(false),
+        },
+    })
+}
+
 fn create_shell_tool_for_sandbox(sandbox_policy: &SandboxPolicy) -> OpenAiTool {
     let mut properties = BTreeMap::new();
     properties.insert(
@@ -534,25 +581,27 @@ pub(crate) fn get_openai_tools(
 ) -> Vec<OpenAiTool> {
     let mut tools: Vec<OpenAiTool> = Vec::new();
 
-    match &config.shell_type {
-        ConfigShellToolType::DefaultShell => {
-            tools.push(create_shell_tool());
-        }
-        ConfigShellToolType::ShellWithRequest { sandbox_policy } => {
-            tools.push(create_shell_tool_for_sandbox(sandbox_policy));
-        }
-        ConfigShellToolType::LocalShell => {
-            tools.push(OpenAiTool::LocalShell {});
-        }
-        ConfigShellToolType::StreamableShell => {
-            tools.push(OpenAiTool::Function(
-                crate::exec_command::create_exec_command_tool_for_responses_api(),
-            ));
-            tools.push(OpenAiTool::Function(
-                crate::exec_command::create_write_stdin_tool_for_responses_api(),
-            ));
-        }
-    }
+    // match &config.shell_type {
+    //     ConfigShellToolType::DefaultShell => {
+    //         tools.push(create_shell_tool());
+    //     }
+    //     ConfigShellToolType::ShellWithRequest { sandbox_policy } => {
+    //         tools.push(create_shell_tool_for_sandbox(sandbox_policy));
+    //     }
+    //     ConfigShellToolType::LocalShell => {
+    //         tools.push(OpenAiTool::LocalShell {});
+    //     }
+    //     ConfigShellToolType::StreamableShell => {
+    //         tools.push(OpenAiTool::Function(
+    //             crate::exec_command::create_exec_command_tool_for_responses_api(),
+    //         ));
+    //         tools.push(OpenAiTool::Function(
+    //             crate::exec_command::create_write_stdin_tool_for_responses_api(),
+    //         ));
+    //     }
+    // }
+
+    tools.push(create_unified_exec_tool());
 
     if config.plan_tool {
         tools.push(PLAN_TOOL.clone());
@@ -577,10 +626,7 @@ 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));
 
@@ -647,7 +693,13 @@ mod tests {
 
         assert_eq_tool_names(
             &tools,
-            &["local_shell", "update_plan", "web_search", "view_image"],
+            &[
+                "local_shell",
+                "unified_exec",
+                "update_plan",
+                "web_search",
+                "view_image",
+            ],
         );
     }
 
@@ -668,7 +720,13 @@ mod tests {
 
         assert_eq_tool_names(
             &tools,
-            &["shell", "update_plan", "web_search", "view_image"],
+            &[
+                "shell",
+                "unified_exec",
+                "update_plan",
+                "web_search",
+                "view_image",
+            ],
         );
     }
 
@@ -727,6 +785,7 @@ mod tests {
             &tools,
             &[
                 "shell",
+                "unified_exec",
                 "web_search",
                 "view_image",
                 "test_server/do_something_cool",
@@ -734,7 +793,7 @@ mod tests {
         );
 
         assert_eq!(
-            tools[3],
+            tools[4],
             OpenAiTool::Function(ResponsesApiTool {
                 name: "test_server/do_something_cool".to_string(),
                 parameters: JsonSchema::Object {
@@ -846,6 +905,7 @@ mod tests {
             &tools,
             &[
                 "shell",
+                "unified_exec",
                 "view_image",
                 "test_server/cool",
                 "test_server/do",
@@ -893,11 +953,17 @@ mod tests {
 
         assert_eq_tool_names(
             &tools,
-            &["shell", "web_search", "view_image", "dash/search"],
+            &[
+                "shell",
+                "unified_exec",
+                "web_search",
+                "view_image",
+                "dash/search",
+            ],
         );
 
         assert_eq!(
-            tools[3],
+            tools[4],
             OpenAiTool::Function(ResponsesApiTool {
                 name: "dash/search".to_string(),
                 parameters: JsonSchema::Object {
@@ -953,10 +1019,16 @@ mod tests {
 
         assert_eq_tool_names(
             &tools,
-            &["shell", "web_search", "view_image", "dash/paginate"],
+            &[
+                "shell",
+                "unified_exec",
+                "web_search",
+                "view_image",
+                "dash/paginate",
+            ],
         );
         assert_eq!(
-            tools[3],
+            tools[4],
             OpenAiTool::Function(ResponsesApiTool {
                 name: "dash/paginate".to_string(),
                 parameters: JsonSchema::Object {
@@ -1008,9 +1080,18 @@ mod tests {
             )])),
         );
 
-        assert_eq_tool_names(&tools, &["shell", "web_search", "view_image", "dash/tags"]);
+        assert_eq_tool_names(
+            &tools,
+            &[
+                "shell",
+                "unified_exec",
+                "web_search",
+                "view_image",
+                "dash/tags",
+            ],
+        );
         assert_eq!(
-            tools[3],
+            tools[4],
             OpenAiTool::Function(ResponsesApiTool {
                 name: "dash/tags".to_string(),
                 parameters: JsonSchema::Object {
@@ -1065,9 +1146,18 @@ mod tests {
             )])),
         );
 
-        assert_eq_tool_names(&tools, &["shell", "web_search", "view_image", "dash/value"]);
+        assert_eq_tool_names(
+            &tools,
+            &[
+                "shell",
+                "unified_exec",
+                "web_search",
+                "view_image",
+                "dash/value",
+            ],
+        );
         assert_eq!(
-            tools[3],
+            tools[4],
             OpenAiTool::Function(ResponsesApiTool {
                 name: "dash/value".to_string(),
                 parameters: JsonSchema::Object {

codex-rs/core/src/rollout/policy.rs

@@ -7,6 +7,7 @@ pub(crate) fn is_persisted_response_item(item: &ResponseItem) -> bool {
         ResponseItem::Message { .. }
         | ResponseItem::Reasoning { .. }
         | ResponseItem::LocalShellCall { .. }
+        | ResponseItem::UnifiedExec { .. }
         | ResponseItem::FunctionCall { .. }
         | ResponseItem::FunctionCallOutput { .. }
         | ResponseItem::CustomToolCall { .. }

codex-rs/core/src/truncate.rs

@@ -0,0 +1,180 @@
+//! Utilities for truncating large chunks of output while preserving a prefix
+//! and suffix on UTF-8 boundaries.
+
+/// 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)
+/// if truncation occurred; otherwise returns the original string and `None`.
+pub(crate) fn truncate_middle(s: &str, max_bytes: usize) -> (String, Option<u64>) {
+    if s.len() <= max_bytes {
+        return (s.to_string(), None);
+    }
+
+    let est_tokens = (s.len() as u64).div_ceil(4);
+    if max_bytes == 0 {
+        return (format!("…{est_tokens} tokens truncated…"), Some(est_tokens));
+    }
+
+    fn truncate_on_boundary(input: &str, max_len: usize) -> &str {
+        if input.len() <= max_len {
+            return input;
+        }
+        let mut end = max_len;
+        while end > 0 && !input.is_char_boundary(end) {
+            end -= 1;
+        }
+        &input[..end]
+    }
+
+    fn pick_prefix_end(s: &str, left_budget: usize) -> usize {
+        if let Some(head) = s.get(..left_budget)
+            && let Some(i) = head.rfind('\n')
+        {
+            return i + 1;
+        }
+        truncate_on_boundary(s, left_budget).len()
+    }
+
+    fn pick_suffix_start(s: &str, right_budget: usize) -> usize {
+        let start_tail = s.len().saturating_sub(right_budget);
+        if let Some(tail) = s.get(start_tail..)
+            && let Some(i) = tail.find('\n')
+        {
+            return start_tail + i + 1;
+        }
+
+        let mut idx = start_tail.min(s.len());
+        while idx < s.len() && !s.is_char_boundary(idx) {
+            idx += 1;
+        }
+        idx
+    }
+
+    let mut guess_tokens = est_tokens;
+    for _ in 0..4 {
+        let marker = format!("…{guess_tokens} tokens truncated…");
+        let marker_len = marker.len();
+        let keep_budget = max_bytes.saturating_sub(marker_len);
+        if keep_budget == 0 {
+            return (format!("…{est_tokens} tokens truncated…"), Some(est_tokens));
+        }
+
+        let left_budget = keep_budget / 2;
+        let right_budget = keep_budget - left_budget;
+        let prefix_end = pick_prefix_end(s, left_budget);
+        let mut suffix_start = pick_suffix_start(s, right_budget);
+        if suffix_start < prefix_end {
+            suffix_start = prefix_end;
+        }
+
+        let kept_content_bytes = prefix_end + (s.len() - suffix_start);
+        let truncated_content_bytes = s.len().saturating_sub(kept_content_bytes);
+        let new_tokens = (truncated_content_bytes as u64).div_ceil(4);
+
+        if new_tokens == guess_tokens {
+            let mut out = String::with_capacity(marker_len + kept_content_bytes + 1);
+            out.push_str(&s[..prefix_end]);
+            out.push_str(&marker);
+            out.push('\n');
+            out.push_str(&s[suffix_start..]);
+            return (out, Some(est_tokens));
+        }
+
+        guess_tokens = new_tokens;
+    }
+
+    let marker = format!("…{guess_tokens} tokens truncated…");
+    let marker_len = marker.len();
+    let keep_budget = max_bytes.saturating_sub(marker_len);
+    if keep_budget == 0 {
+        return (format!("…{est_tokens} tokens truncated…"), Some(est_tokens));
+    }
+
+    let left_budget = keep_budget / 2;
+    let right_budget = keep_budget - left_budget;
+    let prefix_end = pick_prefix_end(s, left_budget);
+    let suffix_start = pick_suffix_start(s, right_budget);
+
+    let mut out = String::with_capacity(marker_len + prefix_end + (s.len() - suffix_start) + 1);
+    out.push_str(&s[..prefix_end]);
+    out.push_str(&marker);
+    out.push('\n');
+    out.push_str(&s[suffix_start..]);
+    (out, Some(est_tokens))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::truncate_middle;
+
+    #[test]
+    fn truncate_middle_no_newlines_fallback() {
+        let s = "abcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ*";
+        let max_bytes = 32;
+        let (out, original) = truncate_middle(s, max_bytes);
+        assert!(out.starts_with("abc"));
+        assert!(out.contains("tokens truncated"));
+        assert!(out.ends_with("XYZ*"));
+        assert_eq!(original, Some((s.len() as u64).div_ceil(4)));
+    }
+
+    #[test]
+    fn truncate_middle_prefers_newline_boundaries() {
+        let mut s = String::new();
+        for i in 1..=20 {
+            s.push_str(&format!("{i:03}\n"));
+        }
+        assert_eq!(s.len(), 80);
+
+        let max_bytes = 64;
+        let (out, tokens) = truncate_middle(&s, max_bytes);
+        assert!(out.starts_with("001\n002\n003\n004\n"));
+        assert!(out.contains("tokens truncated"));
+        assert!(out.ends_with("017\n018\n019\n020\n"));
+        assert_eq!(tokens, Some(20));
+    }
+
+    #[test]
+    fn truncate_middle_handles_utf8_content() {
+        let s = "😀😀😀😀😀😀😀😀😀😀\nsecond line with ascii text\n";
+        let max_bytes = 32;
+        let (out, tokens) = truncate_middle(s, max_bytes);
+
+        assert!(out.contains("tokens truncated"));
+        assert!(!out.contains('\u{fffd}'));
+        assert_eq!(tokens, Some((s.len() as u64).div_ceil(4)));
+    }
+
+    #[test]
+    fn truncate_middle_prefers_newline_boundaries_2() {
+        // Build a multi-line string of 20 numbered lines (each "NNN\n").
+        let mut s = String::new();
+        for i in 1..=20 {
+            s.push_str(&format!("{i:03}\n"));
+        }
+        // Total length: 20 lines * 4 bytes per line = 80 bytes.
+        assert_eq!(s.len(), 80);
+
+        // Choose a cap that forces truncation while leaving room for
+        // a few lines on each side after accounting for the marker.
+        let max_bytes = 64;
+        // Expect exact output: first 4 lines, marker, last 4 lines, and correct token estimate (80/4 = 20).
+        assert_eq!(
+            truncate_middle(&s, max_bytes),
+            (
+                r#"001
+002
+003
+004
+…12 tokens truncated…
+017
+018
+019
+020
+"#
+                .to_string(),
+                Some(20)
+            )
+        );
+    }
+}

codex-rs/core/src/unified_exec/errors.rs

@@ -0,0 +1,26 @@
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub(crate) enum UnifiedExecError {
+    #[error("Failed to create unified exec session: {pty_error}")]
+    CreateSession {
+        #[source]
+        pty_error: anyhow::Error,
+    },
+    #[error("Unknown session id {session_id}")]
+    UnknownSessionId { session_id: i32 },
+    #[error("failed to write to stdin")]
+    WriteToStdin,
+    #[error("missing command line for unified exec request")]
+    MissingCommandLine,
+    #[error("invalid command line: {command_line}")]
+    InvalidCommandLine { command_line: String },
+    #[error("command not found: {command}")]
+    CommandNotFound { command: String },
+}
+
+impl UnifiedExecError {
+    pub(crate) fn create_session(error: anyhow::Error) -> Self {
+        Self::CreateSession { pty_error: error }
+    }
+}

codex-rs/core/src/unified_exec/mod.rs

@@ -0,0 +1,643 @@
+use portable_pty::CommandBuilder;
+use portable_pty::PtySize;
+use portable_pty::native_pty_system;
+use std::collections::HashMap;
+use std::collections::VecDeque;
+use std::io::ErrorKind;
+use std::io::Read;
+use std::sync::Arc;
+use std::sync::Mutex as StdMutex;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::AtomicI32;
+use std::sync::atomic::Ordering;
+use tokio::sync::Mutex;
+use tokio::sync::Notify;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+use tokio::time::Duration;
+use tokio::time::Instant;
+
+use crate::exec_command::ExecCommandSession;
+use crate::truncate::truncate_middle;
+
+mod errors;
+mod path;
+
+pub(crate) use errors::UnifiedExecError;
+
+use path::command_from_chunks;
+use path::join_input_chunks;
+use path::resolve_command_path;
+
+const DEFAULT_TIMEOUT_MS: u64 = 1_000;
+const MAX_TIMEOUT_MS: u64 = 60_000;
+const UNIFIED_EXEC_OUTPUT_MAX_BYTES: usize = 128 * 1024; // 128 KiB
+
+#[derive(Debug)]
+pub(crate) struct UnifiedExecRequest<'a> {
+    pub session_id: Option<i32>,
+    pub input_chunks: &'a [String],
+    pub timeout_ms: Option<u64>,
+}
+
+#[derive(Debug, Clone, PartialEq)]
+pub(crate) struct UnifiedExecResult {
+    pub session_id: Option<i32>,
+    pub output: String,
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct UnifiedExecSessionManager {
+    next_session_id: AtomicI32,
+    sessions: Mutex<HashMap<i32, ManagedUnifiedExecSession>>,
+}
+
+#[derive(Debug)]
+struct ManagedUnifiedExecSession {
+    session: ExecCommandSession,
+    output_buffer: OutputBuffer,
+    output_notify: Arc<Notify>,
+    output_task: JoinHandle<()>,
+}
+
+#[derive(Debug, Default)]
+struct OutputBufferState {
+    chunks: VecDeque<Vec<u8>>,
+    total_bytes: usize,
+}
+
+impl OutputBufferState {
+    fn push_chunk(&mut self, chunk: Vec<u8>) {
+        self.total_bytes = self.total_bytes.saturating_add(chunk.len());
+        self.chunks.push_back(chunk);
+
+        while self.total_bytes > UNIFIED_EXEC_OUTPUT_MAX_BYTES {
+            match self.chunks.pop_front() {
+                Some(removed) => {
+                    self.total_bytes = self.total_bytes.saturating_sub(removed.len());
+                }
+                None => break,
+            }
+        }
+    }
+
+    fn drain(&mut self) -> Vec<Vec<u8>> {
+        let drained: Vec<Vec<u8>> = self.chunks.drain(..).collect();
+        self.total_bytes = 0;
+        drained
+    }
+}
+
+type OutputBuffer = Arc<Mutex<OutputBufferState>>;
+type OutputHandles = (OutputBuffer, Arc<Notify>);
+
+impl ManagedUnifiedExecSession {
+    fn new(session: ExecCommandSession) -> Self {
+        let output_buffer = Arc::new(Mutex::new(OutputBufferState::default()));
+        let output_notify = Arc::new(Notify::new());
+        let mut receiver = session.output_receiver();
+        let buffer_clone = Arc::clone(&output_buffer);
+        let notify_clone = Arc::clone(&output_notify);
+        let output_task = tokio::spawn(async move {
+            while let Ok(chunk) = receiver.recv().await {
+                let mut guard = buffer_clone.lock().await;
+                guard.push_chunk(chunk);
+                drop(guard);
+                notify_clone.notify_waiters();
+            }
+        });
+
+        Self {
+            session,
+            output_buffer,
+            output_notify,
+            output_task,
+        }
+    }
+
+    fn writer_sender(&self) -> mpsc::Sender<Vec<u8>> {
+        self.session.writer_sender()
+    }
+
+    fn output_handles(&self) -> OutputHandles {
+        (
+            Arc::clone(&self.output_buffer),
+            Arc::clone(&self.output_notify),
+        )
+    }
+
+    fn has_exited(&self) -> bool {
+        self.session.has_exited()
+    }
+}
+
+impl Drop for ManagedUnifiedExecSession {
+    fn drop(&mut self) {
+        self.output_task.abort();
+    }
+}
+
+impl UnifiedExecSessionManager {
+    pub async fn handle_request(
+        &self,
+        request: UnifiedExecRequest<'_>,
+    ) -> Result<UnifiedExecResult, UnifiedExecError> {
+        let (timeout_ms, timeout_warning) = match request.timeout_ms {
+            Some(requested) if requested > MAX_TIMEOUT_MS => (
+                MAX_TIMEOUT_MS,
+                Some(format!(
+                    "Warning: requested timeout {requested}ms exceeds maximum of {MAX_TIMEOUT_MS}ms; clamping to {MAX_TIMEOUT_MS}ms.\n"
+                )),
+            ),
+            Some(requested) => (requested, None),
+            None => (DEFAULT_TIMEOUT_MS, None),
+        };
+
+        let mut new_session: Option<ManagedUnifiedExecSession> = None;
+        let session_id;
+        let writer_tx;
+        let output_buffer;
+        let output_notify;
+
+        if let Some(existing_id) = request.session_id {
+            let mut sessions = self.sessions.lock().await;
+            match sessions.get(&existing_id) {
+                Some(session) => {
+                    if session.has_exited() {
+                        sessions.remove(&existing_id);
+                        return Err(UnifiedExecError::UnknownSessionId {
+                            session_id: existing_id,
+                        });
+                    }
+                    let (buffer, notify) = session.output_handles();
+                    session_id = existing_id;
+                    writer_tx = session.writer_sender();
+                    output_buffer = buffer;
+                    output_notify = notify;
+                }
+                None => {
+                    return Err(UnifiedExecError::UnknownSessionId {
+                        session_id: existing_id,
+                    });
+                }
+            }
+            drop(sessions);
+        } else {
+            let command = command_from_chunks(request.input_chunks)?;
+            let new_id = self.next_session_id.fetch_add(1, Ordering::SeqCst);
+            let session = create_unified_exec_session(&command).await?;
+            let managed_session = ManagedUnifiedExecSession::new(session);
+            let (buffer, notify) = managed_session.output_handles();
+            writer_tx = managed_session.writer_sender();
+            output_buffer = buffer;
+            output_notify = notify;
+            session_id = new_id;
+            new_session = Some(managed_session);
+        };
+
+        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(UnifiedExecError::WriteToStdin);
+            }
+        }
+
+        let mut collected: Vec<u8> = Vec::with_capacity(4096);
+        let start = Instant::now();
+        let deadline = start + Duration::from_millis(timeout_ms);
+
+        loop {
+            let drained_chunks;
+            let mut wait_for_output = None;
+            {
+                let mut guard = output_buffer.lock().await;
+                drained_chunks = guard.drain();
+                if drained_chunks.is_empty() {
+                    wait_for_output = Some(output_notify.notified());
+                }
+            }
+
+            if drained_chunks.is_empty() {
+                if Instant::now() >= deadline {
+                    break;
+                }
+
+                let remaining = deadline.saturating_duration_since(Instant::now());
+                let notified = wait_for_output.unwrap_or_else(|| output_notify.notified());
+                tokio::pin!(notified);
+                tokio::select! {
+                    _ = &mut notified => {}
+                    _ = tokio::time::sleep(remaining) => break,
+                }
+                continue;
+            }
+
+            for chunk in drained_chunks {
+                collected.extend_from_slice(&chunk);
+            }
+
+            if Instant::now() >= deadline {
+                break;
+            }
+        }
+
+        let (output, _maybe_tokens) = truncate_middle(
+            &String::from_utf8_lossy(&collected),
+            UNIFIED_EXEC_OUTPUT_MAX_BYTES,
+        );
+        let output = if let Some(warning) = timeout_warning {
+            format!("{warning}{output}")
+        } else {
+            output
+        };
+
+        let should_store_session = if let Some(session) = new_session.as_ref() {
+            !session.has_exited()
+        } else if request.session_id.is_some() {
+            let mut sessions = self.sessions.lock().await;
+            if let Some(existing) = sessions.get(&session_id) {
+                if existing.has_exited() {
+                    sessions.remove(&session_id);
+                    false
+                } else {
+                    true
+                }
+            } else {
+                false
+            }
+        } else {
+            true
+        };
+
+        if should_store_session {
+            if let Some(session) = new_session {
+                self.sessions.lock().await.insert(session_id, session);
+            }
+            Ok(UnifiedExecResult {
+                session_id: Some(session_id),
+                output,
+            })
+        } else {
+            Ok(UnifiedExecResult {
+                session_id: None,
+                output,
+            })
+        }
+    }
+}
+
+async fn create_unified_exec_session(
+    command: &[String],
+) -> Result<ExecCommandSession, UnifiedExecError> {
+    if command.is_empty() {
+        return Err(UnifiedExecError::MissingCommandLine);
+    }
+
+    let pty_system = native_pty_system();
+
+    let pair = pty_system
+        .openpty(PtySize {
+            rows: 24,
+            cols: 80,
+            pixel_width: 0,
+            pixel_height: 0,
+        })
+        .map_err(UnifiedExecError::create_session)?;
+
+    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);
+    }
+
+    let mut child = pair
+        .slave
+        .spawn_command(command_builder)
+        .map_err(UnifiedExecError::create_session)?;
+    let killer = child.clone_killer();
+
+    let (writer_tx, mut writer_rx) = mpsc::channel::<Vec<u8>>(128);
+    let (output_tx, _) = tokio::sync::broadcast::channel::<Vec<u8>>(256);
+
+    let mut reader = pair
+        .master
+        .try_clone_reader()
+        .map_err(UnifiedExecError::create_session)?;
+    let output_tx_clone = output_tx.clone();
+    let reader_handle = tokio::task::spawn_blocking(move || {
+        let mut buf = [0u8; 8192];
+        loop {
+            match reader.read(&mut buf) {
+                Ok(0) => break,
+                Ok(n) => {
+                    let _ = output_tx_clone.send(buf[..n].to_vec());
+                }
+                Err(ref e) if e.kind() == ErrorKind::Interrupted => continue,
+                Err(ref e) if e.kind() == ErrorKind::WouldBlock => {
+                    std::thread::sleep(Duration::from_millis(5));
+                    continue;
+                }
+                Err(_) => break,
+            }
+        }
+    });
+
+    let writer = pair
+        .master
+        .take_writer()
+        .map_err(UnifiedExecError::create_session)?;
+    let writer = Arc::new(StdMutex::new(writer));
+    let writer_handle = tokio::spawn({
+        let writer = writer.clone();
+        async move {
+            while let Some(bytes) = writer_rx.recv().await {
+                let writer = writer.clone();
+                let _ = tokio::task::spawn_blocking(move || {
+                    if let Ok(mut guard) = writer.lock() {
+                        use std::io::Write;
+                        let _ = guard.write_all(&bytes);
+                        let _ = guard.flush();
+                    }
+                })
+                .await;
+            }
+        }
+    });
+
+    let exit_status = Arc::new(AtomicBool::new(false));
+    let wait_exit_status = Arc::clone(&exit_status);
+    let wait_handle = tokio::task::spawn_blocking(move || {
+        let _ = child.wait();
+        wait_exit_status.store(true, Ordering::SeqCst);
+    });
+
+    Ok(ExecCommandSession::new(
+        writer_tx,
+        output_tx,
+        killer,
+        reader_handle,
+        writer_handle,
+        wait_handle,
+        exit_status,
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::path::parse_command_line;
+    use super::*;
+
+    #[test]
+    fn parse_command_line_splits_words() {
+        assert_eq!(
+            parse_command_line("echo codex").unwrap(),
+            vec!["echo".to_string(), "codex".to_string()]
+        );
+    }
+
+    #[test]
+    fn parse_command_line_trims_whitespace() {
+        assert_eq!(
+            parse_command_line("  ls  -la  \n").unwrap(),
+            vec!["ls".to_string(), "-la".to_string()]
+        );
+    }
+
+    #[test]
+    fn parse_command_line_rejects_empty() {
+        let err = parse_command_line("   ").expect_err("expected error");
+        assert!(matches!(err, UnifiedExecError::MissingCommandLine));
+    }
+
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn unified_exec_persists_across_requests() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+
+        let open_shell = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["bash".to_string(), "-i".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+        let session_id = open_shell.session_id.expect("expected session_id");
+
+        manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &["export CODEX_INTERACTIVE_SHELL_VAR=codex\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        let out_2 = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &["echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        assert!(out_2.output.contains("codex"));
+
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn multi_unified_exec_sessions() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+
+        let shell_a = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["/bin/bash".to_string(), "-i".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+        let session_a = shell_a.session_id.expect("expected session id");
+
+        manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_a),
+                input_chunks: &["export CODEX_INTERACTIVE_SHELL_VAR=codex\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        let out_2 = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["/bin/echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+        assert!(!out_2.output.contains("codex"));
+
+        let out_3 = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_a),
+                input_chunks: &["echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+        assert!(out_3.output.contains("codex"));
+
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn unified_exec_timeouts() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+
+        let open_shell = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["/bin/bash".to_string(), "-i".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+        let session_id = open_shell.session_id.expect("expected session id");
+
+        manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &["export CODEX_INTERACTIVE_SHELL_VAR=codex\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        let out_2 = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &["sleep 5 && echo $CODEX_INTERACTIVE_SHELL_VAR\n".to_string()],
+                timeout_ms: Some(10),
+            })
+            .await?;
+        assert!(!out_2.output.contains("codex"));
+
+        tokio::time::sleep(Duration::from_secs(7)).await;
+
+        let empty = Vec::new();
+        let out_3 = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &empty,
+                timeout_ms: Some(100),
+            })
+            .await?;
+
+        assert!(out_3.output.contains("codex"));
+
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn requests_with_large_timeout_are_capped() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+
+        let result = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["/bin/echo".to_string(), "codex".to_string()],
+                timeout_ms: Some(120_000),
+            })
+            .await?;
+
+        assert!(result.output.starts_with(
+            "Warning: requested timeout 120000ms exceeds maximum of 60000ms; clamping to 60000ms.\n"
+        ));
+        assert!(result.output.contains("codex"));
+
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn completed_commands_do_not_persist_sessions() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+        let result = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["/bin/echo".to_string(), "codex".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        assert!(result.session_id.is_none());
+        assert!(result.output.contains("codex"));
+
+        assert!(manager.sessions.lock().await.is_empty());
+
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn correct_path_resolution() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+        let result = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["echo".to_string(), "codex".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        assert!(result.session_id.is_none());
+        assert!(result.output.contains("codex"));
+
+        assert!(manager.sessions.lock().await.is_empty());
+
+        Ok(())
+    }
+
+    #[cfg(unix)]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn reusing_completed_session_returns_unknown_session() -> Result<(), UnifiedExecError> {
+        let manager = UnifiedExecSessionManager::default();
+
+        let open_shell = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: None,
+                input_chunks: &["/bin/bash".to_string(), "-i".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+        let session_id = open_shell.session_id.expect("expected session id");
+
+        manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &["exit\n".to_string()],
+                timeout_ms: Some(1_500),
+            })
+            .await?;
+
+        tokio::time::sleep(Duration::from_millis(200)).await;
+
+        let err = manager
+            .handle_request(UnifiedExecRequest {
+                session_id: Some(session_id),
+                input_chunks: &[],
+                timeout_ms: Some(100),
+            })
+            .await
+            .expect_err("expected unknown session error");
+
+        match err {
+            UnifiedExecError::UnknownSessionId { session_id: err_id } => {
+                assert_eq!(err_id, session_id);
+            }
+            other => panic!("expected UnknownSessionId, got {other:?}"),
+        }
+
+        assert!(!manager.sessions.lock().await.contains_key(&session_id));
+
+        Ok(())
+    }
+}

codex-rs/core/src/unified_exec/path.rs

@@ -0,0 +1,194 @@
+use std::env;
+use std::path::Path;
+use std::path::PathBuf;
+
+use super::errors::UnifiedExecError;
+
+pub(crate) fn parse_command_line(line: &str) -> Result<Vec<String>, UnifiedExecError> {
+    let trimmed = line.trim();
+    if trimmed.is_empty() {
+        return Err(UnifiedExecError::MissingCommandLine);
+    }
+
+    match parse_command_line_impl(trimmed) {
+        Some(parts) if !parts.is_empty() => Ok(parts),
+        _ => Err(UnifiedExecError::InvalidCommandLine {
+            command_line: trimmed.to_string(),
+        }),
+    }
+}
+
+#[cfg(not(windows))]
+fn parse_command_line_impl(trimmed: &str) -> Option<Vec<String>> {
+    shlex::split(trimmed)
+}
+
+#[cfg(windows)]
+fn parse_command_line_impl(trimmed: &str) -> Option<Vec<String>> {
+    windows_split_command_line(trimmed)
+}
+
+#[cfg(windows)]
+fn windows_split_command_line(input: &str) -> Option<Vec<String>> {
+    let mut args = Vec::new();
+    let mut current = String::new();
+    let mut in_quotes = false;
+    let mut backslashes = 0usize;
+
+    for ch in input.chars() {
+        match ch {
+            '\\' => {
+                backslashes += 1;
+            }
+            '"' => {
+                current.extend(std::iter::repeat_n('\\', backslashes / 2));
+                if backslashes % 2 == 0 {
+                    in_quotes = !in_quotes;
+                } else {
+                    current.push('"');
+                }
+                backslashes = 0;
+            }
+            c if c.is_whitespace() && !in_quotes => {
+                current.extend(std::iter::repeat_n('\\', backslashes));
+                backslashes = 0;
+                if !current.is_empty() {
+                    args.push(std::mem::take(&mut current));
+                }
+            }
+            other => {
+                current.extend(std::iter::repeat_n('\\', backslashes));
+                backslashes = 0;
+                current.push(other);
+            }
+        }
+    }
+
+    if in_quotes {
+        return None;
+    }
+
+    current.extend(std::iter::repeat_n('\\', backslashes));
+    if !current.is_empty() {
+        args.push(current);
+    }
+
+    if args.is_empty() { None } else { Some(args) }
+}
+
+pub(crate) fn command_from_chunks(chunks: &[String]) -> Result<Vec<String>, UnifiedExecError> {
+    match chunks {
+        [] => Err(UnifiedExecError::MissingCommandLine),
+        [single] => parse_command_line(single),
+        _ => Ok(chunks.to_vec()),
+    }
+}
+
+pub(crate) fn join_input_chunks(chunks: &[String]) -> String {
+    match chunks {
+        [] => String::new(),
+        [single] => single.clone(),
+        _ => chunks.concat(),
+    }
+}
+
+pub(crate) fn resolve_command_path(command: &str) -> Result<String, UnifiedExecError> {
+    if command.is_empty() {
+        return Err(UnifiedExecError::MissingCommandLine);
+    }
+
+    if is_explicit_path(command) {
+        return ensure_executable(Path::new(command))
+            .then_some(command.to_string())
+            .ok_or_else(|| UnifiedExecError::CommandNotFound {
+                command: command.to_string(),
+            });
+    }
+
+    if let Some(resolved) = find_in_path(command) {
+        return Ok(resolved.to_string_lossy().to_string());
+    }
+
+    Err(UnifiedExecError::CommandNotFound {
+        command: command.to_string(),
+    })
+}
+
+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<PathBuf> {
+    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<PathBuf> {
+    build_platform_candidates(dir.join(command))
+}
+
+#[cfg(unix)]
+fn build_platform_candidates(candidate: PathBuf) -> Vec<PathBuf> {
+    vec![candidate]
+}
+
+#[cfg(windows)]
+fn build_platform_candidates(candidate: PathBuf) -> Vec<PathBuf> {
+    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()
+}
+
+#[cfg(all(test, windows))]
+mod tests {
+    use super::parse_command_line;
+
+    #[test]
+    fn parses_windows_command_with_spaces() {
+        let out = parse_command_line(r#""C:\Program Files\Git\bin\bash.exe" -i"#).unwrap();
+        assert_eq!(
+            out,
+            vec![
+                "C:\\Program Files\\Git\\bin\\bash.exe".to_string(),
+                "-i".to_string(),
+            ]
+        );
+    }
+}

codex-rs/core/tests/suite/prompt_caching.rs

@@ -191,7 +191,13 @@ async fn prompt_tools_are_consistent_across_requests() {
     let expected_instructions: &str = include_str!("../../prompt.md");
     // our internal implementation is responsible for keeping tools in sync
     // with the OpenAI schema, so we just verify the tool presence here
-    let expected_tools_names: &[&str] = &["shell", "update_plan", "apply_patch", "view_image"];
+    let expected_tools_names: &[&str] = &[
+        "shell",
+        "unified_exec",
+        "update_plan",
+        "apply_patch",
+        "view_image",
+    ];
     let body0 = requests[0].body_json::<serde_json::Value>().unwrap();
     assert_eq!(
         body0["instructions"],

codex-rs/protocol/src/models.rs

@@ -114,6 +114,19 @@ pub enum ResponseItem {
         status: Option<String>,
         action: WebSearchAction,
     },
+    UnifiedExec {
+        #[serde(skip_serializing)]
+        id: Option<String>,
+        /// Identifier for the interactive shell session. When absent, a new
+        /// session should be created.
+        session_id: Option<String>,
+        /// Characters that should be written to the interactive shell's
+        /// standard input.
+        arguments: Vec<String>,
+        /// Maximum amount of time to wait for additional output after writing
+        /// to stdin.
+        timeout_ms: Option<u64>,
+    },
 
     #[serde(other)]
     Other,