From b8ec97c0ef37e4acc6b6243d893989183a478467 Mon Sep 17 00:00:00 2001 From: Celia Chen Date: Thu, 13 Nov 2025 16:25:01 -0800 Subject: [PATCH] [App-server] add new v2 events:`item/reasoning/delta`, `item/agentMessage/delta` & `item/reasoning/summaryPartAdded` (#6559) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit core event to app server event mapping: 1. `codex/event/reasoning_content_delta` -> `item/reasoning/summaryTextDelta`. 2. `codex/event/reasoning_raw_content_delta` -> `item/reasoning/textDelta` 3. `codex/event/agent_message_content_delta` → `item/agentMessage/delta`. 4. `codex/event/agent_reasoning_section_break` -> `item/reasoning/summaryPartAdded`. Also added a change in core to pass down content index, summary index and item id from events. Tested with the `git checkout owen/app_server_test_client && cargo run -p codex-app-server-test-client -- send-message-v2 "hello"` and verified that new events are emitted correctly. --- .../src/protocol/common.rs | 3 ++ .../app-server-protocol/src/protocol/v2.rs | 50 ++++++++++++++----- .../app-server/src/codex_message_processor.rs | 46 +++++++++++++++++ codex-rs/core/src/chat_completions.rs | 20 ++++++-- codex-rs/core/src/client.rs | 26 +++++++--- codex-rs/core/src/client_common.rs | 14 ++++-- codex-rs/core/src/codex.rs | 31 +++++++++--- codex-rs/core/tests/chat_completions_sse.rs | 40 +++++++++++++-- codex-rs/core/tests/common/responses.rs | 2 + codex-rs/protocol/src/protocol.rs | 14 +++++- 10 files changed, 203 insertions(+), 43 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index f754ece56..d36e81be0 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -508,6 +508,9 @@ server_notification_definitions! { McpToolCallProgress => "item/mcpToolCall/progress" (v2::McpToolCallProgressNotification), AccountUpdated => "account/updated" (v2::AccountUpdatedNotification), AccountRateLimitsUpdated => "account/rateLimits/updated" (v2::AccountRateLimitsUpdatedNotification), + ReasoningSummaryTextDelta => "item/reasoning/summaryTextDelta" (v2::ReasoningSummaryTextDeltaNotification), + ReasoningSummaryPartAdded => "item/reasoning/summaryPartAdded" (v2::ReasoningSummaryPartAddedNotification), + ReasoningTextDelta => "item/reasoning/textDelta" (v2::ReasoningTextDeltaNotification), #[serde(rename = "account/login/completed")] #[ts(rename = "account/login/completed")] diff --git a/codex-rs/app-server-protocol/src/protocol/v2.rs b/codex-rs/app-server-protocol/src/protocol/v2.rs index 10196884a..4642798c6 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2.rs @@ -516,7 +516,10 @@ pub enum ThreadItem { }, Reasoning { id: String, - text: String, + #[serde(default)] + summary: Vec, + #[serde(default)] + content: Vec, }, CommandExecution { id: String, @@ -575,17 +578,11 @@ impl From for ThreadItem { .collect::(); ThreadItem::AgentMessage { id: agent.id, text } } - CoreTurnItem::Reasoning(reasoning) => { - let text = if !reasoning.summary_text.is_empty() { - reasoning.summary_text.join("\n") - } else { - reasoning.raw_content.join("\n") - }; - ThreadItem::Reasoning { - id: reasoning.id, - text, - } - } + CoreTurnItem::Reasoning(reasoning) => ThreadItem::Reasoning { + id: reasoning.id, + summary: reasoning.summary_text, + content: reasoning.raw_content, + }, CoreTurnItem::WebSearch(search) => ThreadItem::WebSearch { id: search.id, query: search.query, @@ -719,6 +716,32 @@ pub struct AgentMessageDeltaNotification { pub delta: String, } +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ReasoningSummaryTextDeltaNotification { + pub item_id: String, + pub delta: String, + pub summary_index: i64, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ReasoningSummaryPartAddedNotification { + pub item_id: String, + pub summary_index: i64, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct ReasoningTextDeltaNotification { + pub item_id: String, + pub delta: String, + pub content_index: i64, +} + #[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] #[serde(rename_all = "camelCase")] #[ts(export_to = "v2/")] @@ -867,7 +890,8 @@ mod tests { ThreadItem::from(reasoning_item), ThreadItem::Reasoning { id: "reasoning-1".to_string(), - text: "line one\nline two".to_string(), + summary: vec!["line one".to_string(), "line two".to_string()], + content: vec![], } ); diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 23fbe59a9..b012458da 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -12,6 +12,7 @@ use codex_app_server_protocol::AccountRateLimitsUpdatedNotification; use codex_app_server_protocol::AccountUpdatedNotification; use codex_app_server_protocol::AddConversationListenerParams; use codex_app_server_protocol::AddConversationSubscriptionResponse; +use codex_app_server_protocol::AgentMessageDeltaNotification; use codex_app_server_protocol::ApplyPatchApprovalParams; use codex_app_server_protocol::ApplyPatchApprovalResponse; use codex_app_server_protocol::ArchiveConversationParams; @@ -62,6 +63,9 @@ use codex_app_server_protocol::ModelListParams; use codex_app_server_protocol::ModelListResponse; use codex_app_server_protocol::NewConversationParams; use codex_app_server_protocol::NewConversationResponse; +use codex_app_server_protocol::ReasoningSummaryPartAddedNotification; +use codex_app_server_protocol::ReasoningSummaryTextDeltaNotification; +use codex_app_server_protocol::ReasoningTextDeltaNotification; use codex_app_server_protocol::RemoveConversationListenerParams; use codex_app_server_protocol::RemoveConversationSubscriptionResponse; use codex_app_server_protocol::RequestId; @@ -2681,6 +2685,48 @@ async fn apply_bespoke_event_handling( on_patch_approval_response(event_id, rx, conversation).await; }); } + EventMsg::AgentMessageContentDelta(event) => { + let notification = AgentMessageDeltaNotification { + item_id: event.item_id, + delta: event.delta, + }; + outgoing + .send_server_notification(ServerNotification::AgentMessageDelta(notification)) + .await; + } + EventMsg::ReasoningContentDelta(event) => { + let notification = ReasoningSummaryTextDeltaNotification { + item_id: event.item_id, + delta: event.delta, + summary_index: event.summary_index, + }; + outgoing + .send_server_notification(ServerNotification::ReasoningSummaryTextDelta( + notification, + )) + .await; + } + EventMsg::ReasoningRawContentDelta(event) => { + let notification = ReasoningTextDeltaNotification { + item_id: event.item_id, + delta: event.delta, + content_index: event.content_index, + }; + outgoing + .send_server_notification(ServerNotification::ReasoningTextDelta(notification)) + .await; + } + EventMsg::AgentReasoningSectionBreak(event) => { + let notification = ReasoningSummaryPartAddedNotification { + item_id: event.item_id, + summary_index: event.summary_index, + }; + outgoing + .send_server_notification(ServerNotification::ReasoningSummaryPartAdded( + notification, + )) + .await; + } EventMsg::ExecApprovalRequest(ExecApprovalRequestEvent { call_id, command, diff --git a/codex-rs/core/src/chat_completions.rs b/codex-rs/core/src/chat_completions.rs index d8d3a63d1..5b9578e75 100644 --- a/codex-rs/core/src/chat_completions.rs +++ b/codex-rs/core/src/chat_completions.rs @@ -477,10 +477,14 @@ async fn append_reasoning_text( .. }) = reasoning_item { + let content_index = content.len() as i64; content.push(ReasoningItemContent::ReasoningText { text: text.clone() }); let _ = tx_event - .send(Ok(ResponseEvent::ReasoningContentDelta(text.clone()))) + .send(Ok(ResponseEvent::ReasoningContentDelta { + delta: text.clone(), + content_index, + })) .await; } } @@ -898,20 +902,26 @@ where continue; } } - Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))) => { + Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta { + delta, + content_index, + }))) => { // Always accumulate reasoning deltas so we can emit a final Reasoning item at Completed. this.cumulative_reasoning.push_str(&delta); if matches!(this.mode, AggregateMode::Streaming) { // In streaming mode, also forward the delta immediately. - return Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta(delta)))); + return Poll::Ready(Some(Ok(ResponseEvent::ReasoningContentDelta { + delta, + content_index, + }))); } else { continue; } } - Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta(_)))) => { + Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryDelta { .. }))) => { continue; } - Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryPartAdded))) => { + Poll::Ready(Some(Ok(ResponseEvent::ReasoningSummaryPartAdded { .. }))) => { continue; } Poll::Ready(Some(Ok(ResponseEvent::OutputItemAdded(item)))) => { diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index d1ade2182..98775e3d3 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -560,6 +560,8 @@ struct SseEvent { response: Option, item: Option, delta: Option, + summary_index: Option, + content_index: Option, } #[derive(Debug, Deserialize)] @@ -819,16 +821,22 @@ async fn process_sse( } } "response.reasoning_summary_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::ReasoningSummaryDelta(delta); + if let (Some(delta), Some(summary_index)) = (event.delta, event.summary_index) { + let event = ResponseEvent::ReasoningSummaryDelta { + delta, + summary_index, + }; if tx_event.send(Ok(event)).await.is_err() { return; } } } "response.reasoning_text.delta" => { - if let Some(delta) = event.delta { - let event = ResponseEvent::ReasoningContentDelta(delta); + if let (Some(delta), Some(content_index)) = (event.delta, event.content_index) { + let event = ResponseEvent::ReasoningContentDelta { + delta, + content_index, + }; if tx_event.send(Ok(event)).await.is_err() { return; } @@ -905,10 +913,12 @@ async fn process_sse( } } "response.reasoning_summary_part.added" => { - // Boundary between reasoning summary sections (e.g., titles). - let event = ResponseEvent::ReasoningSummaryPartAdded; - if tx_event.send(Ok(event)).await.is_err() { - return; + if let Some(summary_index) = event.summary_index { + // Boundary between reasoning summary sections (e.g., titles). + let event = ResponseEvent::ReasoningSummaryPartAdded { summary_index }; + if tx_event.send(Ok(event)).await.is_err() { + return; + } } } "response.reasoning_summary_text.done" => {} diff --git a/codex-rs/core/src/client_common.rs b/codex-rs/core/src/client_common.rs index 7d33751e2..a628e0d32 100644 --- a/codex-rs/core/src/client_common.rs +++ b/codex-rs/core/src/client_common.rs @@ -203,9 +203,17 @@ pub enum ResponseEvent { token_usage: Option, }, OutputTextDelta(String), - ReasoningSummaryDelta(String), - ReasoningContentDelta(String), - ReasoningSummaryPartAdded, + ReasoningSummaryDelta { + delta: String, + summary_index: i64, + }, + ReasoningContentDelta { + delta: String, + content_index: i64, + }, + ReasoningSummaryPartAdded { + summary_index: i64, + }, RateLimits(RateLimitSnapshot), } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 273f7a497..4422b3884 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -2232,13 +2232,17 @@ async fn try_run_turn( error_or_panic("ReasoningSummaryDelta without active item".to_string()); } } - ResponseEvent::ReasoningSummaryDelta(delta) => { + ResponseEvent::ReasoningSummaryDelta { + delta, + summary_index, + } => { if let Some(active) = active_item.as_ref() { let event = ReasoningContentDeltaEvent { thread_id: sess.conversation_id.to_string(), turn_id: turn_context.sub_id.clone(), item_id: active.id(), - delta: delta.clone(), + delta, + summary_index, }; sess.send_event(&turn_context, EventMsg::ReasoningContentDelta(event)) .await; @@ -2246,18 +2250,29 @@ async fn try_run_turn( error_or_panic("ReasoningSummaryDelta without active item".to_string()); } } - ResponseEvent::ReasoningSummaryPartAdded => { - let event = - EventMsg::AgentReasoningSectionBreak(AgentReasoningSectionBreakEvent {}); - sess.send_event(&turn_context, event).await; + ResponseEvent::ReasoningSummaryPartAdded { summary_index } => { + if let Some(active) = active_item.as_ref() { + let event = + EventMsg::AgentReasoningSectionBreak(AgentReasoningSectionBreakEvent { + item_id: active.id(), + summary_index, + }); + sess.send_event(&turn_context, event).await; + } else { + error_or_panic("ReasoningSummaryPartAdded without active item".to_string()); + } } - ResponseEvent::ReasoningContentDelta(delta) => { + ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => { if let Some(active) = active_item.as_ref() { let event = ReasoningRawContentDeltaEvent { thread_id: sess.conversation_id.to_string(), turn_id: turn_context.sub_id.clone(), item_id: active.id(), - delta: delta.clone(), + delta, + content_index, }; sess.send_event(&turn_context, EventMsg::ReasoningRawContentDelta(event)) .await; diff --git a/codex-rs/core/tests/chat_completions_sse.rs b/codex-rs/core/tests/chat_completions_sse.rs index 46378b082..b499b76e8 100644 --- a/codex-rs/core/tests/chat_completions_sse.rs +++ b/codex-rs/core/tests/chat_completions_sse.rs @@ -215,7 +215,13 @@ async fn streams_reasoning_from_string_delta() { } match &events[1] { - ResponseEvent::ReasoningContentDelta(text) => assert_eq!(text, "think1"), + ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => { + assert_eq!(delta, "think1"); + assert_eq!(content_index, &0); + } other => panic!("expected reasoning delta, got {other:?}"), } @@ -267,12 +273,24 @@ async fn streams_reasoning_from_object_delta() { } match &events[1] { - ResponseEvent::ReasoningContentDelta(text) => assert_eq!(text, "partA"), + ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => { + assert_eq!(delta, "partA"); + assert_eq!(content_index, &0); + } other => panic!("expected reasoning delta, got {other:?}"), } match &events[2] { - ResponseEvent::ReasoningContentDelta(text) => assert_eq!(text, "partB"), + ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => { + assert_eq!(delta, "partB"); + assert_eq!(content_index, &1); + } other => panic!("expected reasoning delta, got {other:?}"), } @@ -319,7 +337,13 @@ async fn streams_reasoning_from_final_message() { } match &events[1] { - ResponseEvent::ReasoningContentDelta(text) => assert_eq!(text, "final-cot"), + ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => { + assert_eq!(delta, "final-cot"); + assert_eq!(content_index, &0); + } other => panic!("expected reasoning delta, got {other:?}"), } @@ -354,7 +378,13 @@ async fn streams_reasoning_before_tool_call() { } match &events[1] { - ResponseEvent::ReasoningContentDelta(text) => assert_eq!(text, "pre-tool"), + ResponseEvent::ReasoningContentDelta { + delta, + content_index, + } => { + assert_eq!(delta, "pre-tool"); + assert_eq!(content_index, &0); + } other => panic!("expected reasoning delta, got {other:?}"), } diff --git a/codex-rs/core/tests/common/responses.rs b/codex-rs/core/tests/common/responses.rs index 703461f94..b78f921bc 100644 --- a/codex-rs/core/tests/common/responses.rs +++ b/codex-rs/core/tests/common/responses.rs @@ -296,6 +296,7 @@ pub fn ev_reasoning_summary_text_delta(delta: &str) -> Value { serde_json::json!({ "type": "response.reasoning_summary_text.delta", "delta": delta, + "summary_index": 0, }) } @@ -303,6 +304,7 @@ pub fn ev_reasoning_text_delta(delta: &str) -> Value { serde_json::json!({ "type": "response.reasoning_text.delta", "delta": delta, + "content_index": 0, }) } diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 2d0b0f013..2b9b43d79 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -617,6 +617,9 @@ pub struct ReasoningContentDeltaEvent { pub turn_id: String, pub item_id: String, pub delta: String, + // load with default value so it's backward compatible with the old format. + #[serde(default)] + pub summary_index: i64, } impl HasLegacyEvent for ReasoningContentDeltaEvent { @@ -633,6 +636,9 @@ pub struct ReasoningRawContentDeltaEvent { pub turn_id: String, pub item_id: String, pub delta: String, + // load with default value so it's backward compatible with the old format. + #[serde(default)] + pub content_index: i64, } impl HasLegacyEvent for ReasoningRawContentDeltaEvent { @@ -923,7 +929,13 @@ pub struct AgentReasoningRawContentDeltaEvent { } #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] -pub struct AgentReasoningSectionBreakEvent {} +pub struct AgentReasoningSectionBreakEvent { + // load with default value so it's backward compatible with the old format. + #[serde(default)] + pub item_id: String, + #[serde(default)] + pub summary_index: i64, +} #[derive(Debug, Clone, Deserialize, Serialize, JsonSchema, TS)] pub struct AgentReasoningDeltaEvent {