From 5cccaa5a0e0fd51babeb4c6723f4ca042a9d7e6f Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Wed, 10 Sep 2025 12:10:51 -0700 Subject: [PATCH] tests --- codex-rs/core/src/codex.rs | 52 ++++++++++++++++++++++++++------------ 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 168feef277..840fba90ec 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -543,7 +543,7 @@ impl Session { InitialHistory::New => { // Build and record initial items (user instructions + environment context) let items = self.build_initial_context(turn_context); - self.record_conversation_items(&items).await; + self.record_response_items(&items).await; } InitialHistory::Resumed(_) | InitialHistory::Forked(_) => { let rollout_items = conversation_history.get_rollout_items(); @@ -552,7 +552,13 @@ impl Session { // Always add response items to conversation history let response_items = conversation_history.get_response_items(); if !response_items.is_empty() { - self.record_into_history(&response_items); + self.record_into_history_response_items(&response_items); + } + + // Always add event msgs to conversation history + let event_msgs = conversation_history.get_event_msgs(); + if let Some(event_msgs) = event_msgs { + self.record_into_history_event_msgs(&event_msgs); } // If persisting, persist all rollout items as-is (recorder filters) @@ -566,13 +572,8 @@ impl Session { /// Persist the event to rollout and send it to clients. pub(crate) async fn send_event(&self, event: Event) { // Persist the event into event_msgs in memory - self.state - .lock_unchecked() - .event_msgs - .record_items(std::slice::from_ref(&event.msg)); - // 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; + self.record_conversation_event_msgs(std::slice::from_ref(&event.msg)) + .await; if let Err(e) = self.tx_event.send(event).await { error!("failed to send tool call event: {e}"); } @@ -662,21 +663,34 @@ impl Session { state.approved_commands.insert(cmd); } + async fn record_conversation_event_msgs(&self, items: &[EventMsg]) { + self.record_into_history_event_msgs(items); + self.persist_rollout_event_msgs(items).await; + } + /// Records input items: always append to conversation history and /// persist these response items to rollout. - async fn record_conversation_items(&self, items: &[ResponseItem]) { - self.record_into_history(items); + async fn record_response_items(&self, items: &[ResponseItem]) { + self.record_into_history_response_items(items); self.persist_rollout_response_items(items).await; } /// Append ResponseItems to the in-memory conversation history only. - fn record_into_history(&self, items: &[ResponseItem]) { + fn record_into_history_response_items(&self, items: &[ResponseItem]) { self.state .lock_unchecked() .response_items .record_items(items.iter()); } + /// Append EventMsgs to the in-memory conversation history only. + fn record_into_history_event_msgs(&self, items: &[EventMsg]) { + self.state + .lock_unchecked() + .event_msgs + .record_items(items.iter()); + } + async fn persist_rollout_response_items(&self, items: &[ResponseItem]) { let rollout_items: Vec = items .iter() @@ -686,6 +700,12 @@ impl Session { self.persist_rollout_items(&rollout_items).await; } + async fn persist_rollout_event_msgs(&self, items: &[EventMsg]) { + let rollout_items: Vec = + items.iter().cloned().map(RolloutItem::EventMsg).collect(); + self.persist_rollout_items(&rollout_items).await; + } + fn build_initial_context(&self, turn_context: &TurnContext) -> Vec { let mut items = Vec::::with_capacity(2); if let Some(user_instructions) = turn_context.user_instructions.as_deref() { @@ -717,7 +737,7 @@ impl Session { async fn record_input_and_rollout_usermsg(&self, response_input: &ResponseInputItem) { let response_item: ResponseItem = response_input.clone().into(); // Add to conversation history and persist response item to rollout - self.record_conversation_items(std::slice::from_ref(&response_item)) + self.record_response_items(std::slice::from_ref(&response_item)) .await; // Derive user message events and persist only UserMessage to rollout @@ -1172,7 +1192,7 @@ async fn submission_loop( // Install the new persistent context for subsequent tasks/turns. turn_context = Arc::new(new_turn_context); if cwd.is_some() || approval_policy.is_some() || sandbox_policy.is_some() { - sess.record_conversation_items(&[ResponseItem::from(EnvironmentContext::new( + sess.record_response_items(&[ResponseItem::from(EnvironmentContext::new( cwd, approval_policy, sandbox_policy, @@ -1460,7 +1480,7 @@ async fn run_task( .into_iter() .map(ResponseItem::from) .collect::>(); - sess.record_conversation_items(&pending_input).await; + sess.record_response_items(&pending_input).await; // Construct the input that we will send to the model. When using the // Chat completions API (or ZDR clients), the model needs the full @@ -1587,7 +1607,7 @@ async fn run_task( // Only attempt to take the lock if there is something to record. if !items_to_record_in_conversation_history.is_empty() { - sess.record_conversation_items(&items_to_record_in_conversation_history) + sess.record_response_items(&items_to_record_in_conversation_history) .await; }