From ae4158f97993eafb1d67fce0d292decc54e0994a Mon Sep 17 00:00:00 2001 From: Dylan Hurd Date: Sun, 8 Mar 2026 16:16:15 -0700 Subject: [PATCH] reproduce issue --- codex-rs/core/tests/suite/items.rs | 98 ++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/codex-rs/core/tests/suite/items.rs b/codex-rs/core/tests/suite/items.rs index 01136a84ab..da6614c56e 100644 --- a/codex-rs/core/tests/suite/items.rs +++ b/codex-rs/core/tests/suite/items.rs @@ -432,6 +432,104 @@ async fn agent_message_content_delta_has_item_metadata() -> anyhow::Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn interleaved_reasoning_and_assistant_streams_keep_item_ids_aligned() -> anyhow::Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + + let TestCodex { codex, .. } = test_codex().build(&server).await?; + + let stream = sse(vec![ + ev_response_created("resp-1"), + ev_reasoning_item_added("reasoning-1", &[""]), + ev_message_item_added("msg-1", ""), + ev_reasoning_summary_text_delta("thinking"), + ev_reasoning_item("reasoning-1", &["thinking"], &[]), + ev_output_text_delta("answer"), + ev_assistant_message("msg-1", "answer"), + ev_completed("resp-1"), + ]); + mount_sse_once(&server, stream).await; + + codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "please reason and answer".into(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + }) + .await?; + + let reasoning_started = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemStarted(ItemStartedEvent { + item: TurnItem::Reasoning(item), + .. + }) => Some(item.clone()), + _ => None, + }) + .await; + let agent_started = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemStarted(ItemStartedEvent { + item: TurnItem::AgentMessage(item), + .. + }) => Some(item.clone()), + _ => None, + }) + .await; + let reasoning_delta = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ReasoningContentDelta(event) => Some(event.clone()), + _ => None, + }) + .await; + let reasoning_completed = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemCompleted(ItemCompletedEvent { + item: TurnItem::Reasoning(item), + .. + }) => Some(item.clone()), + _ => None, + }) + .await; + let agent_delta = wait_for_event_match(&codex, |ev| match ev { + EventMsg::AgentMessageContentDelta(event) => Some(event.clone()), + _ => None, + }) + .await; + let agent_completed = wait_for_event_match(&codex, |ev| match ev { + EventMsg::ItemCompleted(ItemCompletedEvent { + item: TurnItem::AgentMessage(item), + .. + }) => Some(item.clone()), + _ => None, + }) + .await; + + assert_eq!(reasoning_started.id, "reasoning-1"); + assert_eq!(reasoning_delta.item_id, reasoning_started.id); + assert_eq!(reasoning_delta.delta, "thinking"); + assert_eq!(reasoning_completed.id, reasoning_started.id); + assert_eq!( + reasoning_completed.summary_text, + reasoning_started.summary_text + ); + assert_eq!( + reasoning_completed.raw_content, + reasoning_started.raw_content + ); + + assert_eq!(agent_started.id, "msg-1"); + assert_eq!(agent_delta.item_id, agent_started.id); + assert_eq!(agent_delta.delta, "answer"); + assert_eq!(agent_completed.id, agent_started.id); + let Some(AgentMessageContent::Text { text }) = agent_completed.content.first() else { + panic!("expected completed agent message text content"); + }; + assert_eq!(text, "answer"); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn plan_mode_emits_plan_item_from_proposed_plan_block() -> anyhow::Result<()> { skip_if_no_network!(Ok(()));