diff --git a/codex-rs/exec/src/event_processor_with_jsonl_output.rs b/codex-rs/exec/src/event_processor_with_jsonl_output.rs index 348d93505b..76e73e82a4 100644 --- a/codex-rs/exec/src/event_processor_with_jsonl_output.rs +++ b/codex-rs/exec/src/event_processor_with_jsonl_output.rs @@ -200,22 +200,28 @@ impl EventProcessorWithJsonOutput { vec![ThreadEvent::ItemCompleted(ItemCompletedEvent { item })] } - fn handle_exec_command_begin(&mut self, ev: &ExecCommandBeginEvent) -> Vec { + + fn start_command_item( + &mut self, + call_id: &str, + command: &[String], + warn_subject: &'static str, + ) -> Vec { let item_id = self.get_next_item_id(); - let command_string = match shlex::try_join(ev.command.iter().map(String::as_str)) { + let command_string = match shlex::try_join(command.iter().map(String::as_str)) { Ok(command_string) => command_string, Err(e) => { warn!( - call_id = ev.call_id, - "Failed to stringify command: {e:?}; skipping item.started" + call_id = call_id, + "Failed to stringify {warn_subject}: {e:?}; skipping item.started" ); - ev.command.join(" ") + command.join(" ") } }; self.running_commands.insert( - ev.call_id.clone(), + call_id.to_string(), RunningCommand { command: command_string.clone(), item_id: item_id.clone(), @@ -235,6 +241,45 @@ impl EventProcessorWithJsonOutput { vec![ThreadEvent::ItemStarted(ItemStartedEvent { item })] } + fn finish_command_item( + &mut self, + call_id: &str, + aggregated_output: &str, + exit_code: i32, + warn_event_names: (&'static str, &'static str), + ) -> Vec { + let some_running_command = self.running_commands.remove(call_id); + let Some(RunningCommand { command, item_id }) = some_running_command else { + let (end_event, begin_event) = warn_event_names; + warn!( + call_id = call_id, + "{end_event} without matching {begin_event}; skipping item.completed" + ); + return Vec::new(); + }; + + let status = if exit_code == 0 { + CommandExecutionStatus::Completed + } else { + CommandExecutionStatus::Failed + }; + + let item = ThreadItem { + id: item_id, + details: ThreadItemDetails::CommandExecution(CommandExecutionItem { + command, + aggregated_output: aggregated_output.to_string(), + exit_code: Some(exit_code), + status, + }), + }; + + vec![ThreadEvent::ItemCompleted(ItemCompletedEvent { item })] + } + fn handle_exec_command_begin(&mut self, ev: &ExecCommandBeginEvent) -> Vec { + self.start_command_item(&ev.call_id, &ev.command, "command") + } + fn handle_mcp_tool_call_begin(&mut self, ev: &McpToolCallBeginEvent) -> Vec { let item_id = self.get_next_item_id(); let server = ev.invocation.server.clone(); @@ -327,67 +372,16 @@ impl EventProcessorWithJsonOutput { } fn handle_user_command_begin(&mut self, ev: &UserCommandBeginEvent) -> Vec { - let item_id = self.get_next_item_id(); - - let command_string = match shlex::try_join(ev.command.iter().map(String::as_str)) { - Ok(command_string) => command_string, - Err(e) => { - warn!( - call_id = ev.call_id, - "Failed to stringify user command: {e:?}; skipping item.started" - ); - ev.command.join(" ") - } - }; - - self.running_commands.insert( - ev.call_id.clone(), - RunningCommand { - command: command_string.clone(), - item_id: item_id.clone(), - }, - ); - - let item = ThreadItem { - id: item_id, - details: ThreadItemDetails::CommandExecution(CommandExecutionItem { - command: command_string, - aggregated_output: String::new(), - exit_code: None, - status: CommandExecutionStatus::InProgress, - }), - }; - - vec![ThreadEvent::ItemStarted(ItemStartedEvent { item })] + self.start_command_item(&ev.call_id, &ev.command, "user command") } fn handle_user_command_end(&mut self, ev: &UserCommandEndEvent) -> Vec { - let Some(RunningCommand { command, item_id }) = self.running_commands.remove(&ev.call_id) - else { - warn!( - call_id = ev.call_id, - "UserCommandEnd without matching UserCommandBegin; skipping item.completed" - ); - return Vec::new(); - }; - - let status = if ev.exit_code == 0 { - CommandExecutionStatus::Completed - } else { - CommandExecutionStatus::Failed - }; - - let item = ThreadItem { - id: item_id, - details: ThreadItemDetails::CommandExecution(CommandExecutionItem { - command, - aggregated_output: ev.aggregated_output.clone(), - exit_code: Some(ev.exit_code), - status, - }), - }; - - vec![ThreadEvent::ItemCompleted(ItemCompletedEvent { item })] + self.finish_command_item( + &ev.call_id, + &ev.aggregated_output, + ev.exit_code, + ("UserCommandEnd", "UserCommandBegin"), + ) } fn handle_patch_apply_begin(&mut self, ev: &PatchApplyBeginEvent) -> Vec { @@ -435,31 +429,12 @@ impl EventProcessorWithJsonOutput { } fn handle_exec_command_end(&mut self, ev: &ExecCommandEndEvent) -> Vec { - let Some(RunningCommand { command, item_id }) = self.running_commands.remove(&ev.call_id) - else { - warn!( - call_id = ev.call_id, - "ExecCommandEnd without matching ExecCommandBegin; skipping item.completed" - ); - return Vec::new(); - }; - let status = if ev.exit_code == 0 { - CommandExecutionStatus::Completed - } else { - CommandExecutionStatus::Failed - }; - let item = ThreadItem { - id: item_id, - - details: ThreadItemDetails::CommandExecution(CommandExecutionItem { - command, - aggregated_output: ev.aggregated_output.clone(), - exit_code: Some(ev.exit_code), - status, - }), - }; - - vec![ThreadEvent::ItemCompleted(ItemCompletedEvent { item })] + self.finish_command_item( + &ev.call_id, + &ev.aggregated_output, + ev.exit_code, + ("ExecCommandEnd", "ExecCommandBegin"), + ) } fn todo_items_from_plan(&self, args: &UpdatePlanArgs) -> Vec {