From 3e071c460daf2016bb6acbdc65e82ec4c8baa39b Mon Sep 17 00:00:00 2001 From: jif-oai Date: Tue, 6 Jan 2026 13:46:04 +0000 Subject: [PATCH] NITs --- codex-rs/core/src/agent/control.rs | 4 +- codex-rs/core/src/codex.rs | 10 ++-- codex-rs/core/src/conversation_manager.rs | 66 +++++++++++++++++++---- 3 files changed, 64 insertions(+), 16 deletions(-) diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 4a2619c9bf..181051b46c 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -94,8 +94,8 @@ impl AgentControl { /// When an agent is spawned "headless" (no UI/view attached), there may be no consumer polling /// `CodexConversation::next_event()`. The underlying event channel is unbounded, so the producer can -/// accumulate events indefinitely. This drain task prevents that memory growth while still allowing -/// `AgentBus` to be updated from the received events. +/// accumulate events indefinitely. This drain task prevents that memory growth by polling and +/// discarding events until shutdown. fn spawn_headless_drain( conversation: Arc, conversation_id: ConversationId, diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 222f60d5f3..2d73b47f5a 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -982,11 +982,6 @@ impl Session { /// Persist the event to rollout and send it to clients. pub(crate) async fn send_event(&self, turn_context: &TurnContext, msg: EventMsg) { - self.services - .agent_control - .bus - .on_event(self.conversation_id, &msg) - .await; let legacy_source = msg.clone(); let event = Event { id: turn_context.sub_id.clone(), @@ -1005,6 +1000,11 @@ impl Session { } pub(crate) async fn send_event_raw(&self, event: Event) { + self.services + .agent_control + .bus + .on_event(self.conversation_id, &event.msg) + .await; // Persist the event into rollout (recorder filters as needed) let rollout_items = vec![RolloutItem::EventMsg(event.msg.clone())]; self.persist_rollout_items(&rollout_items).await; diff --git a/codex-rs/core/src/conversation_manager.rs b/codex-rs/core/src/conversation_manager.rs index 8c56a136d8..1092a488b4 100644 --- a/codex-rs/core/src/conversation_manager.rs +++ b/codex-rs/core/src/conversation_manager.rs @@ -140,11 +140,6 @@ impl ConversationManager { self.state.get_conversation(conversation_id).await } - /// Submit an `Op` to an existing conversation. - pub async fn send_op(&self, conversation_id: ConversationId, op: Op) -> CodexResult { - self.state.send_op(conversation_id, op).await - } - pub async fn new_conversation(&self, config: Config) -> CodexResult { self.state .spawn_conversation( @@ -351,10 +346,12 @@ fn truncate_before_nth_user_message(history: InitialHistory, n: usize) -> Initia #[cfg(test)] mod tests { use super::*; + use crate::codex::make_session_and_context; use assert_matches::assert_matches; use codex_protocol::models::ContentItem; use codex_protocol::models::ReasoningItemReasoningSummary; use codex_protocol::models::ResponseItem; + use pretty_assertions::assert_eq; fn user_msg(text: &str) -> ResponseItem { ResponseItem::Message { @@ -397,11 +394,62 @@ mod tests { name: "tool".to_string(), arguments: "{}".to_string(), }, + assistant_msg("a4"), ]; - let history = - InitialHistory::Forked(items.into_iter().map(RolloutItem::ResponseItem).collect()); - let forked = truncate_before_nth_user_message(history, 1); - assert_matches!(forked, InitialHistory::Forked(rolled) if rolled.len() == 3); + let initial: Vec = items + .iter() + .cloned() + .map(RolloutItem::ResponseItem) + .collect(); + let truncated = truncate_before_nth_user_message(InitialHistory::Forked(initial), 1); + let got_items = truncated.get_rollout_items(); + let expected_items = vec![ + RolloutItem::ResponseItem(items[0].clone()), + RolloutItem::ResponseItem(items[1].clone()), + RolloutItem::ResponseItem(items[2].clone()), + ]; + assert_eq!( + serde_json::to_value(&got_items).unwrap(), + serde_json::to_value(&expected_items).unwrap() + ); + + let initial2: Vec = items + .iter() + .cloned() + .map(RolloutItem::ResponseItem) + .collect(); + let truncated2 = truncate_before_nth_user_message(InitialHistory::Forked(initial2), 2); + assert_matches!(truncated2, InitialHistory::New); + } + + #[tokio::test] + async fn ignores_session_prefix_messages_when_truncating() { + let (session, turn_context) = make_session_and_context().await; + let mut items = session.build_initial_context(&turn_context); + items.push(user_msg("feature request")); + items.push(assistant_msg("ack")); + items.push(user_msg("second question")); + items.push(assistant_msg("answer")); + + let rollout_items: Vec = items + .iter() + .cloned() + .map(RolloutItem::ResponseItem) + .collect(); + + let truncated = truncate_before_nth_user_message(InitialHistory::Forked(rollout_items), 1); + let got_items = truncated.get_rollout_items(); + + let expected: Vec = vec![ + RolloutItem::ResponseItem(items[0].clone()), + RolloutItem::ResponseItem(items[1].clone()), + RolloutItem::ResponseItem(items[2].clone()), + ]; + + assert_eq!( + serde_json::to_value(&got_items).unwrap(), + serde_json::to_value(&expected).unwrap() + ); } }