diff --git a/crates/agentic-server-core/src/events/normalize.rs b/crates/agentic-server-core/src/events/normalize.rs index c83da97d..f74f6334 100644 --- a/crates/agentic-server-core/src/events/normalize.rs +++ b/crates/agentic-server-core/src/events/normalize.rs @@ -51,8 +51,10 @@ fn extract_payload(event_type: SSEEventType, json: &Value) -> EventPayload { SSEEventType::CustomToolCallInputDelta => extract_custom_tool_call_input_delta(json), SSEEventType::CustomToolCallInputDone => extract_custom_tool_call_input_done(json), - SSEEventType::ReasoningTextDelta | SSEEventType::ReasoningSummaryTextDelta => extract_reasoning_delta(json), - SSEEventType::ReasoningTextDone | SSEEventType::ReasoningSummaryTextDone => extract_reasoning_done(json), + SSEEventType::ReasoningTextDelta => extract_reasoning_text_delta(json), + SSEEventType::ReasoningTextDone => extract_reasoning_text_done(json), + SSEEventType::ReasoningSummaryTextDelta => extract_reasoning_summary_text_delta(json), + SSEEventType::ReasoningSummaryTextDone => extract_reasoning_summary_text_done(json), SSEEventType::ContentPartAdded | SSEEventType::ContentPartDone @@ -173,16 +175,38 @@ fn extract_custom_tool_call_input_done(json: &Value) -> EventPayload { } } -fn extract_reasoning_delta(json: &Value) -> EventPayload { - EventPayload::ReasoningDelta { +fn extract_reasoning_text_delta(json: &Value) -> EventPayload { + EventPayload::ReasoningTextDelta { delta: json_str(json, "delta"), item_id: json_str(json, "item_id"), + output_index: json_u32(json, "output_index"), + content_index: json_u32(json, "content_index"), + } +} + +fn extract_reasoning_text_done(json: &Value) -> EventPayload { + EventPayload::ReasoningTextDone { + text: json_str(json, "text"), + item_id: json_str(json, "item_id"), + output_index: json_u32(json, "output_index"), + content_index: json_u32(json, "content_index"), + } +} + +fn extract_reasoning_summary_text_delta(json: &Value) -> EventPayload { + EventPayload::ReasoningSummaryTextDelta { + delta: json_str(json, "delta"), + item_id: json_str(json, "item_id"), + output_index: json_u32(json, "output_index"), + summary_index: json_u32(json, "summary_index"), } } -fn extract_reasoning_done(json: &Value) -> EventPayload { - EventPayload::ReasoningDone { +fn extract_reasoning_summary_text_done(json: &Value) -> EventPayload { + EventPayload::ReasoningSummaryTextDone { text: json_str(json, "text"), item_id: json_str(json, "item_id"), + output_index: json_u32(json, "output_index"), + summary_index: json_u32(json, "summary_index"), } } diff --git a/crates/agentic-server-core/src/events/types.rs b/crates/agentic-server-core/src/events/types.rs index ae012736..5dfb50c7 100644 --- a/crates/agentic-server-core/src/events/types.rs +++ b/crates/agentic-server-core/src/events/types.rs @@ -308,11 +308,37 @@ pub enum EventPayload { output_index: u32, }, + /// `response.reasoning_text.delta` + ReasoningTextDelta { + delta: String, + item_id: String, + output_index: u32, + content_index: u32, + }, + + /// `response.reasoning_text.done` + ReasoningTextDone { + text: String, + item_id: String, + output_index: u32, + content_index: u32, + }, + /// `response.reasoning_summary_text.delta` - ReasoningDelta { delta: String, item_id: String }, + ReasoningSummaryTextDelta { + delta: String, + item_id: String, + output_index: u32, + summary_index: u32, + }, /// `response.reasoning_summary_text.done` - ReasoningDone { text: String, item_id: String }, + ReasoningSummaryTextDone { + text: String, + item_id: String, + output_index: u32, + summary_index: u32, + }, /// Events we classify but don't deeply parse yet. Raw(Value), diff --git a/crates/agentic-server-core/src/executor/accumulator.rs b/crates/agentic-server-core/src/executor/accumulator.rs index 2c68b542..4c3d4289 100644 --- a/crates/agentic-server-core/src/executor/accumulator.rs +++ b/crates/agentic-server-core/src/executor/accumulator.rs @@ -21,7 +21,7 @@ use crate::types::event::{MessageStatus, ResponseStatus}; use crate::types::io::output::McpListTools; use crate::types::io::{ ApplyDone, CompactionItem, CustomToolCall, FunctionToolCall, OutputItem, OutputMessage, OutputTextContent, - ReasoningOutput, ReasoningTextContent, ResponseUsage, + ReasoningOutput, ResponseUsage, }; use crate::types::io::{McpCall, WebSearchCall}; use crate::types::request_response::{IncompleteDetails, ResponsePayload}; @@ -32,7 +32,7 @@ use crate::utils::uuid7_str; /// accumulated text/arguments buffer. enum InFlight { Message { item: OutputMessage, text: String }, - Reasoning { item: ReasoningOutput, text: String }, + Reasoning { item: ReasoningOutput }, FunctionCall { item: FunctionToolCall, arguments: String }, CustomToolCall { item: CustomToolCall, input: String }, WebSearchCall { item: Option }, @@ -59,12 +59,7 @@ impl std::fmt::Debug for InFlight { impl InFlight { fn finalize(self) -> Option { match self { - Self::Reasoning { mut item, text } => { - if !text.is_empty() { - item.content.push(ReasoningTextContent::new(text)); - } - Some(OutputItem::Reasoning(item)) - } + Self::Reasoning { item } => Some(OutputItem::Reasoning(item)), Self::FunctionCall { mut item, arguments } => { if !arguments.is_empty() && item.arguments.is_empty() { item.arguments = arguments; @@ -367,18 +362,20 @@ impl ResponseAccumulator { (SSEEventType::OutputItemDone, payload @ EventPayload::OutputItemDone { .. }) => { self.complete_call_item(payload); } - (SSEEventType::ReasoningTextDelta, EventPayload::ReasoningDelta { delta, item_id }) => { - if let Some(InFlight::Reasoning { text, .. }) = - self.in_flight.get_mut(item_id).map(|entry| &mut entry.item) - { - text.push_str(delta); - } - } - (SSEEventType::ReasoningTextDone, EventPayload::ReasoningDone { item_id, .. }) => { - if let Some(InFlight::Reasoning { item, text }) = - self.in_flight.get_mut(item_id).map(|entry| &mut entry.item) - { - item.apply_done(&frame.payload, text); + ( + SSEEventType::ReasoningTextDone, + payload @ EventPayload::ReasoningTextDone { + item_id, output_index, .. + }, + ) + | ( + SSEEventType::ReasoningSummaryTextDone, + payload @ EventPayload::ReasoningSummaryTextDone { + item_id, output_index, .. + }, + ) => { + if let Some(item) = self.in_flight_reasoning_mut(item_id, *output_index) { + item.apply_done(payload, &mut String::new()); } } ( @@ -459,10 +456,9 @@ impl ResponseAccumulator { return; }; let item = match item_type { - SSEItemType::Reasoning => ReasoningOutput::try_from(payload).ok().map(|item| InFlight::Reasoning { - item, - text: String::with_capacity(256), - }), + SSEItemType::Reasoning => ReasoningOutput::try_from(payload) + .ok() + .map(|item| InFlight::Reasoning { item }), SSEItemType::FunctionCall => FunctionToolCall::try_from(payload) .ok() .map(|item| InFlight::FunctionCall { @@ -530,10 +526,19 @@ impl ResponseAccumulator { else { return; }; - let in_flight_key = self.in_flight_call_key(item_id, *item_type, *output_index); - let done_item = deserialize_from_value_opt::(raw_item.clone()); + let in_flight_key = if *item_type == SSEItemType::Reasoning { + self.in_flight_reasoning_key(item_id, *output_index) + } else { + self.in_flight_call_key(item_id, *item_type, *output_index) + }; + let done_item = if *item_type == SSEItemType::Reasoning { + ReasoningOutput::try_from(payload).ok().map(OutputItem::Reasoning) + } else { + deserialize_from_value_opt::(raw_item.clone()) + }; if let Some(entry) = in_flight_key.as_deref().and_then(|key| self.in_flight.get_mut(key)) { match (&mut entry.item, done_item) { + (InFlight::Reasoning { item }, _) => item.apply_done(payload, &mut String::new()), (InFlight::FunctionCall { item, arguments }, _) => item.apply_done(payload, arguments), (InFlight::CustomToolCall { item, input }, _) => item.apply_done(payload, input), (InFlight::McpCall { item }, _) => item.apply_done(payload, &mut String::new()), @@ -554,7 +559,8 @@ impl ResponseAccumulator { } if let Some( - mut output_item @ (OutputItem::FunctionCall(_) + mut output_item @ (OutputItem::Reasoning(_) + | OutputItem::FunctionCall(_) | OutputItem::CustomToolCall(_) | OutputItem::WebSearchCall(_) | OutputItem::McpCall(_) @@ -573,6 +579,27 @@ impl ResponseAccumulator { } } + fn in_flight_reasoning_mut(&mut self, item_id: &str, output_index: u32) -> Option<&mut ReasoningOutput> { + let key = self.in_flight_reasoning_key(item_id, output_index)?; + let InFlight::Reasoning { item } = &mut self.in_flight.get_mut(&key)?.item else { + return None; + }; + Some(item) + } + + fn in_flight_reasoning_key(&self, item_id: &str, output_index: u32) -> Option { + self.in_flight + .get(item_id) + .filter(|entry| entry.output_index == output_index && matches!(entry.item, InFlight::Reasoning { .. })) + .map(|_| item_id.to_owned()) + .or_else(|| { + self.in_flight.iter().find_map(|(key, entry)| { + (entry.output_index == output_index && matches!(entry.item, InFlight::Reasoning { .. })) + .then(|| key.clone()) + }) + }) + } + fn in_flight_call_key(&self, item_id: &str, item_type: SSEItemType, output_index: u32) -> Option { self.in_flight .get(item_id) @@ -1216,6 +1243,179 @@ mod tests { } } + #[test] + fn completed_reasoning_replaces_partial_deltas_without_duplication() { + let lines = [ + r#"data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[],"summary":[]}}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"partial content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"partial summary"}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[{"type":"reasoning_text","text":"complete content"},{"type":"reasoning_text","text":"second content"}],"summary":[{"type":"summary_text","text":"complete summary"}],"encrypted_content":"opaque-state","status":"completed"}}"#.to_owned(), + r#"data: {"type":"response.completed","response":{"id":"resp_1","status":"completed"}}"#.to_owned(), + ]; + + let acc = ResponseAccumulator::from_sse_lines(lines, None); + + assert_eq!(acc.output.len(), 1); + assert_eq!( + serde_json::to_value(&acc.output[0]).unwrap(), + serde_json::json!({ + "type": "reasoning", + "id": "rs_1", + "content": [ + {"type": "reasoning_text", "text": "complete content"}, + {"type": "reasoning_text", "text": "second content"}, + ], + "summary": [{"type": "summary_text", "text": "complete summary"}], + "encrypted_content": "opaque-state", + "status": "completed", + }) + ); + } + + #[test] + fn reasoning_done_events_keep_part_index_order() { + let lines = [ + r#"data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_1","type":"reasoning"}}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.done","item_id":"rs_1","output_index":0,"content_index":1,"text":"second content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.done","item_id":"rs_1","output_index":0,"content_index":0,"text":"first content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.done","item_id":"rs_1","output_index":0,"summary_index":1,"text":"second summary"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.done","item_id":"rs_1","output_index":0,"summary_index":0,"text":"first summary"}"#.to_owned(), + r#"data: {"type":"response.completed","response":{"id":"resp_1","status":"completed"}}"#.to_owned(), + ]; + + let acc = ResponseAccumulator::from_sse_lines(lines, None); + let OutputItem::Reasoning(reasoning) = &acc.output[0] else { + panic!("expected reasoning output"); + }; + + assert_eq!( + reasoning + .content + .iter() + .map(|part| part.text.as_str()) + .collect::>(), + ["first content", "second content"] + ); + assert_eq!( + reasoning.summary, + [ + serde_json::json!({"type": "summary_text", "text": "first summary"}), + serde_json::json!({"type": "summary_text", "text": "second summary"}), + ] + ); + } + + #[test] + fn completed_reasoning_preserves_done_fields_when_omitted() { + let lines = [ + r#"data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_1","type":"reasoning"}}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.done","item_id":"rs_1","output_index":0,"content_index":0,"text":"completed content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.done","item_id":"rs_1","output_index":0,"summary_index":0,"text":"completed summary"}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","encrypted_content":{"token":"opaque"},"status":"completed"}}"#.to_owned(), + r#"data: {"type":"response.completed","response":{"id":"resp_1","status":"completed"}}"#.to_owned(), + ]; + + let acc = ResponseAccumulator::from_sse_lines(lines, None); + let OutputItem::Reasoning(reasoning) = &acc.output[0] else { + panic!("expected reasoning output"); + }; + + assert_eq!(reasoning.content[0].text, "completed content"); + assert_eq!( + reasoning.summary, + [serde_json::json!({"type": "summary_text", "text": "completed summary"})] + ); + assert_eq!( + reasoning.encrypted_content, + Some(serde_json::json!({"token": "opaque"})) + ); + assert_eq!(reasoning.status.as_deref(), Some("completed")); + } + + #[test] + fn completed_reasoning_null_and_empty_fields_are_authoritative_independently() { + let content_null = [ + r#"data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_content_null","type":"reasoning"}}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.done","item_id":"rs_content_null","output_index":0,"content_index":0,"text":"discarded content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.done","item_id":"rs_content_null","output_index":0,"summary_index":0,"text":"kept summary"}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_content_null","type":"reasoning","content":null}}"#.to_owned(), + ]; + let summary_empty = [ + r#"data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_summary_empty","type":"reasoning"}}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.done","item_id":"rs_summary_empty","output_index":0,"content_index":0,"text":"kept content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.done","item_id":"rs_summary_empty","output_index":0,"summary_index":0,"text":"discarded summary"}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_summary_empty","type":"reasoning","summary":[]}}"#.to_owned(), + ]; + + let content_null = ResponseAccumulator::from_sse_lines(content_null, None); + let OutputItem::Reasoning(content_null) = &content_null.output[0] else { + panic!("expected reasoning output"); + }; + assert!(content_null.content.is_empty()); + assert_eq!(content_null.summary[0]["text"], "kept summary"); + + let summary_empty = ResponseAccumulator::from_sse_lines(summary_empty, None); + let OutputItem::Reasoning(summary_empty) = &summary_empty.output[0] else { + panic!("expected reasoning output"); + }; + assert_eq!(summary_empty.content[0].text, "kept content"); + assert!(summary_empty.summary.is_empty()); + } + + #[test] + fn streaming_and_nonstreaming_nullable_reasoning_fields_are_equivalent() { + let streaming = ResponseAccumulator::from_sse_lines( + [r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":null,"summary":null,"encrypted_content":null,"status":"completed"}}"#.to_owned()], + None, + ); + let nonstreaming = ResponseAccumulator::from_json( + r#"{"id":"resp_1","status":"completed","output":[{"id":"rs_1","type":"reasoning","content":null,"summary":null,"encrypted_content":null,"status":"completed"}]}"#, + None, + ) + .unwrap(); + + assert_eq!( + serde_json::to_value(&streaming.output).unwrap(), + serde_json::to_value(&nonstreaming.output).unwrap() + ); + } + + #[test] + fn done_only_reasoning_uses_output_index_order() { + let lines = [ + r#"data: {"type":"response.output_item.added","output_index":1,"item":{"id":"msg_1","type":"message"}}"#.to_owned(), + r#"data: {"type":"response.output_text.delta","item_id":"msg_1","output_index":1,"content_index":0,"delta":"answer"}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[{"type":"reasoning_text","text":"thinking"}],"summary":[],"encrypted_content":null,"status":"completed"}}"#.to_owned(), + r#"data: {"type":"response.completed","response":{"id":"resp_1","status":"completed"}}"#.to_owned(), + ]; + + let acc = ResponseAccumulator::from_sse_lines(lines, None); + + assert_eq!(acc.output.len(), 2); + assert!(matches!(acc.output[0], OutputItem::Reasoning(_))); + assert!(matches!(acc.output[1], OutputItem::Message(_))); + } + + #[test] + fn malformed_completed_reasoning_retains_done_fields() { + let lines = [ + r#"data: {"type":"response.output_item.added","output_index":0,"item":{"id":"rs_1","type":"reasoning"}}"#.to_owned(), + r#"data: {"type":"response.reasoning_text.done","item_id":"rs_1","output_index":0,"content_index":0,"text":"completed content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.done","item_id":"rs_1","output_index":0,"summary_index":0,"text":"completed summary"}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":"malformed","summary":[{"type":"summary_text","text":"ignored completion"}],"encrypted_content":"ignored"}}"#.to_owned(), + r#"data: {"type":"response.completed","response":{"id":"resp_1","status":"completed"}}"#.to_owned(), + ]; + + let acc = ResponseAccumulator::from_sse_lines(lines, None); + let OutputItem::Reasoning(reasoning) = &acc.output[0] else { + panic!("expected reasoning output"); + }; + + assert_eq!(reasoning.content[0].text, "completed content"); + assert_eq!(reasoning.summary[0]["text"], "completed summary"); + assert!(reasoning.encrypted_content.is_none()); + } + #[test] fn test_accumulator_message_then_reasoning_preserves_order() { let lines = vec![ diff --git a/crates/agentic-server-core/src/storage/models/item.rs b/crates/agentic-server-core/src/storage/models/item.rs index e6cc81a8..821d1813 100644 --- a/crates/agentic-server-core/src/storage/models/item.rs +++ b/crates/agentic-server-core/src/storage/models/item.rs @@ -358,22 +358,42 @@ mod tests { } #[test] - fn test_as_inout_uses_stored_kind_for_reasoning_output() { + fn complete_reasoning_round_trip_strips_storage_marker() { let mut reasoning = ReasoningOutput::new("rs_1"); - reasoning.content.push(ReasoningTextContent::new("thinking...")); + reasoning.content.extend([ + ReasoningTextContent::new("first thought"), + ReasoningTextContent::new("second thought"), + ]); + reasoning + .summary + .push(serde_json::json!({"type": "summary_text", "text": "concise summary"})); + reasoning.encrypted_content = Some(serde_json::json!({"ciphertext": "opaque"})); + reasoning.status = Some("completed".to_owned()); let stored = InOutItem::Output(OutputItem::Reasoning(reasoning)); + let stored_json = String::try_from(&stored).expect("serialization failed"); + assert!(stored_json.contains(STORED_ITEM_KIND_KEY)); let item = Item { id: "item_reasoning".to_string(), - data: String::try_from(&stored).expect("serialization failed"), + data: stored_json, created_at: 1_704_067_200, conversation_id: None, seq: None, }; - assert!(matches!( - item.as_inout(), - Some(InOutItem::Output(OutputItem::Reasoning(_))) - )); + let Some(InOutItem::Output(OutputItem::Reasoning(reasoning))) = item.as_inout() else { + panic!("expected stored reasoning output"); + }; + assert_eq!(reasoning.id, "rs_1"); + assert_eq!(reasoning.content.len(), 2); + assert_eq!(reasoning.summary[0]["text"], "concise summary"); + assert_eq!( + reasoning.encrypted_content, + Some(serde_json::json!({"ciphertext": "opaque"})) + ); + assert_eq!(reasoning.status.as_deref(), Some("completed")); + + let reconstructed = serde_json::to_value(OutputItem::Reasoning(reasoning)).expect("reasoning value"); + assert!(reconstructed.get(STORED_ITEM_KIND_KEY).is_none()); } #[test] diff --git a/crates/agentic-server-core/src/types/io/output.rs b/crates/agentic-server-core/src/types/io/output.rs index 85867627..16439496 100644 --- a/crates/agentic-server-core/src/types/io/output.rs +++ b/crates/agentic-server-core/src/types/io/output.rs @@ -540,14 +540,22 @@ impl ReasoningTextContent { pub struct ReasoningOutput { #[serde(default)] pub id: String, - #[serde(default)] + #[serde(default, deserialize_with = "deserialize_nullable_vec")] pub content: Vec, - #[serde(default)] + #[serde(default, deserialize_with = "deserialize_nullable_vec")] pub summary: Vec, pub encrypted_content: Option, pub status: Option, } +fn deserialize_nullable_vec<'de, D, T>(deserializer: D) -> Result, D::Error> +where + D: Deserializer<'de>, + T: Deserialize<'de>, +{ + Option::>::deserialize(deserializer).map(Option::unwrap_or_default) +} + impl ReasoningOutput { pub fn new(id: impl Into) -> Self { Self { @@ -564,15 +572,32 @@ impl TryFrom<&EventPayload> for ReasoningOutput { type Error = ExecutorError; fn try_from(payload: &EventPayload) -> Result { - let EventPayload::OutputItemAdded { item_id, .. } = payload else { - return Err(ExecutorError::ParseError("expected OutputItemAdded payload".into())); - }; - let id = if item_id.is_empty() { - uuid7_str("rs_") - } else { - item_id.clone() - }; - Ok(Self::new(id)) + match payload { + EventPayload::OutputItemAdded { item_id, .. } => { + let id = if item_id.is_empty() { + uuid7_str("rs_") + } else { + item_id.clone() + }; + Ok(Self::new(id)) + } + EventPayload::OutputItemDone { item, .. } => { + let Some(OutputItem::Reasoning(item)) = deserialize_from_value_opt::(item.clone()) else { + return Err(ExecutorError::ParseError( + "expected a complete reasoning output item".into(), + )); + }; + if item.id.is_empty() { + return Err(ExecutorError::ParseError( + "complete reasoning output item is missing its id".into(), + )); + } + Ok(item) + } + _ => Err(ExecutorError::ParseError( + "expected a reasoning output-item lifecycle payload".into(), + )), + } } } @@ -587,21 +612,62 @@ pub trait ApplyDone { impl ApplyDone for ReasoningOutput { fn apply_done(&mut self, payload: &EventPayload, buffer: &mut String) { - let EventPayload::ReasoningDone { text, .. } = payload else { - return; - }; - let text = if text.is_empty() { - std::mem::take(buffer) - } else { - buffer.clear(); - text.clone() - }; - if !text.is_empty() { - self.content.push(ReasoningTextContent::new(text)); + match payload { + EventPayload::ReasoningTextDone { + text, content_index, .. + } => { + let text = final_text(text, buffer); + if !text.is_empty() { + insert_at_part_index(&mut self.content, *content_index, ReasoningTextContent::new(text)); + } + } + EventPayload::ReasoningSummaryTextDone { + text, summary_index, .. + } => { + let text = final_text(text, buffer); + if !text.is_empty() { + insert_at_part_index( + &mut self.summary, + *summary_index, + serde_json::json!({"type": "summary_text", "text": text}), + ); + } + } + EventPayload::OutputItemDone { item, .. } => { + let Some(raw_item) = item.as_object() else { + return; + }; + let Ok(mut completed) = Self::try_from(payload) else { + return; + }; + + if !raw_item.contains_key("content") { + completed.content = std::mem::take(&mut self.content); + } + if !raw_item.contains_key("summary") { + completed.summary = std::mem::take(&mut self.summary); + } + *self = completed; + } + _ => {} } } } +fn insert_at_part_index(parts: &mut Vec, part_index: u32, part: T) { + let index = usize::try_from(part_index).unwrap_or(usize::MAX).min(parts.len()); + parts.insert(index, part); +} + +fn final_text(text: &str, buffer: &mut String) -> String { + if text.is_empty() { + std::mem::take(buffer) + } else { + buffer.clear(); + text.to_owned() + } +} + impl ApplyDone for FunctionToolCall { fn apply_done(&mut self, payload: &EventPayload, buffer: &mut String) { match payload { @@ -868,6 +934,94 @@ mod tests { assert_eq!(serialized["id"], "rs_abc"); } + #[test] + fn reasoning_output_builds_from_added_and_applies_indexed_done_events() { + let added = EventPayload::OutputItemAdded { + item_id: "rs_1".to_owned(), + item_type: crate::events::SSEItemType::Reasoning, + output_index: 2, + name: None, + namespace: None, + call_id: None, + }; + let mut item = ReasoningOutput::try_from(&added).unwrap(); + + for (content_index, text) in [(1, "second thought"), (0, "first thought")] { + item.apply_done( + &EventPayload::ReasoningTextDone { + text: text.to_owned(), + item_id: "rs_1".to_owned(), + output_index: 2, + content_index, + }, + &mut String::new(), + ); + } + for (summary_index, text) in [(1, "second summary"), (0, "first summary")] { + item.apply_done( + &EventPayload::ReasoningSummaryTextDone { + text: text.to_owned(), + item_id: "rs_1".to_owned(), + output_index: 2, + summary_index, + }, + &mut String::new(), + ); + } + + assert_eq!(item.id, "rs_1"); + assert_eq!( + item.content.iter().map(|part| part.text.as_str()).collect::>(), + ["first thought", "second thought"] + ); + assert_eq!(item.summary[0]["text"], "first summary"); + assert_eq!(item.summary[1]["text"], "second summary"); + } + + #[test] + fn reasoning_output_done_owns_authoritative_field_reconciliation() { + let mut item = ReasoningOutput::new("rs_1"); + item.content.push(ReasoningTextContent::new("buffered thought")); + item.summary + .push(serde_json::json!({"type": "summary_text", "text": "buffered summary"})); + let done = EventPayload::OutputItemDone { + item_id: "rs_1".to_owned(), + item_type: crate::events::SSEItemType::Reasoning, + output_index: 0, + item: serde_json::json!({ + "id": "rs_1", + "type": "reasoning", + "summary": null, + "encrypted_content": "opaque-state", + "status": "completed", + }), + }; + + let parsed = ReasoningOutput::try_from(&done).unwrap(); + assert!(parsed.content.is_empty()); + assert!(parsed.summary.is_empty()); + + item.apply_done(&done, &mut String::new()); + assert_eq!(item.content[0].text, "buffered thought"); + assert!(item.summary.is_empty()); + assert_eq!(item.encrypted_content, Some(serde_json::json!("opaque-state"))); + assert_eq!(item.status.as_deref(), Some("completed")); + + let before = serde_json::to_value(&item).unwrap(); + let malformed = EventPayload::OutputItemDone { + item_id: "rs_1".to_owned(), + item_type: crate::events::SSEItemType::Reasoning, + output_index: 0, + item: serde_json::json!({ + "id": "rs_1", + "type": "reasoning", + "content": "not-an-array", + }), + }; + item.apply_done(&malformed, &mut String::new()); + assert_eq!(serde_json::to_value(item).unwrap(), before); + } + #[test] fn reasoning_input_round_trips_through_serde() { let reasoning = ReasoningOutput::new("rs_1"); diff --git a/crates/agentic-server-core/tests/event_normalizer_test.rs b/crates/agentic-server-core/tests/event_normalizer_test.rs index 1304e69e..33f1c33e 100644 --- a/crates/agentic-server-core/tests/event_normalizer_test.rs +++ b/crates/agentic-server-core/tests/event_normalizer_test.rs @@ -239,27 +239,43 @@ fn test_no_sequence_number() { #[test] fn test_reasoning_delta() { - let line = r#"data: {"type":"response.reasoning_summary_text.delta","delta":"Let me think","item_id":"rs_1","sequence_number":3}"#; + let line = r#"data: {"type":"response.reasoning_summary_text.delta","delta":"Let me think","item_id":"rs_1","output_index":2,"summary_index":1,"sequence_number":3}"#; let frame = normalize_sse_line(line).unwrap(); assert_eq!(frame.event_type, SSEEventType::ReasoningSummaryTextDelta); - if let EventPayload::ReasoningDelta { delta, item_id } = &frame.payload { + if let EventPayload::ReasoningSummaryTextDelta { + delta, + item_id, + output_index, + summary_index, + } = &frame.payload + { assert_eq!(delta, "Let me think"); assert_eq!(item_id, "rs_1"); + assert_eq!(*output_index, 2); + assert_eq!(*summary_index, 1); } else { - panic!("expected ReasoningDelta payload"); + panic!("expected ReasoningSummaryTextDelta payload"); } } #[test] fn test_reasoning_done_reads_text_not_delta() { - let line = r#"data: {"type":"response.reasoning_summary_text.done","text":"Full reasoning summary here","item_id":"rs_1","sequence_number":5}"#; + let line = r#"data: {"type":"response.reasoning_summary_text.done","text":"Full reasoning summary here","item_id":"rs_1","output_index":2,"summary_index":1,"sequence_number":5}"#; let frame = normalize_sse_line(line).unwrap(); assert_eq!(frame.event_type, SSEEventType::ReasoningSummaryTextDone); - if let EventPayload::ReasoningDone { text, item_id } = &frame.payload { + if let EventPayload::ReasoningSummaryTextDone { + text, + item_id, + output_index, + summary_index, + } = &frame.payload + { assert_eq!(text, "Full reasoning summary here"); assert_eq!(item_id, "rs_1"); + assert_eq!(*output_index, 2); + assert_eq!(*summary_index, 1); } else { - panic!("expected ReasoningDone payload"); + panic!("expected ReasoningSummaryTextDone payload"); } } @@ -268,11 +284,19 @@ fn test_reasoning_text_delta() { let line = r#"data: {"type":"response.reasoning_text.delta","delta":"The user asks","item_id":"rs_1","output_index":0,"content_index":0,"sequence_number":4}"#; let frame = normalize_sse_line(line).unwrap(); assert_eq!(frame.event_type, SSEEventType::ReasoningTextDelta); - if let EventPayload::ReasoningDelta { delta, item_id } = &frame.payload { + if let EventPayload::ReasoningTextDelta { + delta, + item_id, + output_index, + content_index, + } = &frame.payload + { assert_eq!(delta, "The user asks"); assert_eq!(item_id, "rs_1"); + assert_eq!(*output_index, 0); + assert_eq!(*content_index, 0); } else { - panic!("expected ReasoningDelta payload"); + panic!("expected ReasoningTextDelta payload"); } } @@ -281,11 +305,19 @@ fn test_reasoning_text_done() { let line = r#"data: {"type":"response.reasoning_text.done","text":"The user asks about math.","item_id":"rs_1","output_index":0,"content_index":0,"sequence_number":10}"#; let frame = normalize_sse_line(line).unwrap(); assert_eq!(frame.event_type, SSEEventType::ReasoningTextDone); - if let EventPayload::ReasoningDone { text, item_id } = &frame.payload { + if let EventPayload::ReasoningTextDone { + text, + item_id, + output_index, + content_index, + } = &frame.payload + { assert_eq!(text, "The user asks about math."); assert_eq!(item_id, "rs_1"); + assert_eq!(*output_index, 0); + assert_eq!(*content_index, 0); } else { - panic!("expected ReasoningDone payload"); + panic!("expected ReasoningTextDone payload"); } }