[codex-core] Preserve raw inline compaction event order [ci changed_files]

Co-authored-by: Codex <noreply@openai.com>

codex-rs/core/src/codex.rs

@@ -3271,8 +3271,6 @@ impl Session {
             1,
             &[("mode", "server_side")],
         );
-        self.send_raw_response_items(turn_context, std::slice::from_ref(&item))
-            .await;
         self.recompute_token_usage(turn_context).await;
     }
 

codex-rs/core/src/stream_events_utils.rs

@@ -25,6 +25,8 @@ use codex_protocol::models::FunctionCallOutputBody;
 use codex_protocol::models::FunctionCallOutputPayload;
 use codex_protocol::models::ResponseInputItem;
 use codex_protocol::models::ResponseItem;
+use codex_protocol::protocol::EventMsg;
+use codex_protocol::protocol::RawResponseItemEvent;
 use codex_utils_stream_parser::strip_proposed_plan_blocks;
 use futures::Future;
 use tracing::debug;
@@ -185,6 +187,12 @@ pub(crate) async fn handle_output_item_done(
         ctx.sess
             .emit_turn_item_completed(&ctx.turn_context, turn_item)
             .await;
+        ctx.sess
+            .send_event(
+                &ctx.turn_context,
+                EventMsg::RawResponseItem(RawResponseItemEvent { item: item.clone() }),
+            )
+            .await;
         output.pending_server_side_compaction = Some(PendingServerSideCompactionCheckpoint {
             history_at_checkpoint: ctx.sess.clone_history().await.raw_items().to_vec(),
             item,

codex-rs/core/tests/suite/compact_remote.rs

@@ -636,6 +636,100 @@ async fn auto_server_side_compaction_emits_events_before_later_streamed_items()
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn auto_server_side_compaction_preserves_raw_response_item_order() -> Result<()> {
+    skip_if_no_network!(Ok(()));
+
+    let compact_threshold = 120;
+    let inline_summary = summary_with_prefix("INLINE_SERVER_SUMMARY");
+
+    let harness = TestCodexHarness::with_builder(
+        test_codex()
+            .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
+            .with_config(move |config| {
+                config
+                    .features
+                    .enable(Feature::ServerSideCompaction)
+                    .expect("enable server-side compaction");
+                config.model_auto_compact_token_limit = Some(compact_threshold);
+            }),
+    )
+    .await?;
+    let codex = harness.test().codex.clone();
+
+    responses::mount_sse_sequence(
+        harness.server(),
+        vec![
+            responses::sse(vec![
+                responses::ev_assistant_message("m1", "FIRST_REMOTE_REPLY"),
+                responses::ev_completed_with_tokens("resp-1", 500),
+            ]),
+            responses::sse(vec![
+                responses::ev_compaction(&inline_summary),
+                responses::ev_assistant_message("m2", "AFTER_INLINE_REPLY"),
+                responses::ev_completed("resp-2"),
+            ]),
+        ],
+    )
+    .await;
+
+    submit_text_turn_and_wait(&codex, "inline compact turn one").await?;
+
+    codex
+        .submit(Op::UserInput {
+            items: vec![UserInput::Text {
+                text: "inline compact turn two".to_string(),
+                text_elements: Vec::new(),
+            }],
+            final_output_json_schema: None,
+        })
+        .await?;
+
+    let mut event_index = 0usize;
+    let mut raw_compaction_index = None;
+    let mut raw_assistant_index = None;
+    let mut saw_turn_complete = false;
+    while !saw_turn_complete || raw_compaction_index.is_none() || raw_assistant_index.is_none() {
+        let event = codex.next_event().await.expect("event");
+        match event.msg {
+            EventMsg::RawResponseItem(raw)
+                if matches!(raw.item, ResponseItem::Compaction { .. }) =>
+            {
+                raw_compaction_index = Some(event_index);
+            }
+            EventMsg::RawResponseItem(raw)
+                if matches!(
+                    &raw.item,
+                    ResponseItem::Message { role, content, .. }
+                        if role == "assistant"
+                            && content.iter().any(|entry| {
+                                matches!(
+                                    entry,
+                                    ContentItem::OutputText { text }
+                                        if text == "AFTER_INLINE_REPLY"
+                                )
+                            })
+                ) =>
+            {
+                raw_assistant_index = Some(event_index);
+            }
+            EventMsg::TurnComplete(_) => {
+                saw_turn_complete = true;
+            }
+            _ => {}
+        }
+        event_index += 1;
+    }
+
+    assert!(
+        raw_compaction_index.expect("raw compaction event")
+            < raw_assistant_index.expect("raw assistant event"),
+        "expected raw response item notifications to preserve wire order"
+    );
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn auto_server_side_compaction_keeps_current_turn_inputs_for_follow_ups() -> Result<()> {
     skip_if_no_network!(Ok(()));