diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index d7c39c14a5..58cefb6c83 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -264,6 +264,9 @@ struct State { pending_approvals: HashMap>, pending_input: Vec, history: ConversationHistory, + /// True while awaiting a streaming response from the model API for the + /// current task/turn. + waiting_on_api_response: bool, } /// Context for an initialized model agent @@ -596,6 +599,8 @@ impl Session { } } + + /// Sends the given event to the client and swallows the send event, if /// any, logging it as an error. pub(crate) async fn send_event(&self, event: Event) { @@ -604,6 +609,24 @@ impl Session { } } + /// Mark that we are awaiting (or no longer awaiting) a model API response. + fn set_waiting_on_api_response(&self, waiting: bool) { + let mut state = self.state.lock_unchecked(); + state.waiting_on_api_response = waiting; + } + + /// Returns true if we are currently waiting on a model API response. + fn is_waiting_on_api_response(&self) -> bool { + self.state.lock_unchecked().waiting_on_api_response + } + + /// Helper that sets `waiting_on_api_response = true` and automatically + /// clears it when dropped. + fn begin_waiting_on_api_response(&self) -> WaitingApiGuard<'_> { + self.set_waiting_on_api_response(true); + WaitingApiGuard { sess: self } + } + pub async fn request_command_approval( &self, sub_id: String, @@ -888,8 +911,18 @@ impl Session { /// Returns the input if there was no task running to inject into pub fn inject_input(&self, input: Vec) -> Result<(), Vec> { + let waiting = self.is_waiting_on_api_response(); let mut state = self.state.lock_unchecked(); if state.current_task.is_some() { + // If we're in the middle of streaming a model response, abort the + // current task so we can immediately start a fresh request that + // includes everything recorded so far plus this new user input. + if waiting { + // Do not mutate pending_input here; instruct caller to spawn a new task. + return Err(input); + } + + // Otherwise, queue the input for inclusion on the next turn. state.pending_input.push(input.into()); Ok(()) } else { @@ -908,6 +941,12 @@ impl Session { } } + /// Returns true if there is pending input that should be included on the + /// next turn. Unlike `get_pending_input()`, this does not consume it. + pub fn has_pending_input(&self) -> bool { + !self.state.lock_unchecked().pending_input.is_empty() + } + pub async fn call_tool( &self, server: &str, @@ -966,6 +1005,17 @@ impl Drop for Session { } } +/// RAII guard that flips `waiting_on_api_response` back to false when dropped. +struct WaitingApiGuard<'a> { + sess: &'a Session, +} + +impl Drop for WaitingApiGuard<'_> { + fn drop(&mut self) { + self.sess.set_waiting_on_api_response(false); + } +} + #[derive(Clone, Debug)] pub(crate) struct ExecCommandContext { pub(crate) sub_id: String, @@ -1036,14 +1086,16 @@ impl AgentTask { // TOCTOU? if !self.handle.is_finished() { self.handle.abort(); - let event = Event { - id: self.sub_id, - msg: EventMsg::TurnAborted(TurnAbortedEvent { reason }), - }; - let tx_event = self.sess.tx_event.clone(); - tokio::spawn(async move { - tx_event.send(event).await.ok(); - }); + if matches!(reason, TurnAbortReason::Interrupted) { + let event = Event { + id: self.sub_id, + msg: EventMsg::TurnAborted(TurnAbortedEvent { reason }), + }; + let tx_event = self.sess.tx_event.clone(); + tokio::spawn(async move { + tx_event.send(event).await.ok(); + }); + } } } } @@ -1572,10 +1624,20 @@ async fn run_task( } if responses.is_empty() { - debug!("Turn completed"); + // Turn completed with an assistant message. last_agent_message = get_last_assistant_message_from_turn( &items_to_record_in_conversation_history, ); + + // If there is pending input that arrived during this + // turn, keep the task alive and start another turn so the + // pending input is sent immediately. + if sess.has_pending_input() { + debug!("Turn completed; pending input present — continuing"); + continue; + } + + debug!("Turn completed"); sess.maybe_notify(UserNotification::AgentTurnComplete { turn_id: sub_id.clone(), input_messages: turn_input_messages, @@ -1739,6 +1801,8 @@ async fn try_run_turn( }) }; + // We are now awaiting a response from the model API. + let _waiting_guard = sess.begin_waiting_on_api_response(); let mut stream = turn_context.client.clone().stream(&prompt).await?; let mut output = Vec::new(); diff --git a/codex-rs/core/src/config_types.rs b/codex-rs/core/src/config_types.rs index 88be151fff..0a71617871 100644 --- a/codex-rs/core/src/config_types.rs +++ b/codex-rs/core/src/config_types.rs @@ -74,9 +74,33 @@ pub enum HistoryPersistence { None, } +/// Behavior for handling a user message submitted while a turn is in progress. +#[derive(Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)] +#[serde(rename_all = "lowercase")] +pub enum MessageDuringTurnBehavior { + /// Queue the message to be sent after the current turn completes. + #[default] + Queue, + /// Interrupt the current model stream. If mid tool-call, do not interrupt; + /// send the user message so it will be included with the tool outputs on + /// the next turn. + Interrupt, +} + /// Collection of settings that are specific to the TUI. -#[derive(Deserialize, Debug, Clone, PartialEq, Default)] -pub struct Tui {} +#[derive(Deserialize, Debug, Clone, PartialEq)] +pub struct Tui { + #[serde(default)] + pub message_during_turn_behavior: MessageDuringTurnBehavior, +} + +impl Default for Tui { + fn default() -> Self { + Self { + message_during_turn_behavior: MessageDuringTurnBehavior::Queue, + } + } +} #[derive(Deserialize, Debug, Clone, PartialEq, Default)] pub struct SandboxWorkspaceWrite { diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 57b29262d2..a45fc87d76 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -74,9 +74,10 @@ pub(crate) async fn handle_mcp_tool_call( } async fn notify_mcp_tool_call_event(sess: &Session, sub_id: &str, event: EventMsg) { - sess.send_event(Event { - id: sub_id.to_string(), - msg: event, - }) - .await; + sess + .send_event(Event { + id: sub_id.to_string(), + msg: event, + }) + .await; } diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 4695c6566c..6cb9cedc4c 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -79,6 +79,7 @@ use codex_common::model_presets::builtin_model_presets; use codex_core::ConversationManager; use codex_core::protocol::AskForApproval; use codex_core::protocol::SandboxPolicy; +use codex_core::config_types::MessageDuringTurnBehavior; use codex_core::protocol_config_types::ReasoningEffort as ReasoningEffortConfig; use codex_file_search::FileMatch; use uuid::Uuid; @@ -116,6 +117,7 @@ pub(crate) struct ChatWidget { last_history_was_exec: bool, // User messages queued while a turn is in progress queued_user_messages: VecDeque, + // No TUI-level tracking; core decides mid-tool-call behavior. } struct UserMessage { @@ -272,10 +274,10 @@ impl ChatWidget { /// When there are queued user messages, restore them into the composer /// separated by newlines rather than auto‑submitting the next one. fn on_interrupted_turn(&mut self) { - // Finalize, log a gentle prompt, and clear running state. + // User-initiated interrupt: finalize with a gentle prompt + // and restore any queued messages into the composer. self.finalize_turn_with_error_message("Tell the model what to do differently".to_owned()); - // If any messages were queued during the task, restore them into the composer. if !self.queued_user_messages.is_empty() { let combined = self .queued_user_messages @@ -284,7 +286,6 @@ impl ChatWidget { .collect::>() .join("\n"); self.bottom_pane.set_composer_text(combined); - // Clear the queue and update the status indicator list. self.queued_user_messages.clear(); self.refresh_queued_user_messages(); } @@ -342,6 +343,7 @@ impl ChatWidget { |q| q.push_patch_end(event), |s| s.handle_patch_apply_end_now(ev2), ); + // No TUI-level tracking } fn on_exec_command_end(&mut self, ev: ExecCommandEndEvent) { @@ -727,14 +729,24 @@ impl ChatWidget { _ => { match self.bottom_pane.handle_key_event(key_event) { InputResult::Submitted(text) => { - // If a task is running, queue the user input to be sent after the turn completes. + // Build the message to submit or defer. let user_message = UserMessage { text, image_paths: self.bottom_pane.take_recent_submission_images(), }; if self.bottom_pane.is_task_running() { - self.queued_user_messages.push_back(user_message); - self.refresh_queued_user_messages(); + match self.config.tui.message_during_turn_behavior { + MessageDuringTurnBehavior::Queue => { + // Original behavior: queue for after the turn completes. + self.queued_user_messages.push_back(user_message); + self.refresh_queued_user_messages(); + } + MessageDuringTurnBehavior::Interrupt => { + // Forward to Codex and print immediately; Core will either + // queue alongside tool outputs or replace the running turn. + self.submit_user_message(user_message); + } + } } else { self.submit_user_message(user_message); } @@ -1032,6 +1044,8 @@ impl ChatWidget { } } + // No TUI-level mid-tool-call logic; Core manages it. + // If idle and there are queued inputs, submit exactly one to start the next turn. fn maybe_send_next_queued_input(&mut self) { if self.bottom_pane.is_task_running() {