diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index e894a0844..4ba4b5ef9 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -20,6 +20,7 @@ use crate::parse_command::parse_command; use crate::protocol::EventMsg; use crate::protocol::ExecCommandBeginEvent; use crate::protocol::ExecCommandEndEvent; +use crate::protocol::ExecCommandSource; use crate::protocol::SandboxPolicy; use crate::protocol::TaskStartedEvent; use crate::sandboxing::ExecEnv; @@ -80,7 +81,8 @@ impl SessionTask for UserShellCommandTask { command: shell_invocation.clone(), cwd: turn_context.cwd.clone(), parsed_cmd, - is_user_shell_command: true, + source: ExecCommandSource::UserShell, + interaction_input: None, }), ) .await; diff --git a/codex-rs/core/src/tools/context.rs b/codex-rs/core/src/tools/context.rs index 029bacaa3..18b0d11ee 100644 --- a/codex-rs/core/src/tools/context.rs +++ b/codex-rs/core/src/tools/context.rs @@ -4,17 +4,13 @@ use crate::tools::TELEMETRY_PREVIEW_MAX_BYTES; use crate::tools::TELEMETRY_PREVIEW_MAX_LINES; use crate::tools::TELEMETRY_PREVIEW_TRUNCATION_NOTICE; use crate::turn_diff_tracker::TurnDiffTracker; -use codex_otel::otel_event_manager::OtelEventManager; use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::models::FunctionCallOutputPayload; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ShellToolCallParams; -use codex_protocol::protocol::FileChange; use codex_utils_string::take_bytes_at_char_boundary; use mcp_types::CallToolResult; use std::borrow::Cow; -use std::collections::HashMap; -use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Mutex; @@ -244,25 +240,3 @@ mod tests { assert_eq!(lines.last(), Some(&TELEMETRY_PREVIEW_TRUNCATION_NOTICE)); } } - -#[derive(Clone, Debug)] -#[allow(dead_code)] -pub(crate) struct ExecCommandContext { - pub(crate) turn: Arc, - pub(crate) call_id: String, - pub(crate) command_for_display: Vec, - pub(crate) cwd: PathBuf, - pub(crate) apply_patch: Option, - pub(crate) tool_name: String, - pub(crate) otel_event_manager: OtelEventManager, - // TODO(abhisek-oai): Find a better way to track this. - // https://github.com/openai/codex/pull/2471/files#r2470352242 - pub(crate) is_user_shell_command: bool, -} - -#[derive(Clone, Debug)] -#[allow(dead_code)] -pub(crate) struct ApplyPatchCommandContext { - pub(crate) user_explicitly_approved_this_action: bool, - pub(crate) changes: HashMap, -} diff --git a/codex-rs/core/src/tools/events.rs b/codex-rs/core/src/tools/events.rs index 5f93d51d8..3f8a9accf 100644 --- a/codex-rs/core/src/tools/events.rs +++ b/codex-rs/core/src/tools/events.rs @@ -8,6 +8,7 @@ use crate::parse_command::parse_command; use crate::protocol::EventMsg; use crate::protocol::ExecCommandBeginEvent; use crate::protocol::ExecCommandEndEvent; +use crate::protocol::ExecCommandSource; use crate::protocol::FileChange; use crate::protocol::PatchApplyBeginEvent; use crate::protocol::PatchApplyEndEvent; @@ -60,7 +61,8 @@ pub(crate) async fn emit_exec_command_begin( ctx: ToolEventCtx<'_>, command: &[String], cwd: &Path, - is_user_shell_command: bool, + source: ExecCommandSource, + interaction_input: Option, ) { ctx.session .send_event( @@ -70,7 +72,8 @@ pub(crate) async fn emit_exec_command_begin( command: command.to_vec(), cwd: cwd.to_path_buf(), parsed_cmd: parse_command(command), - is_user_shell_command, + source, + interaction_input, }), ) .await; @@ -80,7 +83,7 @@ pub(crate) enum ToolEmitter { Shell { command: Vec, cwd: PathBuf, - is_user_shell_command: bool, + source: ExecCommandSource, }, ApplyPatch { changes: HashMap, @@ -89,18 +92,17 @@ pub(crate) enum ToolEmitter { UnifiedExec { command: Vec, cwd: PathBuf, - // True for `exec_command` and false for `write_stdin`. - #[allow(dead_code)] - is_startup_command: bool, + source: ExecCommandSource, + interaction_input: Option, }, } impl ToolEmitter { - pub fn shell(command: Vec, cwd: PathBuf, is_user_shell_command: bool) -> Self { + pub fn shell(command: Vec, cwd: PathBuf, source: ExecCommandSource) -> Self { Self::Shell { command, cwd, - is_user_shell_command, + source, } } @@ -111,11 +113,17 @@ impl ToolEmitter { } } - pub fn unified_exec(command: &[String], cwd: PathBuf, is_startup_command: bool) -> Self { + pub fn unified_exec( + command: &[String], + cwd: PathBuf, + source: ExecCommandSource, + interaction_input: Option, + ) -> Self { Self::UnifiedExec { command: command.to_vec(), cwd, - is_startup_command, + source, + interaction_input, } } @@ -125,11 +133,11 @@ impl ToolEmitter { Self::Shell { command, cwd, - is_user_shell_command, + source, }, ToolEventStage::Begin, ) => { - emit_exec_command_begin(ctx, command, cwd.as_path(), *is_user_shell_command).await; + emit_exec_command_begin(ctx, command, cwd.as_path(), *source, None).await; } (Self::Shell { .. }, ToolEventStage::Success(output)) => { emit_exec_end( @@ -217,8 +225,23 @@ impl ToolEmitter { ) => { emit_patch_end(ctx, String::new(), (*message).to_string(), false).await; } - (Self::UnifiedExec { command, cwd, .. }, ToolEventStage::Begin) => { - emit_exec_command_begin(ctx, command, cwd.as_path(), false).await; + ( + Self::UnifiedExec { + command, + cwd, + source, + interaction_input, + }, + ToolEventStage::Begin, + ) => { + emit_exec_command_begin( + ctx, + command, + cwd.as_path(), + *source, + interaction_input.clone(), + ) + .await; } (Self::UnifiedExec { .. }, ToolEventStage::Success(output)) => { emit_exec_end( diff --git a/codex-rs/core/src/tools/handlers/shell.rs b/codex-rs/core/src/tools/handlers/shell.rs index 39f271b2d..173a3ccff 100644 --- a/codex-rs/core/src/tools/handlers/shell.rs +++ b/codex-rs/core/src/tools/handlers/shell.rs @@ -11,6 +11,7 @@ use crate::exec::ExecParams; use crate::exec_env::create_env; use crate::function_tool::FunctionCallError; use crate::is_safe_command::is_known_safe_command; +use crate::protocol::ExecCommandSource; use crate::tools::context::ToolInvocation; use crate::tools::context::ToolOutput; use crate::tools::context::ToolPayload; @@ -285,11 +286,13 @@ impl ShellHandler { } // Regular shell execution path. - let emitter = ToolEmitter::shell( - exec_params.command.clone(), - exec_params.cwd.clone(), - is_user_shell_command, - ); + let source = if is_user_shell_command { + ExecCommandSource::UserShell + } else { + ExecCommandSource::Agent + }; + let emitter = + ToolEmitter::shell(exec_params.command.clone(), exec_params.cwd.clone(), source); let event_ctx = ToolEventCtx::new(session.as_ref(), turn.as_ref(), &call_id, None); emitter.begin(event_ctx).await; diff --git a/codex-rs/core/src/tools/handlers/unified_exec.rs b/codex-rs/core/src/tools/handlers/unified_exec.rs index d5b4ecd71..6790b7359 100644 --- a/codex-rs/core/src/tools/handlers/unified_exec.rs +++ b/codex-rs/core/src/tools/handlers/unified_exec.rs @@ -4,6 +4,7 @@ use crate::function_tool::FunctionCallError; use crate::is_safe_command::is_known_safe_command; use crate::protocol::EventMsg; use crate::protocol::ExecCommandOutputDeltaEvent; +use crate::protocol::ExecCommandSource; use crate::protocol::ExecOutputStream; use crate::shell::get_shell_by_model_provided_path; use crate::tools::context::ToolInvocation; @@ -162,8 +163,12 @@ impl ToolHandler for UnifiedExecHandler { &context.call_id, None, ); - - let emitter = ToolEmitter::unified_exec(&command, cwd.clone(), true); + let emitter = ToolEmitter::unified_exec( + &command, + cwd.clone(), + ExecCommandSource::UnifiedExecStartup, + None, + ); emitter.emit(event_ctx, ToolEventStage::Begin).await; manager @@ -191,6 +196,7 @@ impl ToolHandler for UnifiedExecHandler { })?; manager .write_stdin(WriteStdinRequest { + call_id: &call_id, session_id: args.session_id, input: &args.chars, yield_time_ms: args.yield_time_ms, diff --git a/codex-rs/core/src/unified_exec/mod.rs b/codex-rs/core/src/unified_exec/mod.rs index 951032c07..8674564f3 100644 --- a/codex-rs/core/src/unified_exec/mod.rs +++ b/codex-rs/core/src/unified_exec/mod.rs @@ -74,6 +74,7 @@ pub(crate) struct ExecCommandRequest { #[derive(Debug)] pub(crate) struct WriteStdinRequest<'a> { + pub call_id: &'a str, pub session_id: i32, pub input: &'a str, pub yield_time_ms: u64, @@ -89,6 +90,7 @@ pub(crate) struct UnifiedExecResponse { pub session_id: Option, pub exit_code: Option, pub original_token_count: Option, + pub session_command: Option>, } #[derive(Default)] @@ -213,6 +215,7 @@ mod tests { .services .unified_exec_manager .write_stdin(WriteStdinRequest { + call_id: "write-stdin", session_id, input, yield_time_ms, diff --git a/codex-rs/core/src/unified_exec/session_manager.rs b/codex-rs/core/src/unified_exec/session_manager.rs index 55e9102b7..e7daee88e 100644 --- a/codex-rs/core/src/unified_exec/session_manager.rs +++ b/codex-rs/core/src/unified_exec/session_manager.rs @@ -6,12 +6,18 @@ use tokio::sync::mpsc; use tokio::time::Duration; use tokio::time::Instant; +use crate::codex::Session; +use crate::codex::TurnContext; use crate::exec::ExecToolCallOutput; use crate::exec::StreamOutput; use crate::exec_env::create_env; +use crate::protocol::BackgroundEventEvent; +use crate::protocol::EventMsg; +use crate::protocol::ExecCommandSource; use crate::sandboxing::ExecEnv; use crate::tools::events::ToolEmitter; use crate::tools::events::ToolEventCtx; +use crate::tools::events::ToolEventFailure; use crate::tools::events::ToolEventStage; use crate::tools::orchestrator::ToolOrchestrator; use crate::tools::runtimes::unified_exec::UnifiedExecRequest as UnifiedExecToolRequest; @@ -66,15 +72,18 @@ impl UnifiedExecSessionManager { let text = String::from_utf8_lossy(&collected).to_string(); let (output, original_token_count) = truncate_output_to_tokens(&text, max_tokens); let chunk_id = generate_chunk_id(); - let exit_code = session.exit_code(); - let session_id = if session.has_exited() { - None - } else { - Some( - self.store_session(session, context, &request.command, cwd.clone(), start) - .await, - ) - }; + let has_exited = session.has_exited(); + let stored_id = self + .store_session(session, context, &request.command, cwd.clone(), start) + .await; + let exit_code = self + .sessions + .lock() + .await + .get(&stored_id) + .map(|entry| entry.session.exit_code()); + // Only include a session_id in the response if the process is still alive. + let session_id = if has_exited { None } else { Some(stored_id) }; let response = UnifiedExecResponse { event_call_id: context.call_id.clone(), @@ -82,10 +91,15 @@ impl UnifiedExecSessionManager { wall_time, output, session_id, - exit_code, + exit_code: exit_code.flatten(), original_token_count, + session_command: Some(request.command.clone()), }; + if response.session_id.is_some() { + Self::emit_waiting_status(&context.session, &context.turn, &request.command).await; + } + // If the command completed during this call, emit an ExecCommandEnd via the emitter. if response.session_id.is_none() { let exit = response.exit_code.unwrap_or(-1); @@ -109,11 +123,46 @@ impl UnifiedExecSessionManager { ) -> Result { let session_id = request.session_id; - let (writer_tx, output_buffer, output_notify) = - self.prepare_session_handles(session_id).await?; + let ( + writer_tx, + output_buffer, + output_notify, + session_ref, + turn_ref, + session_command, + session_cwd, + ) = self.prepare_session_handles(session_id).await?; + + let interaction_emitter = ToolEmitter::unified_exec( + &session_command, + session_cwd.clone(), + ExecCommandSource::UnifiedExecInteraction, + (!request.input.is_empty()).then(|| request.input.to_string()), + ); + let make_event_ctx = || { + ToolEventCtx::new( + session_ref.as_ref(), + turn_ref.as_ref(), + request.call_id, + None, + ) + }; + interaction_emitter + .emit(make_event_ctx(), ToolEventStage::Begin) + .await; if !request.input.is_empty() { - Self::send_input(&writer_tx, request.input.as_bytes()).await?; + if let Err(err) = Self::send_input(&writer_tx, request.input.as_bytes()).await { + interaction_emitter + .emit( + make_event_ctx(), + ToolEventStage::Failure(ToolEventFailure::Message(format!( + "write_stdin failed: {err:?}" + ))), + ) + .await; + return Err(err); + } tokio::time::sleep(Duration::from_millis(100)).await; } @@ -151,8 +200,28 @@ impl UnifiedExecSessionManager { session_id, exit_code, original_token_count, + session_command: Some(session_command.clone()), }; + let interaction_output = ExecToolCallOutput { + exit_code: response.exit_code.unwrap_or(0), + stdout: StreamOutput::new(response.output.clone()), + stderr: StreamOutput::new(String::new()), + aggregated_output: StreamOutput::new(response.output.clone()), + duration: response.wall_time, + timed_out: false, + }; + interaction_emitter + .emit( + make_event_ctx(), + ToolEventStage::Success(interaction_output), + ) + .await; + + if response.session_id.is_some() { + Self::emit_waiting_status(&session_ref, &turn_ref, &session_command).await; + } + if let (Some(exit), Some(entry)) = (response.exit_code, completion_entry) { let total_duration = Instant::now().saturating_duration_since(entry.started_at); Self::emit_exec_end_from_entry(entry, response.output.clone(), exit, total_duration) @@ -189,17 +258,44 @@ impl UnifiedExecSessionManager { async fn prepare_session_handles( &self, session_id: i32, - ) -> Result<(mpsc::Sender>, OutputBuffer, Arc), UnifiedExecError> { + ) -> Result< + ( + mpsc::Sender>, + OutputBuffer, + Arc, + Arc, + Arc, + Vec, + PathBuf, + ), + UnifiedExecError, + > { let sessions = self.sessions.lock().await; - let (output_buffer, output_notify, writer_tx) = + let (output_buffer, output_notify, writer_tx, session, turn, command, cwd) = if let Some(entry) = sessions.get(&session_id) { let (buffer, notify) = entry.session.output_handles(); - (buffer, notify, entry.session.writer_sender()) + ( + buffer, + notify, + entry.session.writer_sender(), + Arc::clone(&entry.session_ref), + Arc::clone(&entry.turn_ref), + entry.command.clone(), + entry.cwd.clone(), + ) } else { return Err(UnifiedExecError::UnknownSessionId { session_id }); }; - Ok((writer_tx, output_buffer, output_notify)) + Ok(( + writer_tx, + output_buffer, + output_notify, + session, + turn, + command, + cwd, + )) } async fn send_input( @@ -256,7 +352,12 @@ impl UnifiedExecSessionManager { &entry.call_id, None, ); - let emitter = ToolEmitter::unified_exec(&entry.command, entry.cwd, true); + let emitter = ToolEmitter::unified_exec( + &entry.command, + entry.cwd, + ExecCommandSource::UnifiedExecStartup, + None, + ); emitter .emit(event_ctx, ToolEventStage::Success(output)) .await; @@ -284,12 +385,28 @@ impl UnifiedExecSessionManager { &context.call_id, None, ); - let emitter = ToolEmitter::unified_exec(command, cwd, true); + let emitter = + ToolEmitter::unified_exec(command, cwd, ExecCommandSource::UnifiedExecStartup, None); emitter .emit(event_ctx, ToolEventStage::Success(output)) .await; } + async fn emit_waiting_status( + session: &Arc, + turn: &Arc, + command: &[String], + ) { + let command_display = command.join(" "); + let message = format!("Waiting for `{command_display}`"); + session + .send_event( + turn.as_ref(), + EventMsg::BackgroundEvent(BackgroundEventEvent { message }), + ) + .await; + } + pub(crate) async fn open_session_with_exec_env( &self, env: &ExecEnv, diff --git a/codex-rs/core/tests/suite/unified_exec.rs b/codex-rs/core/tests/suite/unified_exec.rs index 07b17f78a..4dda8f1db 100644 --- a/codex-rs/core/tests/suite/unified_exec.rs +++ b/codex-rs/core/tests/suite/unified_exec.rs @@ -7,6 +7,7 @@ use anyhow::Result; use codex_core::features::Feature; use codex_core::protocol::AskForApproval; use codex_core::protocol::EventMsg; +use codex_core::protocol::ExecCommandSource; use codex_core::protocol::Op; use codex_core::protocol::SandboxPolicy; use codex_protocol::config_types::ReasoningSummary; @@ -568,7 +569,109 @@ async fn unified_exec_emits_output_delta_for_write_stdin() -> Result<()> { } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn unified_exec_skips_begin_event_for_empty_input() -> Result<()> { +async fn unified_exec_emits_begin_for_write_stdin() -> Result<()> { + skip_if_no_network!(Ok(())); + skip_if_sandbox!(Ok(())); + + let server = start_mock_server().await; + + let mut builder = test_codex().with_config(|config| { + config.use_experimental_unified_exec_tool = true; + config.features.enable(Feature::UnifiedExec); + }); + let TestCodex { + codex, + cwd, + session_configured, + .. + } = builder.build(&server).await?; + + let open_call_id = "uexec-open-for-begin"; + let open_args = json!({ + "cmd": "/bin/sh -c echo ready".to_string(), + "yield_time_ms": 200, + }); + + let stdin_call_id = "uexec-stdin-begin"; + let stdin_args = json!({ + "chars": "echo hello", + "session_id": 0, + "yield_time_ms": 400, + }); + + let responses = vec![ + sse(vec![ + ev_response_created("resp-1"), + ev_function_call( + open_call_id, + "exec_command", + &serde_json::to_string(&open_args)?, + ), + ev_completed("resp-1"), + ]), + sse(vec![ + ev_response_created("resp-2"), + ev_function_call( + stdin_call_id, + "write_stdin", + &serde_json::to_string(&stdin_args)?, + ), + ev_completed("resp-2"), + ]), + sse(vec![ + ev_response_created("resp-3"), + ev_assistant_message("msg-1", "done"), + ev_completed("resp-3"), + ]), + ]; + mount_sse_sequence(&server, responses).await; + + let session_model = session_configured.model.clone(); + + codex + .submit(Op::UserTurn { + items: vec![UserInput::Text { + text: "begin events for stdin".into(), + }], + final_output_json_schema: None, + cwd: cwd.path().to_path_buf(), + approval_policy: AskForApproval::Never, + sandbox_policy: SandboxPolicy::DangerFullAccess, + model: session_model, + effort: None, + summary: ReasoningSummary::Auto, + }) + .await?; + + let begin_event = wait_for_event_match(&codex, |msg| match msg { + EventMsg::ExecCommandBegin(ev) if ev.call_id == stdin_call_id => Some(ev.clone()), + _ => None, + }) + .await; + + assert_eq!( + begin_event.command, + vec![ + "/bin/bash".to_string(), + "-lc".to_string(), + "/bin/sh -c echo ready".to_string() + ] + ); + assert_eq!( + begin_event.interaction_input, + Some("echo hello".to_string()) + ); + assert_eq!( + begin_event.source, + ExecCommandSource::UnifiedExecInteraction + ); + + wait_for_event(&codex, |event| matches!(event, EventMsg::TaskComplete(_))).await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn unified_exec_emits_begin_event_for_write_stdin_requests() -> Result<()> { skip_if_no_network!(Ok(())); skip_if_sandbox!(Ok(())); @@ -593,9 +696,9 @@ async fn unified_exec_skips_begin_event_for_empty_input() -> Result<()> { let poll_call_id = "uexec-poll-empty"; let poll_args = json!({ - "input": Vec::::new(), - "session_id": "0", - "timeout_ms": 150, + "chars": "", + "session_id": 0, + "yield_time_ms": 150, }); let responses = vec![ @@ -654,18 +757,45 @@ async fn unified_exec_skips_begin_event_for_empty_input() -> Result<()> { assert_eq!( begin_events.len(), - 1, - "expected only the initial command to emit begin event" + 2, + "expected begin events for the startup command and the write_stdin call" ); - assert_eq!(begin_events[0].call_id, open_call_id); + + let open_event = begin_events + .iter() + .find(|ev| ev.call_id == open_call_id) + .expect("missing exec_command begin"); assert_eq!( - begin_events[0].command, + open_event.command, vec![ "/bin/bash".to_string(), "-lc".to_string(), "/bin/sh -c echo ready".to_string() ] ); + assert!( + open_event.interaction_input.is_none(), + "startup begin events should not include interaction input" + ); + assert_eq!(open_event.source, ExecCommandSource::UnifiedExecStartup); + + let poll_event = begin_events + .iter() + .find(|ev| ev.call_id == poll_call_id) + .expect("missing write_stdin begin"); + assert_eq!( + poll_event.command, + vec![ + "/bin/bash".to_string(), + "-lc".to_string(), + "/bin/sh -c echo ready".to_string() + ] + ); + assert!( + poll_event.interaction_input.is_none(), + "poll begin events should omit interaction input" + ); + assert_eq!(poll_event.source, ExecCommandSource::UnifiedExecInteraction); Ok(()) } diff --git a/codex-rs/core/tests/suite/user_shell_cmd.rs b/codex-rs/core/tests/suite/user_shell_cmd.rs index 396b6c7d9..95c626984 100644 --- a/codex-rs/core/tests/suite/user_shell_cmd.rs +++ b/codex-rs/core/tests/suite/user_shell_cmd.rs @@ -4,6 +4,7 @@ use codex_core::NewConversation; use codex_core::model_family::find_family_for_model; use codex_core::protocol::EventMsg; use codex_core::protocol::ExecCommandEndEvent; +use codex_core::protocol::ExecCommandSource; use codex_core::protocol::ExecOutputStream; use codex_core::protocol::Op; use codex_core::protocol::SandboxPolicy; @@ -150,7 +151,7 @@ async fn user_shell_command_history_is_persisted_and_shared_with_model() -> anyh _ => None, }) .await; - assert!(begin_event.is_user_shell_command); + assert_eq!(begin_event.source, ExecCommandSource::UserShell); let matches_last_arg = begin_event.command.last() == Some(&command); let matches_split = shlex::split(&command).is_some_and(|split| split == begin_event.command); assert!( diff --git a/codex-rs/exec/tests/event_processor_with_json_output.rs b/codex-rs/exec/tests/event_processor_with_json_output.rs index 5e2ef17dc..f9258a2b7 100644 --- a/codex-rs/exec/tests/event_processor_with_json_output.rs +++ b/codex-rs/exec/tests/event_processor_with_json_output.rs @@ -5,6 +5,7 @@ use codex_core::protocol::Event; use codex_core::protocol::EventMsg; use codex_core::protocol::ExecCommandBeginEvent; use codex_core::protocol::ExecCommandEndEvent; +use codex_core::protocol::ExecCommandSource; use codex_core::protocol::FileChange; use codex_core::protocol::McpInvocation; use codex_core::protocol::McpToolCallBeginEvent; @@ -626,7 +627,8 @@ fn exec_command_end_success_produces_completed_command_item() { command: vec!["bash".to_string(), "-lc".to_string(), "echo hi".to_string()], cwd: std::env::current_dir().unwrap(), parsed_cmd: Vec::new(), - is_user_shell_command: false, + source: ExecCommandSource::Agent, + interaction_input: None, }), ); let out_begin = ep.collect_thread_events(&begin); @@ -687,7 +689,8 @@ fn exec_command_end_failure_produces_failed_command_item() { command: vec!["sh".to_string(), "-c".to_string(), "exit 1".to_string()], cwd: std::env::current_dir().unwrap(), parsed_cmd: Vec::new(), - is_user_shell_command: false, + source: ExecCommandSource::Agent, + interaction_input: None, }), ); assert_eq!( diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 95e64089c..defdd9385 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -1216,6 +1216,21 @@ pub struct ReviewLineRange { pub end: u32, } +#[derive(Debug, Clone, Copy, Deserialize, Serialize, PartialEq, Eq, JsonSchema, TS)] +#[serde(rename_all = "snake_case")] +pub enum ExecCommandSource { + Agent, + UserShell, + UnifiedExecStartup, + UnifiedExecInteraction, +} + +impl Default for ExecCommandSource { + fn default() -> Self { + Self::Agent + } +} + #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct ExecCommandBeginEvent { /// Identifier so this can be paired with the ExecCommandEnd event. @@ -1225,10 +1240,13 @@ pub struct ExecCommandBeginEvent { /// The command's working directory if not the default cwd for the agent. pub cwd: PathBuf, pub parsed_cmd: Vec, - /// True when this exec was initiated directly by the user (e.g. bang command), - /// not by the agent/model. Defaults to false for backwards compatibility. + /// Where the command originated. Defaults to Agent for backward compatibility. #[serde(default)] - pub is_user_shell_command: bool, + pub source: ExecCommandSource, + /// Raw input sent to a unified exec session (if this is an interaction event). + #[serde(default, skip_serializing_if = "Option::is_none")] + #[ts(optional)] + pub interaction_input: Option, } #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index aed73057c..79781f335 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -23,6 +23,7 @@ use codex_core::protocol::EventMsg; use codex_core::protocol::ExecApprovalRequestEvent; use codex_core::protocol::ExecCommandBeginEvent; use codex_core::protocol::ExecCommandEndEvent; +use codex_core::protocol::ExecCommandSource; use codex_core::protocol::ExitedReviewModeEvent; use codex_core::protocol::ListCustomPromptsResponseEvent; use codex_core::protocol::McpListToolsResponseEvent; @@ -129,7 +130,7 @@ const USER_SHELL_COMMAND_HELP_HINT: &str = "Example: !ls"; struct RunningCommand { command: Vec, parsed_cmd: Vec, - is_user_shell_command: bool, + source: ExecCommandSource, } const RATE_LIMIT_WARNING_THRESHOLDS: [f64; 3] = [75.0, 90.0, 95.0]; @@ -724,6 +725,9 @@ impl ChatWidget { fn on_background_event(&mut self, message: String) { debug!("BackgroundEvent: {message}"); + self.bottom_pane.ensure_status_indicator(); + self.bottom_pane.set_interrupt_hint_visible(true); + self.set_status_header(message); } fn on_undo_started(&mut self, event: UndoStartedEvent) { @@ -833,10 +837,16 @@ impl ChatWidget { pub(crate) fn handle_exec_end_now(&mut self, ev: ExecCommandEndEvent) { let running = self.running_commands.remove(&ev.call_id); - let (command, parsed, is_user_shell_command) = match running { - Some(rc) => (rc.command, rc.parsed_cmd, rc.is_user_shell_command), - None => (vec![ev.call_id.clone()], Vec::new(), false), + let (command, parsed, source) = match running { + Some(rc) => (rc.command, rc.parsed_cmd, rc.source), + None => ( + vec![ev.call_id.clone()], + Vec::new(), + ExecCommandSource::Agent, + ), }; + let is_unified_exec_interaction = + matches!(source, ExecCommandSource::UnifiedExecInteraction); let needs_new = self .active_cell @@ -849,7 +859,8 @@ impl ChatWidget { ev.call_id.clone(), command, parsed, - is_user_shell_command, + source, + None, ))); } @@ -858,15 +869,20 @@ impl ChatWidget { .as_mut() .and_then(|c| c.as_any_mut().downcast_mut::()) { - cell.complete_call( - &ev.call_id, + let output = if is_unified_exec_interaction { + CommandOutput { + exit_code: ev.exit_code, + formatted_output: String::new(), + aggregated_output: String::new(), + } + } else { CommandOutput { exit_code: ev.exit_code, formatted_output: ev.formatted_output.clone(), aggregated_output: ev.aggregated_output.clone(), - }, - ev.duration, - ); + } + }; + cell.complete_call(&ev.call_id, output, ev.duration); if cell.should_flush() { self.flush_active_cell(); } @@ -928,9 +944,10 @@ impl ChatWidget { RunningCommand { command: ev.command.clone(), parsed_cmd: ev.parsed_cmd.clone(), - is_user_shell_command: ev.is_user_shell_command, + source: ev.source, }, ); + let interaction_input = ev.interaction_input.clone(); if let Some(cell) = self .active_cell .as_mut() @@ -939,7 +956,8 @@ impl ChatWidget { ev.call_id.clone(), ev.command.clone(), ev.parsed_cmd.clone(), - ev.is_user_shell_command, + ev.source, + interaction_input.clone(), ) { *cell = new_exec; @@ -950,7 +968,8 @@ impl ChatWidget { ev.call_id.clone(), ev.command.clone(), ev.parsed_cmd, - ev.is_user_shell_command, + ev.source, + interaction_input, ))); } diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index f50ba9e9a..bdaae9335 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -18,11 +18,13 @@ use codex_core::protocol::AgentMessageEvent; use codex_core::protocol::AgentReasoningDeltaEvent; use codex_core::protocol::AgentReasoningEvent; use codex_core::protocol::ApplyPatchApprovalRequestEvent; +use codex_core::protocol::BackgroundEventEvent; use codex_core::protocol::Event; use codex_core::protocol::EventMsg; use codex_core::protocol::ExecApprovalRequestEvent; use codex_core::protocol::ExecCommandBeginEvent; use codex_core::protocol::ExecCommandEndEvent; +use codex_core::protocol::ExecCommandSource; use codex_core::protocol::ExitedReviewModeEvent; use codex_core::protocol::FileChange; use codex_core::protocol::Op; @@ -660,7 +662,12 @@ fn exec_approval_decision_truncates_multiline_and_long_commands() { } // --- Small helpers to tersely drive exec begin/end and snapshot active cell --- -fn begin_exec(chat: &mut ChatWidget, call_id: &str, raw_cmd: &str) { +fn begin_exec_with_source( + chat: &mut ChatWidget, + call_id: &str, + raw_cmd: &str, + source: ExecCommandSource, +) { // Build the full command vec and parse it using core's parser, // then convert to protocol variants for the event payload. let command = vec!["bash".to_string(), "-lc".to_string(), raw_cmd.to_string()]; @@ -672,11 +679,16 @@ fn begin_exec(chat: &mut ChatWidget, call_id: &str, raw_cmd: &str) { command, cwd: std::env::current_dir().unwrap_or_else(|_| PathBuf::from(".")), parsed_cmd, - is_user_shell_command: false, + source, + interaction_input: None, }), }); } +fn begin_exec(chat: &mut ChatWidget, call_id: &str, raw_cmd: &str) { + begin_exec_with_source(chat, call_id, raw_cmd, ExecCommandSource::Agent); +} + fn end_exec(chat: &mut ChatWidget, call_id: &str, stdout: &str, stderr: &str, exit_code: i32) { let aggregated = if stderr.is_empty() { stdout.to_string() @@ -933,6 +945,38 @@ fn exec_history_cell_shows_working_then_failed() { assert!(blob.to_lowercase().contains("bloop"), "expected error text"); } +#[test] +fn exec_history_shows_unified_exec_startup_commands() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); + + begin_exec_with_source( + &mut chat, + "call-startup", + "echo unified exec startup", + ExecCommandSource::UnifiedExecStartup, + ); + assert!( + drain_insert_history(&mut rx).is_empty(), + "exec begin should not flush until completion" + ); + + end_exec( + &mut chat, + "call-startup", + "echo unified exec startup\n", + "", + 0, + ); + + let cells = drain_insert_history(&mut rx); + assert_eq!(cells.len(), 1, "expected finalized exec cell to flush"); + let blob = lines_to_single_string(&cells[0]); + assert!( + blob.contains("• Ran echo unified exec startup"), + "expected startup command to render: {blob:?}" + ); +} + /// Selecting the custom prompt option from the review popup sends /// OpenReviewCustomPrompt to the app event channel. #[test] @@ -1744,7 +1788,8 @@ async fn binary_size_transcript_snapshot() { command: e.command, cwd: e.cwd, parsed_cmd, - is_user_shell_command: false, + source: ExecCommandSource::Agent, + interaction_input: e.interaction_input.clone(), }), } } @@ -2164,6 +2209,22 @@ fn status_widget_active_snapshot() { assert_snapshot!("status_widget_active", terminal.backend()); } +#[test] +fn background_event_updates_status_header() { + let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); + + chat.handle_codex_event(Event { + id: "bg-1".into(), + msg: EventMsg::BackgroundEvent(BackgroundEventEvent { + message: "Waiting for `vim`".to_string(), + }), + }); + + assert!(chat.bottom_pane.status_indicator_visible()); + assert_eq!(chat.current_status_header, "Waiting for `vim`"); + assert!(drain_insert_history(&mut rx).is_empty()); +} + #[test] fn apply_patch_events_emit_history_cells() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(); @@ -2815,7 +2876,8 @@ fn chatwidget_exec_and_status_layout_vt100_snapshot() { path: "diff_render.rs".into(), }, ], - is_user_shell_command: false, + source: ExecCommandSource::Agent, + interaction_input: None, }), }); chat.handle_codex_event(Event { diff --git a/codex-rs/tui/src/exec_cell/model.rs b/codex-rs/tui/src/exec_cell/model.rs index 1d6b544eb..943fb8365 100644 --- a/codex-rs/tui/src/exec_cell/model.rs +++ b/codex-rs/tui/src/exec_cell/model.rs @@ -1,6 +1,7 @@ use std::time::Duration; use std::time::Instant; +use codex_core::protocol::ExecCommandSource; use codex_protocol::parse_command::ParsedCommand; #[derive(Clone, Debug, Default)] @@ -18,9 +19,10 @@ pub(crate) struct ExecCall { pub(crate) command: Vec, pub(crate) parsed: Vec, pub(crate) output: Option, - pub(crate) is_user_shell_command: bool, + pub(crate) source: ExecCommandSource, pub(crate) start_time: Option, pub(crate) duration: Option, + pub(crate) interaction_input: Option, } #[derive(Debug)] @@ -38,16 +40,18 @@ impl ExecCell { call_id: String, command: Vec, parsed: Vec, - is_user_shell_command: bool, + source: ExecCommandSource, + interaction_input: Option, ) -> Option { let call = ExecCall { call_id, command, parsed, output: None, - is_user_shell_command, + source, start_time: Some(Instant::now()), duration: None, + interaction_input, }; if self.is_exploring_cell() && Self::is_exploring_call(&call) { Some(Self { @@ -124,3 +128,13 @@ impl ExecCell { }) } } + +impl ExecCall { + pub(crate) fn is_user_shell_command(&self) -> bool { + matches!(self.source, ExecCommandSource::UserShell) + } + + pub(crate) fn is_unified_exec_interaction(&self) -> bool { + matches!(self.source, ExecCommandSource::UnifiedExecInteraction) + } +} diff --git a/codex-rs/tui/src/exec_cell/render.rs b/codex-rs/tui/src/exec_cell/render.rs index 8ebc12512..352a61476 100644 --- a/codex-rs/tui/src/exec_cell/render.rs +++ b/codex-rs/tui/src/exec_cell/render.rs @@ -14,6 +14,7 @@ use crate::wrapping::word_wrap_line; use crate::wrapping::word_wrap_lines; use codex_ansi_escape::ansi_escape_line; use codex_common::elapsed::format_duration; +use codex_core::protocol::ExecCommandSource; use codex_protocol::parse_command::ParsedCommand; use itertools::Itertools; use ratatui::prelude::*; @@ -24,6 +25,7 @@ use unicode_width::UnicodeWidthStr; pub(crate) const TOOL_CALL_MAX_LINES: usize = 5; const USER_SHELL_TOOL_CALL_MAX_LINES: usize = 50; +const MAX_INTERACTION_PREVIEW_CHARS: usize = 80; pub(crate) struct OutputLinesParams { pub(crate) line_limit: usize, @@ -36,19 +38,47 @@ pub(crate) fn new_active_exec_command( call_id: String, command: Vec, parsed: Vec, - is_user_shell_command: bool, + source: ExecCommandSource, + interaction_input: Option, ) -> ExecCell { ExecCell::new(ExecCall { call_id, command, parsed, output: None, - is_user_shell_command, + source, start_time: Some(Instant::now()), duration: None, + interaction_input, }) } +fn format_unified_exec_interaction(command: &[String], input: Option<&str>) -> String { + let command_display = command.join(" "); + match input { + Some(data) if !data.is_empty() => { + let preview = summarize_interaction_input(data); + format!("Interacted with `{command_display}`, sent `{preview}`") + } + _ => format!("Waited for `{command_display}`"), + } +} + +fn summarize_interaction_input(input: &str) -> String { + let single_line = input.replace('\n', "\\n"); + let sanitized = single_line.replace('`', "\\`"); + if sanitized.chars().count() <= MAX_INTERACTION_PREVIEW_CHARS { + return sanitized; + } + + let mut preview = String::new(); + for ch in sanitized.chars().take(MAX_INTERACTION_PREVIEW_CHARS) { + preview.push(ch); + } + preview.push_str("..."); + preview +} + #[derive(Clone)] pub(crate) struct OutputLines { pub(crate) lines: Vec>, @@ -181,7 +211,9 @@ impl HistoryCell for ExecCell { lines.extend(cmd_display); if let Some(output) = call.output.as_ref() { - lines.extend(output.formatted_output.lines().map(ansi_escape_line)); + if !call.is_unified_exec_interaction() { + lines.extend(output.formatted_output.lines().map(ansi_escape_line)); + } let duration = call .duration .map(format_duration) @@ -317,19 +349,29 @@ impl ExecCell { Some(false) => "•".red().bold(), None => spinner(call.start_time), }; - let title = if self.is_active() { + let is_interaction = call.is_unified_exec_interaction(); + let title = if is_interaction { + "" + } else if self.is_active() { "Running" - } else if call.is_user_shell_command { + } else if call.is_user_shell_command() { "You ran" } else { "Ran" }; - let mut header_line = - Line::from(vec![bullet.clone(), " ".into(), title.bold(), " ".into()]); + let mut header_line = if is_interaction { + Line::from(vec![bullet.clone(), " ".into()]) + } else { + Line::from(vec![bullet.clone(), " ".into(), title.bold(), " ".into()]) + }; let header_prefix_width = header_line.width(); - let cmd_display = strip_bash_lc_and_escape(&call.command); + let cmd_display = if call.is_unified_exec_interaction() { + format_unified_exec_interaction(&call.command, call.interaction_input.as_deref()) + } else { + strip_bash_lc_and_escape(&call.command) + }; let highlighted_lines = highlight_bash_to_lines(&cmd_display); let continuation_wrap_width = layout.command_continuation.wrap_width(width); @@ -373,7 +415,7 @@ impl ExecCell { } if let Some(output) = call.output.as_ref() { - let line_limit = if call.is_user_shell_command { + let line_limit = if call.is_user_shell_command() { USER_SHELL_TOOL_CALL_MAX_LINES } else { TOOL_CALL_MAX_LINES @@ -387,18 +429,20 @@ impl ExecCell { include_prefix: false, }, ); - let display_limit = if call.is_user_shell_command { + let display_limit = if call.is_user_shell_command() { USER_SHELL_TOOL_CALL_MAX_LINES } else { layout.output_max_lines }; if raw_output.lines.is_empty() { - lines.extend(prefix_lines( - vec![Line::from("(no output)".dim())], - Span::from(layout.output_block.initial_prefix).dim(), - Span::from(layout.output_block.subsequent_prefix), - )); + if !call.is_unified_exec_interaction() { + lines.extend(prefix_lines( + vec![Line::from("(no output)".dim())], + Span::from(layout.output_block.initial_prefix).dim(), + Span::from(layout.output_block.subsequent_prefix), + )); + } } else { let trimmed_output = Self::truncate_lines_middle( &raw_output.lines, diff --git a/codex-rs/tui/src/history_cell.rs b/codex-rs/tui/src/history_cell.rs index bb451f5a5..ffefdc6e4 100644 --- a/codex-rs/tui/src/history_cell.rs +++ b/codex-rs/tui/src/history_cell.rs @@ -1478,6 +1478,7 @@ mod tests { use serde_json::json; use std::collections::HashMap; + use codex_core::protocol::ExecCommandSource; use mcp_types::CallToolResult; use mcp_types::ContentBlock; use mcp_types::TextContent; @@ -1878,9 +1879,10 @@ mod tests { }, ], output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); // Mark call complete so markers are ✓ cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1)); @@ -1901,9 +1903,10 @@ mod tests { cmd: "rg shimmer_spans".into(), }], output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); // Call 1: Search only cell.complete_call("c1", CommandOutput::default(), Duration::from_millis(1)); @@ -1917,7 +1920,8 @@ mod tests { cmd: "cat shimmer.rs".into(), path: "shimmer.rs".into(), }], - false, + ExecCommandSource::Agent, + None, ) .unwrap(); cell.complete_call("c2", CommandOutput::default(), Duration::from_millis(1)); @@ -1931,7 +1935,8 @@ mod tests { cmd: "cat status_indicator_widget.rs".into(), path: "status_indicator_widget.rs".into(), }], - false, + ExecCommandSource::Agent, + None, ) .unwrap(); cell.complete_call("c3", CommandOutput::default(), Duration::from_millis(1)); @@ -1964,9 +1969,10 @@ mod tests { }, ], output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); cell.complete_call("c1", CommandOutput::default(), Duration::from_millis(1)); let lines = cell.display_lines(80); @@ -1984,9 +1990,10 @@ mod tests { command: vec!["bash".into(), "-lc".into(), cmd], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); // Mark call complete so it renders as "Ran" cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1)); @@ -2006,9 +2013,10 @@ mod tests { command: vec!["echo".into(), "ok".into()], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1)); // Wide enough that it fits inline @@ -2026,9 +2034,10 @@ mod tests { command: vec!["bash".into(), "-lc".into(), long], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1)); let lines = cell.display_lines(24); @@ -2045,9 +2054,10 @@ mod tests { command: vec!["bash".into(), "-lc".into(), cmd], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1)); let lines = cell.display_lines(80); @@ -2065,9 +2075,10 @@ mod tests { command: vec!["bash".into(), "-lc".into(), cmd], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); cell.complete_call(&call_id, CommandOutput::default(), Duration::from_millis(1)); let lines = cell.display_lines(28); @@ -2085,9 +2096,10 @@ mod tests { command: vec!["bash".into(), "-lc".into(), "seq 1 10 1>&2 && false".into()], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); let stderr: String = (1..=10) .map(|n| n.to_string()) @@ -2131,9 +2143,10 @@ mod tests { command: vec!["bash".into(), "-lc".into(), long_cmd.to_string()], parsed: Vec::new(), output: None, - is_user_shell_command: false, + source: ExecCommandSource::Agent, start_time: Some(Instant::now()), duration: None, + interaction_input: None, }); let stderr = "error: first line on stderr\nerror: second line on stderr".to_string(); diff --git a/codex-rs/tui/src/pager_overlay.rs b/codex-rs/tui/src/pager_overlay.rs index 82f43d84d..6393a754e 100644 --- a/codex-rs/tui/src/pager_overlay.rs +++ b/codex-rs/tui/src/pager_overlay.rs @@ -579,6 +579,7 @@ fn render_offset_content( #[cfg(test)] mod tests { use super::*; + use codex_core::protocol::ExecCommandSource; use codex_core::protocol::ReviewDecision; use insta::assert_snapshot; use std::collections::HashMap; @@ -719,7 +720,8 @@ mod tests { "exec-1".into(), vec!["bash".into(), "-lc".into(), "ls".into()], vec![ParsedCommand::Unknown { cmd: "ls".into() }], - false, + ExecCommandSource::Agent, + None, ); exec_cell.complete_call( "exec-1",