diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 8f25074d7..61ef994ac 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -704,22 +704,6 @@ impl AgentControl { result } - /// Append a prebuilt message to an existing agent thread outside the normal user-input path. - #[cfg(test)] - pub(crate) async fn append_message( - &self, - agent_id: ThreadId, - message: ResponseItem, - ) -> CodexResult { - let state = self.upgrade()?; - self.handle_thread_request_result( - agent_id, - &state, - state.append_message(agent_id, message).await, - ) - .await - } - pub(crate) async fn send_inter_agent_communication( &self, agent_id: ThreadId, diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index b5ba938ef..43a6e47bf 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -518,60 +518,6 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig )); } -#[tokio::test] -async fn append_message_records_assistant_message() { - let harness = AgentControlHarness::new().await; - let (thread_id, thread) = harness.start_thread().await; - let message = - "author: /root\nrecipient: /root/worker\nother_recipients: []\nContent: hello from tests"; - - let submission_id = harness - .control - .append_message( - thread_id, - ResponseItem::Message { - id: None, - role: "assistant".to_string(), - content: vec![ContentItem::InputText { - text: message.to_string(), - }], - phase: None, - }, - ) - .await - .expect("append_message should succeed"); - assert!(!submission_id.is_empty()); - - timeout(Duration::from_secs(5), async { - loop { - let history_items = thread - .codex - .session - .clone_history() - .await - .raw_items() - .to_vec(); - let recorded = history_items.iter().any(|item| { - matches!( - item, - ResponseItem::Message { role, content, .. } - if role == "assistant" - && content.iter().any(|content_item| matches!( - content_item, - ContentItem::InputText { text } if text == message - )) - ) - }); - if recorded { - break; - } - sleep(Duration::from_millis(10)).await; - } - }) - .await - .expect("assistant message should be recorded"); -} - #[tokio::test] async fn spawn_agent_creates_thread_and_sends_prompt() { let harness = AgentControlHarness::new().await; diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index cec34c67b..ce8000de6 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -260,7 +260,12 @@ impl CodexThread { &self, items: Vec, ) -> Result<(), Vec> { - self.codex.session.inject_response_items(items).await + let response_items = items.iter().cloned().map(ResponseItem::from).collect(); + self.codex + .session + .inject_if_running(response_items) + .await + .map_err(|_| items) } pub async fn set_app_server_client_info( @@ -366,58 +371,16 @@ impl CodexThread { /// Records a user-role session-prefix message without creating a new user turn boundary. pub(crate) async fn inject_user_message_without_turn(&self, message: String) { - let message = ResponseItem::Message { + let item = ResponseItem::Message { id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: message }], phase: None, }; - let pending_item = match pending_message_input_item(&message) { - Ok(pending_item) => pending_item, - Err(err) => { - debug_assert!(false, "session-prefix message append should succeed: {err}"); - return; - } - }; - if self - .codex + self.codex .session - .inject_response_items(vec![pending_item]) - .await - .is_err() - { - let turn_context = self.codex.session.new_default_turn().await; - self.codex - .session - .record_conversation_items(turn_context.as_ref(), &[message]) - .await; - } - } - - /// Append a prebuilt message to the thread history without treating it as a user turn. - /// - /// If the thread already has an active turn, the message is queued as pending input for that - /// turn. Otherwise it is queued at session scope and a regular turn is started so the agent - /// can consume that pending input through the normal turn pipeline. - #[cfg(test)] - pub(crate) async fn append_message(&self, message: ResponseItem) -> CodexResult { - let submission_id = uuid::Uuid::new_v4().to_string(); - let pending_item = pending_message_input_item(&message)?; - if let Err(items) = self - .codex - .session - .inject_response_items(vec![pending_item]) - .await - { - self.codex - .session - .input_queue - .queue_response_items_for_next_turn(items) - .await; - self.codex.session.maybe_start_turn_for_pending_work().await; - } - - Ok(submission_id) + .inject_no_new_turn(vec![item], /*current_turn_context*/ None) + .await; } /// Append raw Responses API items to the thread's model-visible history. @@ -437,7 +400,7 @@ impl CodexThread { } self.codex .session - .record_conversation_items(turn_context.as_ref(), &items) + .inject_no_new_turn(items, Some(turn_context.as_ref())) .await; self.codex.session.flush_rollout().await?; Ok(()) @@ -598,21 +561,3 @@ impl CodexThread { Ok(*guard) } } - -fn pending_message_input_item(message: &ResponseItem) -> CodexResult { - match message { - ResponseItem::Message { - role, - content, - phase, - .. - } => Ok(ResponseInputItem::Message { - role: role.clone(), - content: content.clone(), - phase: phase.clone(), - }), - _ => Err(CodexErr::InvalidRequest( - "append_message only supports ResponseItem::Message".to_string(), - )), - } -} diff --git a/codex-rs/core/src/compact.rs b/codex-rs/core/src/compact.rs index 1a6256ad7..fd388b273 100644 --- a/codex-rs/core/src/compact.rs +++ b/codex-rs/core/src/compact.rs @@ -568,7 +568,7 @@ async fn drain_to_completed( }; match event { Ok(ResponseEvent::OutputItemDone(item)) => { - sess.record_into_history(std::slice::from_ref(&item), turn_context) + sess.record_conversation_items(turn_context, std::slice::from_ref(&item)) .await; } Ok(ResponseEvent::ServerReasoningIncluded(included)) => { diff --git a/codex-rs/core/src/goals.rs b/codex-rs/core/src/goals.rs index c8e76f7d4..ece020385 100644 --- a/codex-rs/core/src/goals.rs +++ b/codex-rs/core/src/goals.rs @@ -26,7 +26,7 @@ use codex_otel::GOAL_TOKEN_COUNT_METRIC; use codex_otel::GOAL_USAGE_LIMITED_METRIC; use codex_protocol::ThreadId; use codex_protocol::config_types::ModeKind; -use codex_protocol::models::ResponseInputItem; +use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ThreadGoal; use codex_protocol::protocol::ThreadGoalStatus; @@ -177,7 +177,7 @@ pub(crate) struct GoalRuntimeState { struct GoalContinuationCandidate { goal_id: String, - items: Vec, + items: Vec, } impl GoalRuntimeState { @@ -686,7 +686,7 @@ impl Session { .await; if let Some(goal) = goal_for_steering { let item = goal_context_input_item(objective_updated_prompt(&goal)); - if self.inject_response_items(vec![item]).await.is_err() { + if self.inject_if_running(vec![item]).await.is_err() { tracing::debug!( "skipping objective-updated goal steering because no turn is active" ); @@ -1074,7 +1074,7 @@ impl Session { .await; if should_steer_budget_limit { let item = budget_limit_steering_item(&goal); - if self.inject_response_items(vec![item]).await.is_err() { + if self.inject_if_running(vec![item]).await.is_err() { tracing::debug!("skipping budget-limit goal steering because no turn is active"); } *self.goal_runtime.budget_limit_reported_goal_id.lock().await = Some(goal_id); @@ -1332,7 +1332,7 @@ impl Session { candidate .items .into_iter() - .map(TurnInput::ResponseInputItem) + .map(TurnInput::ResponseItem) .collect(), ) .await; @@ -1371,14 +1371,6 @@ impl Session { tracing::debug!("skipping active goal continuation because a turn is already active"); return None; } - if self - .input_queue - .has_queued_response_items_for_next_turn() - .await - { - tracing::debug!("skipping active goal continuation because queued input exists"); - return None; - } if self.input_queue.has_trigger_turn_mailbox_items().await { tracing::debug!( "skipping active goal continuation because trigger-turn mailbox input is pending" @@ -1416,10 +1408,6 @@ impl Session { return None; } if self.active_turn.lock().await.is_some() - || self - .input_queue - .has_queued_response_items_for_next_turn() - .await || self.input_queue.has_trigger_turn_mailbox_items().await { tracing::debug!("skipping active goal continuation because pending work appeared"); @@ -1604,12 +1592,12 @@ fn escape_xml_text(input: &str) -> String { .replace('>', ">") } -fn budget_limit_steering_item(goal: &ThreadGoal) -> ResponseInputItem { +fn budget_limit_steering_item(goal: &ThreadGoal) -> ResponseItem { goal_context_input_item(budget_limit_prompt(goal)) } -fn goal_context_input_item(prompt: String) -> ResponseInputItem { - GoalContext::new(prompt).into_response_input_item() +fn goal_context_input_item(prompt: String) -> ResponseItem { + ResponseItem::from(GoalContext::new(prompt).into_response_input_item()) } pub(crate) fn protocol_goal_from_state(goal: codex_state::ThreadGoal) -> ThreadGoal { @@ -1678,7 +1666,7 @@ mod tests { use codex_protocol::ThreadId; use codex_protocol::config_types::ModeKind; use codex_protocol::models::ContentItem; - use codex_protocol::models::ResponseInputItem; + use codex_protocol::models::ResponseItem; use codex_protocol::protocol::ThreadGoal; use codex_protocol::protocol::ThreadGoalStatus; use codex_protocol::protocol::TokenUsage; @@ -1807,7 +1795,8 @@ mod tests { assert_eq!( item, - ResponseInputItem::Message { + ResponseItem::Message { + id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "\nContinue working.\n".to_string(), diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index b0eb30307..f52792cb6 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -790,12 +790,11 @@ async fn run_review_on_session( } async fn append_guardian_followup_reminder(review_session: &GuardianReviewSession) { - let turn_context = review_session.codex.session.new_default_turn().await; let reminder: ResponseItem = ContextualUserFragment::into(GuardianFollowupReviewReminder); review_session .codex .session - .record_conversation_items(turn_context.as_ref(), std::slice::from_ref(&reminder)) + .inject_no_new_turn(vec![reminder], /*current_turn_context*/ None) .await; } diff --git a/codex-rs/core/src/guardian/tests.rs b/codex-rs/core/src/guardian/tests.rs index 865b1cac5..6df3b69f6 100644 --- a/codex-rs/core/src/guardian/tests.rs +++ b/codex-rs/core/src/guardian/tests.rs @@ -191,7 +191,8 @@ async fn guardian_test_session_and_turn_with_base_url( async fn seed_guardian_parent_history(session: &Arc, turn: &Arc) { session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -225,7 +226,6 @@ async fn seed_guardian_parent_history(session: &Arc, turn: &Arc anyh let (session, turn) = guardian_test_session_and_turn_with_base_url("http://localhost").await; seed_guardian_parent_history(&session, &turn).await; session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -381,7 +382,6 @@ async fn build_guardian_prompt_delta_mode_preserves_original_numbering() -> anyh phase: None, }, ], - turn.as_ref(), ) .await; @@ -516,7 +516,8 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() - ) .await; session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -535,7 +536,6 @@ async fn build_guardian_prompt_stale_delta_version_falls_back_to_full_prompt() - phase: None, }, ], - turn.as_ref(), ) .await; @@ -1469,7 +1469,8 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: ) .await; session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -1488,7 +1489,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: phase: None, }, ], - turn.as_ref(), ) .await; let second_request = GuardianApprovalRequest::Shell { @@ -1513,7 +1513,8 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: ) .await; session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -1532,7 +1533,6 @@ async fn guardian_reuses_prompt_cache_key_and_appends_prior_reviews() -> anyhow: phase: None, }, ], - turn.as_ref(), ) .await; let third_request = GuardianApprovalRequest::Shell { @@ -2005,7 +2005,8 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a ReviewDecision::Approved ); session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -2024,7 +2025,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a phase: None, }, ], - turn.as_ref(), ) .await; @@ -2072,7 +2072,8 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a "second guardian request was not observed" ); session - .record_into_history( + .record_conversation_items( + turn.as_ref(), &[ ResponseItem::Message { id: None, @@ -2091,7 +2092,6 @@ async fn guardian_parallel_reviews_fork_from_last_committed_trunk_history() -> a phase: None, }, ], - turn.as_ref(), ) .await; diff --git a/codex-rs/core/src/hook_runtime.rs b/codex-rs/core/src/hook_runtime.rs index 6452e9d5a..56d51412c 100644 --- a/codex-rs/core/src/hook_runtime.rs +++ b/codex-rs/core/src/hook_runtime.rs @@ -522,7 +522,7 @@ pub(crate) async fn inspect_pending_input( ) .await } - TurnInput::ResponseInputItem(_) => HookRuntimeOutcome { + TurnInput::ResponseItem(_) => HookRuntimeOutcome { should_stop: false, additional_contexts: Vec::new(), }, @@ -540,9 +540,8 @@ pub(crate) async fn record_pending_input( sess.record_user_prompt_and_emit_turn_item(turn_context.as_ref(), content.as_slice()) .await; } - TurnInput::ResponseInputItem(input) => { - let response_item = ResponseItem::from(input); - sess.record_conversation_items(turn_context, std::slice::from_ref(&response_item)) + TurnInput::ResponseItem(item) => { + sess.record_conversation_items(turn_context, std::slice::from_ref(&item)) .await; } } diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index cf46ecba9..0a72d2723 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -27,6 +27,7 @@ use crate::tasks::UserShellCommandTask; use crate::tasks::execute_user_shell_command; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseInputItem; +use codex_protocol::models::ResponseItem; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; @@ -255,7 +256,8 @@ pub(super) async fn user_input_or_turn_inner( }; let mut task_input = additional_context_input .into_iter() - .map(TurnInput::ResponseInputItem) + .map(ResponseItem::from) + .map(TurnInput::ResponseItem) .collect::>(); if !items.is_empty() { task_input.push(TurnInput::UserInput(items)); @@ -899,17 +901,14 @@ Do not assume this also authorizes similar operations with different payloads. Approved action: {approved_action_json}"#, ); - let items = vec![ResponseInputItem::Message { + let items = vec![ResponseItem::from(ResponseInputItem::Message { role: "developer".to_string(), content: vec![ContentItem::InputText { text }], phase: None, - }]; + })]; - if let Err(items) = sess.inject_response_items(items).await { - sess.input_queue - .queue_response_items_for_next_turn(items) - .await; - } + sess.inject_no_new_turn(items, /*current_turn_context*/ None) + .await; } pub(super) fn submission_dispatch_span(sub: &Submission) -> tracing::Span { diff --git a/codex-rs/core/src/session/inject.rs b/codex-rs/core/src/session/inject.rs new file mode 100644 index 000000000..4a7189bbd --- /dev/null +++ b/codex-rs/core/src/session/inject.rs @@ -0,0 +1,50 @@ +use super::input_queue::TurnInput; +use super::session::Session; +use super::turn_context::TurnContext; +use codex_protocol::models::ResponseItem; + +impl Session { + /// Returns the input if there is no active turn to inject into. + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] + pub async fn inject_if_running( + &self, + input: Vec, + ) -> Result<(), Vec> { + let mut active = self.active_turn.lock().await; + match active.as_mut() { + Some(active_turn) => { + self.input_queue + .extend_pending_input_for_turn_state( + active_turn.turn_state.as_ref(), + input.into_iter().map(TurnInput::ResponseItem).collect(), + ) + .await; + Ok(()) + } + None => Err(input), + } + } + + /// Injects items into active work, or records them without starting a turn. + pub(crate) async fn inject_no_new_turn( + &self, + items: Vec, + current_turn_context: Option<&TurnContext>, + ) { + let Err(items) = self.inject_if_running(items).await else { + return; + }; + let default_turn_context; + let turn_context = match current_turn_context { + Some(turn_context) => turn_context, + None => { + default_turn_context = self.new_default_turn().await; + default_turn_context.as_ref() + } + }; + self.record_conversation_items(turn_context, &items).await; + } +} diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index 620c410eb..e317ba57a 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/codex-rs/core/src/session/input_queue.rs @@ -1,7 +1,7 @@ use crate::state::ActiveTurn; use crate::state::MailboxDeliveryPhase; use crate::state::TurnState; -use codex_protocol::models::ResponseInputItem; +use codex_protocol::models::ResponseItem; use codex_protocol::protocol::InterAgentCommunication; use codex_protocol::user_input::UserInput; use std::collections::VecDeque; @@ -12,7 +12,7 @@ use tokio::sync::watch; #[derive(Clone, Debug, PartialEq)] pub(crate) enum TurnInput { UserInput(Vec), - ResponseInputItem(ResponseInputItem), + ResponseItem(ResponseItem), } /// Turn-local pending input storage owned by the input queue flow. @@ -25,8 +25,6 @@ pub(crate) struct TurnInputQueue { pub(crate) struct InputQueue { mailbox_tx: watch::Sender<()>, mailbox_pending_mails: Mutex>, - - idle_pending_input: Mutex>, } impl InputQueue { @@ -35,7 +33,6 @@ impl InputQueue { Self { mailbox_tx, mailbox_pending_mails: Mutex::new(VecDeque::new()), - idle_pending_input: Mutex::new(Vec::new()), } } @@ -70,31 +67,15 @@ impl InputQueue { .any(|mail| mail.trigger_turn) } - pub(crate) async fn drain_mailbox_input_items(&self) -> Vec { + pub(crate) async fn drain_mailbox_input_items(&self) -> Vec { self.mailbox_pending_mails .lock() .await .drain(..) - .map(|mail| mail.to_response_input_item()) + .map(|mail| ResponseItem::from(mail.to_response_input_item())) .collect() } - pub(crate) async fn queue_response_items_for_next_turn(&self, items: Vec) { - if items.is_empty() { - return; - } - - self.idle_pending_input.lock().await.extend(items); - } - - pub(crate) async fn take_queued_response_items_for_next_turn(&self) -> Vec { - std::mem::take(&mut *self.idle_pending_input.lock().await) - } - - pub(crate) async fn has_queued_response_items_for_next_turn(&self) -> bool { - !self.idle_pending_input.lock().await.is_empty() - } - pub(crate) async fn turn_state_for_sub_id( &self, active_turn: &Mutex>, @@ -181,32 +162,6 @@ impl InputQueue { turn_state.lock().await.pending_input.items.split_off(0) } - #[expect( - clippy::await_holding_invalid_type, - reason = "active turn checks and turn state updates must remain atomic" - )] - pub(crate) async fn inject_response_items( - &self, - active_turn: &Mutex>, - input: Vec, - ) -> Result<(), Vec> { - let mut active = active_turn.lock().await; - match active.as_mut() { - Some(active_turn) => { - self.extend_pending_input_for_turn_state( - active_turn.turn_state.as_ref(), - input - .into_iter() - .map(TurnInput::ResponseInputItem) - .collect(), - ) - .await; - Ok(()) - } - None => Err(input), - } - } - #[expect( clippy::await_holding_invalid_type, reason = "active turn checks and turn state updates must remain atomic" @@ -235,7 +190,7 @@ impl InputQueue { .drain_mailbox_input_items() .await .into_iter() - .map(TurnInput::ResponseInputItem); + .map(TurnInput::ResponseItem); if pending_input.is_empty() { mailbox_items.collect() } else { @@ -345,8 +300,8 @@ mod tests { assert_eq!( input_queue.drain_mailbox_input_items().await, vec![ - mail_one.to_response_input_item(), - mail_two.to_response_input_item() + ResponseItem::from(mail_one.to_response_input_item()), + ResponseItem::from(mail_two.to_response_input_item()) ] ); assert!(!input_queue.has_pending_mailbox_items().await); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 512f8acc7..633163566 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -196,6 +196,7 @@ use codex_protocol::exec_output::StreamOutput; mod config_lock; mod handlers; +mod inject; mod input_queue; mod mcp; mod multi_agents; @@ -324,7 +325,6 @@ use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::Personality; use codex_protocol::config_types::ReasoningSummary as ReasoningSummaryConfig; use codex_protocol::config_types::WindowsSandboxLevel; -use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort as ReasoningEffortConfig; @@ -1842,27 +1842,11 @@ impl Session { warn!("execpolicy amendment for {sub_id} had no command prefix"); return; }; - let fragment = ApprovedCommandPrefixSaved::new(prefixes); - let text = fragment.render(); - let message: ResponseItem = ContextualUserFragment::into(fragment); - - if let Some(turn_context) = self.turn_context_for_sub_id(sub_id).await { - self.record_conversation_items(&turn_context, std::slice::from_ref(&message)) - .await; - return; - } - - if self - .inject_response_items(vec![ResponseInputItem::Message { - role: "developer".to_string(), - content: vec![ContentItem::InputText { text }], - phase: None, - }]) - .await - .is_err() - { - warn!("no active turn found to record execpolicy amendment message for {sub_id}"); - } + let message: ResponseItem = + ContextualUserFragment::into(ApprovedCommandPrefixSaved::new(prefixes)); + let turn_context = self.turn_context_for_sub_id(sub_id).await; + self.inject_no_new_turn(vec![message], turn_context.as_deref()) + .await; } pub(crate) async fn persist_network_policy_amendment( @@ -1939,27 +1923,10 @@ impl Session { sub_id: &str, amendment: &NetworkPolicyAmendment, ) { - let fragment = NetworkRuleSaved::new(amendment); - let text = fragment.render(); - let message: ResponseItem = ContextualUserFragment::into(fragment); - - if let Some(turn_context) = self.turn_context_for_sub_id(sub_id).await { - self.record_conversation_items(&turn_context, std::slice::from_ref(&message)) - .await; - return; - } - - if self - .inject_response_items(vec![ResponseInputItem::Message { - role: "developer".to_string(), - content: vec![ContentItem::InputText { text }], - phase: None, - }]) - .await - .is_err() - { - warn!("no active turn found to record network policy amendment message for {sub_id}"); - } + let message: ResponseItem = ContextualUserFragment::into(NetworkRuleSaved::new(amendment)); + let turn_context = self.turn_context_for_sub_id(sub_id).await; + self.inject_no_new_turn(vec![message], turn_context.as_deref()) + .await; } /// Emit an exec approval request event and await the user's decision. @@ -2510,28 +2477,21 @@ impl Session { } } - /// Records input items: always append to conversation history and - /// persist these response items to rollout. + /// Records conversation items: append to history, persist to rollout, and + /// notify clients observing raw response items. pub(crate) async fn record_conversation_items( &self, turn_context: &TurnContext, items: &[ResponseItem], ) { - self.record_into_history(items, turn_context).await; + { + let mut state = self.state.lock().await; + state.record_items(items.iter(), turn_context.truncation_policy); + } self.persist_rollout_response_items(items).await; self.send_raw_response_items(turn_context, items).await; } - /// Append ResponseItems to the in-memory conversation history only. - pub(crate) async fn record_into_history( - &self, - items: &[ResponseItem], - turn_context: &TurnContext, - ) { - let mut state = self.state.lock().await; - state.record_items(items.iter(), turn_context.truncation_policy); - } - async fn maybe_warn_on_server_model_mismatch( self: &Arc, turn_context: &Arc, @@ -3192,7 +3152,8 @@ impl Session { let mut pending_input = additional_context_input .into_iter() - .map(TurnInput::ResponseInputItem) + .map(ResponseItem::from) + .map(TurnInput::ResponseItem) .collect::>(); pending_input.push(TurnInput::UserInput(input)); self.input_queue @@ -3204,16 +3165,6 @@ impl Session { Ok(active_turn_id.clone()) } - /// Returns the input if there was no task running to inject into. - pub async fn inject_response_items( - &self, - input: Vec, - ) -> Result<(), Vec> { - self.input_queue - .inject_response_items(&self.active_turn, input) - .await - } - pub(crate) async fn record_memory_citation_for_turn(&self, sub_id: &str) { let turn_state = self .input_queue diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ca414cd18..7cca56dfb 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -103,7 +103,6 @@ use codex_protocol::config_types::ModeKind; use codex_protocol::config_types::Settings; use codex_protocol::models::BaseInstructions; use codex_protocol::models::ContentItem; -use codex_protocol::models::ResponseInputItem; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::AskForApproval; use codex_protocol::protocol::CodexErrorInfo; @@ -399,6 +398,12 @@ async fn interrupting_regular_turn_waiting_on_startup_prewarm_emits_turn_aborted sess.abort_all_tasks(TurnAbortReason::Interrupted).await; + let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("expected turn aborted marker event") + .expect("channel open"); + assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_))); + let second = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("expected turn aborted event") @@ -1773,7 +1778,7 @@ async fn recompute_token_usage_uses_session_base_instructions() { let item = user_message("hello"); session - .record_into_history(std::slice::from_ref(&item), &turn_context) + .record_conversation_items(&turn_context, std::slice::from_ref(&item)) .await; let history = session.clone_history().await; @@ -2627,7 +2632,7 @@ async fn thread_rollback_fails_without_persisted_thread_history() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = sess.build_initial_context(tc.as_ref()).await; - sess.record_into_history(&initial_context, tc.as_ref()) + sess.record_conversation_items(tc.as_ref(), &initial_context) .await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 1).await; @@ -3000,7 +3005,7 @@ async fn thread_rollback_fails_when_turn_in_progress() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = sess.build_initial_context(tc.as_ref()).await; - sess.record_into_history(&initial_context, tc.as_ref()) + sess.record_conversation_items(tc.as_ref(), &initial_context) .await; *sess.active_turn.lock().await = Some(crate::state::ActiveTurn::default()); @@ -3021,7 +3026,7 @@ async fn thread_rollback_fails_when_num_turns_is_zero() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let initial_context = sess.build_initial_context(tc.as_ref()).await; - sess.record_into_history(&initial_context, tc.as_ref()) + sess.record_conversation_items(tc.as_ref(), &initial_context) .await; handlers::thread_rollback(&sess, "sub-1".to_string(), /*num_turns*/ 0).await; @@ -7611,7 +7616,7 @@ async fn record_context_updates_and_set_reference_context_item_reinjects_full_co phase: None, }; session - .record_into_history(std::slice::from_ref(&compacted_summary), &turn_context) + .record_conversation_items(&turn_context, std::slice::from_ref(&compacted_summary)) .await; session .record_context_updates_and_set_reference_context_item(&turn_context) @@ -8042,7 +8047,7 @@ async fn guardian_helper_review_interrupts_after_three_consecutive_denials() { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[test_log::test] -async fn abort_regular_task_emits_turn_aborted_only() { +async fn abort_regular_task_emits_marker_before_turn_aborted() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput(vec![UserInput::Text { text: "hello".to_string(), @@ -8060,8 +8065,13 @@ async fn abort_regular_task_emits_turn_aborted_only() { sess.abort_all_tasks(TurnAbortReason::Interrupted).await; - // Interrupts persist a model-visible `` marker into history, but there is no - // separate client-visible event for that marker (only `EventMsg::TurnAborted`). + // Interrupts surface the model-visible `` marker before the abort event. + let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("timeout waiting for marker event") + .expect("event"); + assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_))); + let evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("timeout waiting for event") @@ -8075,7 +8085,7 @@ async fn abort_regular_task_emits_turn_aborted_only() { } #[tokio::test] -async fn abort_gracefully_emits_turn_aborted_only() { +async fn abort_gracefully_emits_marker_before_turn_aborted() { let (sess, tc, rx) = make_session_and_context_with_rx().await; let input = vec![TurnInput::UserInput(vec![UserInput::Text { text: "hello".to_string(), @@ -8093,8 +8103,13 @@ async fn abort_gracefully_emits_turn_aborted_only() { sess.abort_all_tasks(TurnAbortReason::Interrupted).await; - // Even if tasks handle cancellation gracefully, interrupts still result in `TurnAborted` - // being the only client-visible signal. + // Gracefully cancelled tasks surface the model-visible marker before the abort event too. + let marker_evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) + .await + .expect("timeout waiting for marker event") + .expect("event"); + assert!(matches!(marker_evt.msg, EventMsg::RawResponseItem(_))); + let evt = tokio::time::timeout(std::time::Duration::from_secs(2), rx.recv()) .await .expect("timeout waiting for event") @@ -8363,66 +8378,11 @@ async fn steer_input_returns_active_turn_id() { assert!(sess.input_queue.has_pending_input(&sess.active_turn).await); } -#[tokio::test] -async fn queued_response_items_for_next_turn_move_into_next_active_turn() { - let (sess, tc, _rx) = make_session_and_context_with_rx().await; - let queued_item = ResponseInputItem::Message { - role: "assistant".to_string(), - content: vec![ContentItem::InputText { - text: "queued before wake".to_string(), - }], - phase: None, - }; - - sess.input_queue - .queue_response_items_for_next_turn(vec![queued_item.clone()]) - .await; - - sess.spawn_task( - Arc::clone(&tc), - Vec::new(), - NeverEndingTask { - kind: TaskKind::Regular, - listen_to_cancellation_token: false, - }, - ) - .await; - - assert_eq!( - sess.input_queue.get_pending_input(&sess.active_turn).await, - vec![TurnInput::ResponseInputItem(queued_item)] - ); -} - -#[tokio::test] -async fn idle_interrupt_does_not_wake_queued_next_turn_items() { - let (sess, _tc, _rx) = make_session_and_context_with_rx().await; - let queued_item = ResponseInputItem::Message { - role: "assistant".to_string(), - content: vec![ContentItem::InputText { - text: "queued before interrupt".to_string(), - }], - phase: None, - }; - - sess.input_queue - .queue_response_items_for_next_turn(vec![queued_item]) - .await; - - sess.abort_all_tasks(TurnAbortReason::Interrupted).await; - - assert!(sess.active_turn.lock().await.is_none()); - assert!( - sess.input_queue - .has_queued_response_items_for_next_turn() - .await - ); -} - #[tokio::test] async fn abort_empty_active_turn_preserves_pending_input() { let (sess, _tc, _rx) = make_session_and_context_with_rx().await; - let pending_item = ResponseInputItem::Message { + let pending_item = ResponseItem::Message { + id: None, role: "user".to_string(), content: vec![ContentItem::InputText { text: "late pending input".to_string(), @@ -8437,7 +8397,7 @@ async fn abort_empty_active_turn_preserves_pending_input() { sess.input_queue .extend_pending_input_for_turn_state( turn_state.as_ref(), - vec![TurnInput::ResponseInputItem(pending_item.clone())], + vec![TurnInput::ResponseItem(pending_item.clone())], ) .await; @@ -8448,7 +8408,7 @@ async fn abort_empty_active_turn_preserves_pending_input() { sess.input_queue .take_pending_input_for_turn_state(turn_state.as_ref()) .await, - vec![TurnInput::ResponseInputItem(pending_item)] + vec![TurnInput::ResponseItem(pending_item)] ); } @@ -8874,7 +8834,7 @@ async fn budget_limited_accounting_steers_active_turn_without_aborting() -> anyh .await?; let pending_input = sess.input_queue.get_pending_input(&sess.active_turn).await; - let [TurnInput::ResponseInputItem(ResponseInputItem::Message { role, content, .. })] = + let [TurnInput::ResponseItem(ResponseItem::Message { role, content, .. })] = pending_input.as_slice() else { panic!("expected one budget-limit steering message, got {pending_input:#?}"); @@ -9101,7 +9061,7 @@ async fn external_objective_change_steers_active_turn() -> anyhow::Result<()> { pending_input.iter().any(|item| { matches!( item, - TurnInput::ResponseInputItem(ResponseInputItem::Message { role, content, .. }) + TurnInput::ResponseItem(ResponseItem::Message { role, content, .. }) if role == "user" && content.iter().any(|content| matches!( content, @@ -9325,9 +9285,9 @@ async fn queue_only_mailbox_mail_waits_for_next_turn_after_answer_boundary() { assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, - vec![TurnInput::ResponseInputItem( + vec![TurnInput::ResponseItem(ResponseItem::from( communication.to_response_input_item() - )], + ))], ); } @@ -9412,7 +9372,7 @@ async fn steered_input_reopens_mailbox_delivery_for_current_turn() { text: "follow up".to_string(), text_elements: Vec::new(), }]), - TurnInput::ResponseInputItem(communication.to_response_input_item()), + TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())), ], ); } @@ -9466,7 +9426,7 @@ async fn stale_defer_mailbox_delivery_does_not_override_steered_input() { text: "follow up".to_string(), text_elements: Vec::new(), }]), - TurnInput::ResponseInputItem(communication.to_response_input_item()), + TurnInput::ResponseItem(ResponseItem::from(communication.to_response_input_item())), ], ); } @@ -9521,9 +9481,9 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { assert!(output.tool_future.is_some()); assert_eq!( sess.input_queue.get_pending_input(&sess.active_turn).await, - vec![TurnInput::ResponseInputItem( + vec![TurnInput::ResponseItem(ResponseItem::from( communication.to_response_input_item() - )], + ))], ); } @@ -9581,8 +9541,7 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() { ); let history = sess.clone_history().await; - // The `` marker is silent in the event stream, so verify it is still - // recorded in history for the model. + // Verify the `` marker is still recorded in history for the model. assert!( history.raw_items().iter().any(|item| { let ResponseItem::Message { role, content, .. } = item else { diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index cc7592903..477e38c81 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -458,7 +458,7 @@ async fn build_skills_and_plugins( .iter() .filter_map(|item| match item { TurnInput::UserInput(content) => Some(content.as_slice()), - TurnInput::ResponseInputItem(_) => None, + TurnInput::ResponseItem(_) => None, }) .flatten() .cloned() @@ -611,7 +611,7 @@ async fn track_turn_resolved_config_analytics( .iter() .filter_map(|item| match item { TurnInput::UserInput(content) => Some(content.as_slice()), - TurnInput::ResponseInputItem(_) => None, + TurnInput::ResponseItem(_) => None, }) .flatten() .filter(|item| { diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index aa161c127..d8b568906 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -44,7 +44,6 @@ use codex_otel::TURN_TOKEN_USAGE_METRIC; use codex_otel::TURN_TOOL_CALL_METRIC; use codex_protocol::models::ResponseItem; use codex_protocol::protocol::EventMsg; -use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::TokenUsage; use codex_protocol::protocol::TurnAbortReason; use codex_protocol::protocol::TurnAbortedEvent; @@ -347,11 +346,7 @@ impl Session { { warn!("failed to apply goal runtime turn-start event: {err}"); } - let queued_response_items = self - .input_queue - .take_queued_response_items_for_next_turn() - .await; - let mailbox_items = self.input_queue.get_pending_input(&self.active_turn).await; + let pending_items = self.input_queue.get_pending_input(&self.active_turn).await; let turn_state = { let mut active = self.active_turn.lock().await; let turn = active.get_or_insert_with(ActiveTurn::default); @@ -359,11 +354,6 @@ impl Session { Arc::clone(&turn.turn_state) }; turn_state.lock().await.token_usage_at_turn_start = token_usage_at_turn_start.clone(); - let mut pending_items = queued_response_items - .into_iter() - .map(TurnInput::ResponseInputItem) - .collect::>(); - pending_items.extend(mailbox_items); self.input_queue .extend_pending_input_for_turn_state(turn_state.as_ref(), pending_items) .await; @@ -452,8 +442,7 @@ impl Session { /// Starts a regular turn when the session is idle and pending work is waiting. /// - /// Pending work currently includes queued next-turn items and mailbox mail marked with - /// `trigger_turn`. + /// Pending work currently includes mailbox mail marked with `trigger_turn`. /// /// This helper generates a fresh sub-id for the synthetic turn before delegating to the /// explicit-sub-id variant. @@ -465,18 +454,13 @@ impl Session { /// Starts a regular turn with the provided sub-id when pending work should wake an idle /// session. /// - /// The turn is created only when there are queued next-turn items or mailbox mail marked with - /// `trigger_turn`, and only if the session is currently idle. + /// The turn is created only when there is mailbox mail marked with `trigger_turn`, and only + /// if the session is currently idle. pub(crate) async fn maybe_start_turn_for_pending_work_with_sub_id( self: &Arc, sub_id: String, ) { - if !self - .input_queue - .has_queued_response_items_for_next_turn() - .await - && !self.input_queue.has_trigger_turn_mailbox_items().await - { + if !self.input_queue.has_trigger_turn_mailbox_items().await { return; } @@ -859,10 +843,11 @@ impl Session { InterruptedTurnHistoryMarker::from_config(task.turn_context.config.as_ref()), ) { - self.record_into_history(std::slice::from_ref(&marker), task.turn_context.as_ref()) - .await; - self.persist_rollout_items(&[RolloutItem::ResponseItem(marker)]) - .await; + self.record_conversation_items( + task.turn_context.as_ref(), + std::slice::from_ref(&marker), + ) + .await; // Ensure the marker is durably visible before emitting TurnAborted: some clients // synchronously re-read the rollout on receipt of the abort event. if let Err(err) = self.flush_rollout().await { diff --git a/codex-rs/core/src/tasks/review.rs b/codex-rs/core/src/tasks/review.rs index ff7f38a7e..514bb13e4 100644 --- a/codex-rs/core/src/tasks/review.rs +++ b/codex-rs/core/src/tasks/review.rs @@ -73,7 +73,7 @@ impl SessionTask for ReviewTask { for item in input { match item { TurnInput::UserInput(mut content) => user_input.append(&mut content), - TurnInput::ResponseInputItem(_) => {} + TurnInput::ResponseItem(_) => {} } } diff --git a/codex-rs/core/src/tasks/user_shell.rs b/codex-rs/core/src/tasks/user_shell.rs index 396aecbee..816a18805 100644 --- a/codex-rs/core/src/tasks/user_shell.rs +++ b/codex-rs/core/src/tasks/user_shell.rs @@ -38,8 +38,6 @@ use super::SessionTask; use super::SessionTaskContext; use crate::session::session::Session; use codex_protocol::models::PermissionProfile; -use codex_protocol::models::ResponseInputItem; -use codex_protocol::models::ResponseItem; const USER_SHELL_TIMEOUT_MS: u64 = 60 * 60 * 1000; // 1 hour @@ -361,30 +359,7 @@ async fn persist_user_shell_output( return; } - let response_input_item = match output_item { - ResponseItem::Message { - role, - content, - phase, - .. - } => ResponseInputItem::Message { - role, - content, - phase, - }, - _ => unreachable!("user shell command output record should always be a message"), - }; - - if let Err(items) = session - .inject_response_items(vec![response_input_item]) - .await - { - let response_items = items - .into_iter() - .map(ResponseItem::from) - .collect::>(); - session - .record_conversation_items(turn_context, &response_items) - .await; - } + session + .inject_no_new_turn(vec![output_item], Some(turn_context)) + .await; } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index d209026b6..3eec8315b 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -33,8 +33,6 @@ use codex_protocol::ThreadId; use codex_protocol::config_types::CollaborationModeMask; use codex_protocol::error::CodexErr; use codex_protocol::error::Result as CodexResult; -#[cfg(test)] -use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ModelPreset; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; @@ -1016,17 +1014,6 @@ impl ThreadManagerState { thread.submit(op).await } - #[cfg(test)] - /// Append a prebuilt message to a thread by ID outside the normal user-input path. - pub(crate) async fn append_message( - &self, - thread_id: ThreadId, - message: ResponseItem, - ) -> CodexResult { - let thread = self.get_thread(thread_id).await?; - thread.append_message(message).await - } - /// Remove a thread from the manager by ID, returning it when present. pub(crate) async fn remove_thread(&self, thread_id: &ThreadId) -> Option> { self.threads.write().await.remove(thread_id) diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index 09d58ea8f..abf4b9bba 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -13,7 +13,7 @@ use codex_code_mode::CodeModeTurnHost; use codex_code_mode::RuntimeResponse; use codex_protocol::models::FunctionCallOutputContentItem; use codex_protocol::models::FunctionCallOutputPayload; -use codex_protocol::models::ResponseInputItem; +use codex_protocol::models::ResponseItem; use serde_json::Value as JsonValue; use tokio_util::sync::CancellationToken; @@ -134,7 +134,7 @@ impl CodeModeTurnHost for CoreTurnHost { } self.exec .session - .inject_response_items(vec![ResponseInputItem::CustomToolCallOutput { + .inject_if_running(vec![ResponseItem::CustomToolCallOutput { call_id, name: Some(PUBLIC_TOOL_NAME.to_string()), output: FunctionCallOutputPayload::from_text(text), diff --git a/codex-rs/core/src/tools/handlers/extension_tools.rs b/codex-rs/core/src/tools/handlers/extension_tools.rs index 8c9f55c46..6e6ef019c 100644 --- a/codex-rs/core/src/tools/handlers/extension_tools.rs +++ b/codex-rs/core/src/tools/handlers/extension_tools.rs @@ -208,7 +208,7 @@ mod tests { phase: None, }; session - .record_into_history(std::slice::from_ref(&history_item), &turn) + .record_conversation_items(&turn, std::slice::from_ref(&history_item)) .await; let invocation = ToolInvocation { session: session.into(), diff --git a/codex-rs/core/src/tools/router_tests.rs b/codex-rs/core/src/tools/router_tests.rs index 71ff7ca9e..cece7fc17 100644 --- a/codex-rs/core/src/tools/router_tests.rs +++ b/codex-rs/core/src/tools/router_tests.rs @@ -332,7 +332,7 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow phase: None, }; session - .record_into_history(std::slice::from_ref(&history_item), &turn) + .record_conversation_items(&turn, std::slice::from_ref(&history_item)) .await; let router = ToolRouter::from_turn_context(