diff --git a/codex-rs/core/src/chat_completions.rs b/codex-rs/core/src/chat_completions.rs index 5ede774b1c..5b4c1bfd3b 100644 --- a/codex-rs/core/src/chat_completions.rs +++ b/codex-rs/core/src/chat_completions.rs @@ -207,6 +207,7 @@ async fn process_chat_sse( } let mut fn_call_state = FunctionCallState::default(); + let mut assistant_text = String::new(); loop { let sse = match timeout(idle_timeout, stream.next()).await { @@ -254,21 +255,40 @@ async fn process_chat_sse( let choice_opt = chunk.get("choices").and_then(|c| c.get(0)); if let Some(choice) = choice_opt { - // Handle assistant content tokens. + // Handle assistant content tokens as streaming deltas. if let Some(content) = choice .get("delta") .and_then(|d| d.get("content")) .and_then(|c| c.as_str()) { - let item = ResponseItem::Message { - role: "assistant".to_string(), - content: vec![ContentItem::OutputText { - text: content.to_string(), - }], - id: None, - }; + assistant_text.push_str(content); + let _ = tx_event + .send(Ok(ResponseEvent::OutputTextDelta(content.to_string()))) + .await; + } - let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + // Forward any reasoning/thinking deltas if present. + if let Some(reasoning) = choice + .get("delta") + .and_then(|d| d.get("reasoning")) + .and_then(|c| c.as_str()) + { + let _ = tx_event + .send(Ok(ResponseEvent::ReasoningSummaryDelta( + reasoning.to_string(), + ))) + .await; + } + if let Some(reasoning_content) = choice + .get("delta") + .and_then(|d| d.get("reasoning_content")) + .and_then(|c| c.as_str()) + { + let _ = tx_event + .send(Ok(ResponseEvent::ReasoningSummaryDelta( + reasoning_content.to_string(), + ))) + .await; } // Handle streaming function / tool calls. @@ -317,7 +337,18 @@ async fn process_chat_sse( let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; } "stop" => { - // Regular turn without tool-call. + // Regular turn without tool-call. Emit the final assistant message + // as a single OutputItemDone so non-delta consumers see the result. + if !assistant_text.is_empty() { + let item = ResponseItem::Message { + role: "assistant".to_string(), + content: vec![ContentItem::OutputText { + text: std::mem::take(&mut assistant_text), + }], + id: None, + }; + let _ = tx_event.send(Ok(ResponseEvent::OutputItemDone(item))).await; + } } _ => {} } @@ -388,16 +419,21 @@ where let is_assistant_delta = matches!(&item, crate::models::ResponseItem::Message { role, .. } if role == "assistant"); if is_assistant_delta { - if let crate::models::ResponseItem::Message { content, .. } = &item { - if let Some(text) = content.iter().find_map(|c| match c { - crate::models::ContentItem::OutputText { text } => Some(text), - _ => None, - }) { - this.cumulative.push_str(text); + // Only use the final assistant message if we have not + // seen any deltas; otherwise, deltas already built the + // cumulative text and this would duplicate it. + if this.cumulative.is_empty() { + if let crate::models::ResponseItem::Message { content, .. } = &item { + if let Some(text) = content.iter().find_map(|c| match c { + crate::models::ContentItem::OutputText { text } => Some(text), + _ => None, + }) { + this.cumulative.push_str(text); + } } } - // Swallow partial assistant chunk; keep polling. + // Swallow assistant message here; emit on Completed. continue; } @@ -439,10 +475,14 @@ where // will never appear in a Chat Completions stream. continue; } - Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(_)))) - | Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => { - // Deltas are ignored here since aggregation waits for the - // final OutputItemDone. + Poll::Ready(Some(Ok(ResponseEvent::OutputTextDelta(delta)))) => { + // Accumulate incremental assistant text deltas. + this.cumulative.push_str(&delta); + continue; + } + Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => { + // Reasoning deltas are intentionally ignored by this + // adapter; they are handled by the UI directly. continue; } } diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 0f7f51c2fc..ba8960026c 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -431,7 +431,7 @@ async fn process_sse( } } } - "response.reasoning_summary_text.delta" => { + "response.reasoning_summary_text.delta" | "response.reasoning_text.delta" => { if let Some(delta) = event.delta { let event = ResponseEvent::ReasoningSummaryDelta(delta); if tx_event.send(Ok(event)).await.is_err() {