From f0b509a18bc9c7aa77e160c140ae65dc02d1164e Mon Sep 17 00:00:00 2001 From: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> Date: Thu, 27 Aug 2026 23:43:58 -0400 Subject: [PATCH 1/4] fix: reconcile complete streamed reasoning items Signed-off-by: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> --- .../src/events/normalize.rs | 36 +- .../agentic-server-core/src/events/types.rs | 30 +- .../src/executor/accumulator.rs | 467 ++++++++++++++++-- .../src/storage/models/item.rs | 34 +- .../src/types/io/output.rs | 47 +- .../tests/event_normalizer_test.rs | 52 +- 6 files changed, 599 insertions(+), 67 deletions(-) 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..107cc3af 100644 --- a/crates/agentic-server-core/src/executor/accumulator.rs +++ b/crates/agentic-server-core/src/executor/accumulator.rs @@ -7,6 +7,7 @@ //! runs on a blocking thread while the async task continues reading from the //! network — keeping the tokio executor thread free between chunk arrivals. +use std::collections::BTreeMap; use std::pin::Pin; use std::sync::mpsc; @@ -28,17 +29,112 @@ use crate::types::request_response::{IncompleteDetails, ResponsePayload}; use crate::utils::common::{deserialize_from_str, deserialize_from_value_opt}; use crate::utils::uuid7_str; +#[derive(Default)] +struct ReasoningFallback { + content: BTreeMap, + summary: BTreeMap, +} + +impl ReasoningFallback { + fn push_content_delta(&mut self, content_index: u32, delta: &str) { + self.content.entry(content_index).or_default().push_str(delta); + } + + fn finish_content(&mut self, content_index: u32, text: &str) { + finish_indexed_text(&mut self.content, content_index, text); + } + + fn push_summary_delta(&mut self, summary_index: u32, delta: &str) { + self.summary.entry(summary_index).or_default().push_str(delta); + } + + fn finish_summary(&mut self, summary_index: u32, text: &str) { + finish_indexed_text(&mut self.summary, summary_index, text); + } + + fn append_content_to(&mut self, content: &mut Vec) { + content.extend( + std::mem::take(&mut self.content) + .into_values() + .filter(|text| !text.is_empty()) + .map(ReasoningTextContent::new), + ); + } + + fn append_summary_to(&mut self, summary: &mut Vec) { + summary.extend( + std::mem::take(&mut self.summary) + .into_values() + .filter(|text| !text.is_empty()) + .map(|text| serde_json::json!({"type": "summary_text", "text": text})), + ); + } + + fn append_to(mut self, item: &mut ReasoningOutput) { + self.append_content_to(&mut item.content); + self.append_summary_to(&mut item.summary); + } +} + +fn finish_indexed_text(parts: &mut BTreeMap, index: u32, text: &str) { + let part = parts.entry(index).or_default(); + if !text.is_empty() { + text.clone_into(part); + } +} + +#[derive(Clone, Copy)] +struct ReasoningFieldPresence { + content: bool, + summary: bool, +} + +fn parse_completed_reasoning(raw_item: &serde_json::Value) -> Option<(ReasoningOutput, ReasoningFieldPresence)> { + let raw_object = raw_item.as_object()?; + let presence = ReasoningFieldPresence { + content: raw_object.contains_key("content"), + summary: raw_object.contains_key("summary"), + }; + let OutputItem::Reasoning(item) = deserialize_from_value_opt::(raw_item.clone())? else { + return None; + }; + if item.id.is_empty() { + return None; + } + Some((item, presence)) +} + /// Tracks a single output item currently being streamed, together with its /// accumulated text/arguments buffer. enum InFlight { - Message { item: OutputMessage, text: String }, - Reasoning { item: ReasoningOutput, text: String }, - FunctionCall { item: FunctionToolCall, arguments: String }, - CustomToolCall { item: CustomToolCall, input: String }, - WebSearchCall { item: Option }, - McpCall { item: McpCall }, - McpListTools { item: McpListTools }, - Compaction { item: CompactionItem }, + Message { + item: OutputMessage, + text: String, + }, + Reasoning { + item: ReasoningOutput, + fallback: ReasoningFallback, + }, + FunctionCall { + item: FunctionToolCall, + arguments: String, + }, + CustomToolCall { + item: CustomToolCall, + input: String, + }, + WebSearchCall { + item: Option, + }, + McpCall { + item: McpCall, + }, + McpListTools { + item: McpListTools, + }, + Compaction { + item: CompactionItem, + }, } impl std::fmt::Debug for InFlight { @@ -59,10 +155,8 @@ 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)); - } + Self::Reasoning { mut item, fallback } => { + fallback.append_to(&mut item); Some(OutputItem::Reasoning(item)) } Self::FunctionCall { mut item, arguments } => { @@ -357,6 +451,9 @@ impl ResponseAccumulator { /// frame (e.g. [`StreamTee`](future)) can call this directly without /// re-parsing from a raw line. pub(crate) fn process_event(&mut self, frame: &EventFrame) { + if self.process_reasoning_event(frame) { + return; + } match (&frame.event_type, &frame.payload) { (SSEEventType::ResponseCreated, EventPayload::Response { id, .. }) if !id.is_empty() => { self.response_id.clone_from(id); @@ -365,21 +462,7 @@ impl ResponseAccumulator { self.start_output_item(payload); } (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); - } + self.complete_output_item(payload); } ( SSEEventType::FunctionCallArgumentsDelta, @@ -448,6 +531,65 @@ impl ResponseAccumulator { } } + fn process_reasoning_event(&mut self, frame: &EventFrame) -> bool { + match (&frame.event_type, &frame.payload) { + ( + SSEEventType::ReasoningTextDelta, + EventPayload::ReasoningTextDelta { + delta, + item_id, + output_index, + content_index, + }, + ) => { + if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { + fallback.push_content_delta(*content_index, delta); + } + } + ( + SSEEventType::ReasoningTextDone, + EventPayload::ReasoningTextDone { + text, + item_id, + output_index, + content_index, + }, + ) => { + if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { + fallback.finish_content(*content_index, text); + } + } + ( + SSEEventType::ReasoningSummaryTextDelta, + EventPayload::ReasoningSummaryTextDelta { + delta, + item_id, + output_index, + summary_index, + }, + ) => { + if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { + fallback.push_summary_delta(*summary_index, delta); + } + } + ( + SSEEventType::ReasoningSummaryTextDone, + EventPayload::ReasoningSummaryTextDone { + text, + item_id, + output_index, + summary_index, + }, + ) => { + if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { + fallback.finish_summary(*summary_index, text); + } + } + _ => return false, + } + true + } + fn start_output_item(&mut self, payload: &EventPayload) { let EventPayload::OutputItemAdded { item_id, @@ -461,7 +603,7 @@ impl ResponseAccumulator { let item = match item_type { SSEItemType::Reasoning => ReasoningOutput::try_from(payload).ok().map(|item| InFlight::Reasoning { item, - text: String::with_capacity(256), + fallback: ReasoningFallback::default(), }), SSEItemType::FunctionCall => FunctionToolCall::try_from(payload) .ok() @@ -519,7 +661,7 @@ impl ResponseAccumulator { self.usage = usage; } - fn complete_call_item(&mut self, payload: &EventPayload) { + fn complete_output_item(&mut self, payload: &EventPayload) { let EventPayload::OutputItemDone { item_id, item_type, @@ -530,6 +672,10 @@ impl ResponseAccumulator { else { return; }; + if *item_type == SSEItemType::Reasoning { + self.complete_reasoning_item(item_id, *output_index, raw_item); + 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()); if let Some(entry) = in_flight_key.as_deref().and_then(|key| self.in_flight.get_mut(key)) { @@ -573,6 +719,78 @@ impl ResponseAccumulator { } } + fn complete_reasoning_item(&mut self, item_id: &str, output_index: u32, raw_item: &serde_json::Value) { + let in_flight_key = self.in_flight_reasoning_key(item_id, output_index); + let completed = parse_completed_reasoning(raw_item); + + if let Some(key) = in_flight_key { + let Some((mut completed, presence)) = completed else { + return; + }; + let Some(entry) = self.in_flight.get_mut(&key) else { + return; + }; + entry.output_index = output_index; + let InFlight::Reasoning { item, fallback } = &mut entry.item else { + return; + }; + + if presence.content { + fallback.content.clear(); + } else { + completed.content = std::mem::take(&mut item.content); + fallback.append_content_to(&mut completed.content); + } + if presence.summary { + fallback.summary.clear(); + } else { + completed.summary = std::mem::take(&mut item.summary); + fallback.append_summary_to(&mut completed.summary); + } + *item = completed; + return; + } + + if let Some((completed, _)) = completed { + self.upsert_completed_reasoning(output_index, completed); + } + } + + fn upsert_completed_reasoning(&mut self, output_index: u32, item: ReasoningOutput) { + if let Some((existing_index, existing)) = self.completed.iter_mut().find_map(|(existing_index, output_item)| { + let OutputItem::Reasoning(existing) = output_item else { + return None; + }; + (existing.id == item.id).then_some((existing_index, existing)) + }) { + *existing_index = output_index; + *existing = item; + return; + } + self.completed.push((output_index, OutputItem::Reasoning(item))); + } + + fn in_flight_reasoning_fallback_mut(&mut self, item_id: &str, output_index: u32) -> Option<&mut ReasoningFallback> { + let key = self.in_flight_reasoning_key(item_id, output_index)?; + let InFlight::Reasoning { fallback, .. } = &mut self.in_flight.get_mut(&key)?.item else { + return None; + }; + Some(fallback) + } + + fn in_flight_reasoning_key(&self, item_id: &str, output_index: u32) -> Option { + self.in_flight + .get(item_id) + .filter(|entry| 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 +1434,197 @@ 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_content_and_summary_fallbacks_keep_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.delta","item_id":"rs_1","output_index":0,"content_index":1,"delta":"second"}"#.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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"first content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":1,"delta":"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_uses_fallback_only_for_omitted_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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"fallback content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"fallback 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, "fallback content"); + assert_eq!( + reasoning.summary, + [serde_json::json!({"type": "summary_text", "text": "fallback 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.delta","item_id":"rs_content_null","output_index":0,"content_index":0,"delta":"discarded content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_content_null","output_index":0,"summary_index":0,"delta":"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.delta","item_id":"rs_summary_empty","output_index":0,"content_index":0,"delta":"kept content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_summary_empty","output_index":0,"summary_index":0,"delta":"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_valid_fallbacks() { + 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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"fallback content"}"#.to_owned(), + r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"fallback 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, "fallback content"); + assert_eq!(reasoning.summary[0]["text"], "fallback summary"); + assert!(reasoning.encrypted_content.is_none()); + } + + #[test] + fn repeated_done_only_reasoning_is_upserted_by_id() { + let lines = [ + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[{"type":"reasoning_text","text":"first"}],"summary":[]}}"#.to_owned(), + r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[{"type":"reasoning_text","text":"final"}],"summary":[]}}"#.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); + let OutputItem::Reasoning(reasoning) = &acc.output[0] else { + panic!("expected reasoning output"); + }; + assert_eq!(reasoning.content[0].text, "final"); + } + #[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..7a83f825 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 { @@ -587,21 +595,34 @@ 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, .. } => { + let text = final_text(text, buffer); + if !text.is_empty() { + self.content.push(ReasoningTextContent::new(text)); + } + } + EventPayload::ReasoningSummaryTextDone { text, .. } => { + let text = final_text(text, buffer); + if !text.is_empty() { + self.summary + .push(serde_json::json!({"type": "summary_text", "text": text})); + } + } + _ => {} } } } +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 { 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"); } } From ea645c0b9b87127618c50fb88eada5751452c415 Mon Sep 17 00:00:00 2001 From: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> Date: Fri, 28 Aug 2026 00:24:11 -0400 Subject: [PATCH 2/4] refactor: centralize reasoning lifecycle reconciliation Signed-off-by: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> --- .../src/executor/accumulator.rs | 186 ++++++++---------- .../src/types/io/output.rs | 161 +++++++++++++-- 2 files changed, 224 insertions(+), 123 deletions(-) diff --git a/crates/agentic-server-core/src/executor/accumulator.rs b/crates/agentic-server-core/src/executor/accumulator.rs index 107cc3af..b15d852c 100644 --- a/crates/agentic-server-core/src/executor/accumulator.rs +++ b/crates/agentic-server-core/src/executor/accumulator.rs @@ -22,7 +22,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}; @@ -30,78 +30,49 @@ use crate::utils::common::{deserialize_from_str, deserialize_from_value_opt}; use crate::utils::uuid7_str; #[derive(Default)] -struct ReasoningFallback { +struct ReasoningBuffers { content: BTreeMap, summary: BTreeMap, } -impl ReasoningFallback { +impl ReasoningBuffers { fn push_content_delta(&mut self, content_index: u32, delta: &str) { self.content.entry(content_index).or_default().push_str(delta); } - fn finish_content(&mut self, content_index: u32, text: &str) { - finish_indexed_text(&mut self.content, content_index, text); + fn take_content(&mut self, content_index: u32) -> String { + self.content.remove(&content_index).unwrap_or_default() } fn push_summary_delta(&mut self, summary_index: u32, delta: &str) { self.summary.entry(summary_index).or_default().push_str(delta); } - fn finish_summary(&mut self, summary_index: u32, text: &str) { - finish_indexed_text(&mut self.summary, summary_index, text); - } - - fn append_content_to(&mut self, content: &mut Vec) { - content.extend( - std::mem::take(&mut self.content) - .into_values() - .filter(|text| !text.is_empty()) - .map(ReasoningTextContent::new), - ); - } - - fn append_summary_to(&mut self, summary: &mut Vec) { - summary.extend( - std::mem::take(&mut self.summary) - .into_values() - .filter(|text| !text.is_empty()) - .map(|text| serde_json::json!({"type": "summary_text", "text": text})), - ); - } - - fn append_to(mut self, item: &mut ReasoningOutput) { - self.append_content_to(&mut item.content); - self.append_summary_to(&mut item.summary); + fn take_summary(&mut self, summary_index: u32) -> String { + self.summary.remove(&summary_index).unwrap_or_default() } } -fn finish_indexed_text(parts: &mut BTreeMap, index: u32, text: &str) { - let part = parts.entry(index).or_default(); - if !text.is_empty() { - text.clone_into(part); +fn apply_reasoning_buffers(item: &mut ReasoningOutput, buffers: ReasoningBuffers, output_index: u32) { + for (content_index, mut buffer) in buffers.content { + let done = EventPayload::ReasoningTextDone { + text: String::new(), + item_id: item.id.clone(), + output_index, + content_index, + }; + item.apply_done(&done, &mut buffer); } -} - -#[derive(Clone, Copy)] -struct ReasoningFieldPresence { - content: bool, - summary: bool, -} -fn parse_completed_reasoning(raw_item: &serde_json::Value) -> Option<(ReasoningOutput, ReasoningFieldPresence)> { - let raw_object = raw_item.as_object()?; - let presence = ReasoningFieldPresence { - content: raw_object.contains_key("content"), - summary: raw_object.contains_key("summary"), - }; - let OutputItem::Reasoning(item) = deserialize_from_value_opt::(raw_item.clone())? else { - return None; - }; - if item.id.is_empty() { - return None; - } - Some((item, presence)) + for (summary_index, mut buffer) in buffers.summary { + let done = EventPayload::ReasoningSummaryTextDone { + text: String::new(), + item_id: item.id.clone(), + output_index, + summary_index, + }; + item.apply_done(&done, &mut buffer); + } } /// Tracks a single output item currently being streamed, together with its @@ -113,7 +84,7 @@ enum InFlight { }, Reasoning { item: ReasoningOutput, - fallback: ReasoningFallback, + buffers: ReasoningBuffers, }, FunctionCall { item: FunctionToolCall, @@ -153,10 +124,10 @@ impl std::fmt::Debug for InFlight { } impl InFlight { - fn finalize(self) -> Option { + fn finalize(self, output_index: u32) -> Option { match self { - Self::Reasoning { mut item, fallback } => { - fallback.append_to(&mut item); + Self::Reasoning { mut item, buffers } => { + apply_reasoning_buffers(&mut item, buffers, output_index); Some(OutputItem::Reasoning(item)) } Self::FunctionCall { mut item, arguments } => { @@ -353,11 +324,12 @@ impl ResponseAccumulator { /// Finalizes all streaming items in upstream `output_index` order. pub(crate) fn finalize_all(&mut self) { - self.completed.extend( - self.in_flight - .drain(..) - .filter_map(|(_, entry)| entry.item.finalize().map(|item| (entry.output_index, item))), - ); + self.completed.extend(self.in_flight.drain(..).filter_map(|(_, entry)| { + entry + .item + .finalize(entry.output_index) + .map(|item| (entry.output_index, item)) + })); self.completed.sort_by_key(|(output_index, _)| *output_index); self.output .extend(self.completed.drain(..).map(|(_, output_item)| output_item)); @@ -451,7 +423,7 @@ impl ResponseAccumulator { /// frame (e.g. [`StreamTee`](future)) can call this directly without /// re-parsing from a raw line. pub(crate) fn process_event(&mut self, frame: &EventFrame) { - if self.process_reasoning_event(frame) { + if self.route_reasoning_event(frame) { return; } match (&frame.event_type, &frame.payload) { @@ -531,7 +503,7 @@ impl ResponseAccumulator { } } - fn process_reasoning_event(&mut self, frame: &EventFrame) -> bool { + fn route_reasoning_event(&mut self, frame: &EventFrame) -> bool { match (&frame.event_type, &frame.payload) { ( SSEEventType::ReasoningTextDelta, @@ -542,21 +514,22 @@ impl ResponseAccumulator { content_index, }, ) => { - if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { - fallback.push_content_delta(*content_index, delta); + if let Some((_, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { + buffers.push_content_delta(*content_index, delta); } } ( SSEEventType::ReasoningTextDone, - EventPayload::ReasoningTextDone { - text, + payload @ EventPayload::ReasoningTextDone { item_id, output_index, content_index, + .. }, ) => { - if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { - fallback.finish_content(*content_index, text); + if let Some((item, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { + let mut buffer = buffers.take_content(*content_index); + item.apply_done(payload, &mut buffer); } } ( @@ -568,21 +541,22 @@ impl ResponseAccumulator { summary_index, }, ) => { - if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { - fallback.push_summary_delta(*summary_index, delta); + if let Some((_, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { + buffers.push_summary_delta(*summary_index, delta); } } ( SSEEventType::ReasoningSummaryTextDone, - EventPayload::ReasoningSummaryTextDone { - text, + payload @ EventPayload::ReasoningSummaryTextDone { item_id, output_index, summary_index, + .. }, ) => { - if let Some(fallback) = self.in_flight_reasoning_fallback_mut(item_id, *output_index) { - fallback.finish_summary(*summary_index, text); + if let Some((item, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { + let mut buffer = buffers.take_summary(*summary_index); + item.apply_done(payload, &mut buffer); } } _ => return false, @@ -603,7 +577,7 @@ impl ResponseAccumulator { let item = match item_type { SSEItemType::Reasoning => ReasoningOutput::try_from(payload).ok().map(|item| InFlight::Reasoning { item, - fallback: ReasoningFallback::default(), + buffers: ReasoningBuffers::default(), }), SSEItemType::FunctionCall => FunctionToolCall::try_from(payload) .ok() @@ -673,7 +647,7 @@ impl ResponseAccumulator { return; }; if *item_type == SSEItemType::Reasoning { - self.complete_reasoning_item(item_id, *output_index, raw_item); + self.complete_reasoning_item(payload); return; } let in_flight_key = self.in_flight_call_key(item_id, *item_type, *output_index); @@ -719,40 +693,30 @@ impl ResponseAccumulator { } } - fn complete_reasoning_item(&mut self, item_id: &str, output_index: u32, raw_item: &serde_json::Value) { - let in_flight_key = self.in_flight_reasoning_key(item_id, output_index); - let completed = parse_completed_reasoning(raw_item); + fn complete_reasoning_item(&mut self, payload: &EventPayload) { + let EventPayload::OutputItemDone { + item_id, output_index, .. + } = payload + else { + return; + }; + let in_flight_key = self.in_flight_reasoning_key(item_id, *output_index); if let Some(key) = in_flight_key { - let Some((mut completed, presence)) = completed else { - return; - }; let Some(entry) = self.in_flight.get_mut(&key) else { return; }; - entry.output_index = output_index; - let InFlight::Reasoning { item, fallback } = &mut entry.item else { + entry.output_index = *output_index; + let InFlight::Reasoning { item, buffers } = &mut entry.item else { return; }; - - if presence.content { - fallback.content.clear(); - } else { - completed.content = std::mem::take(&mut item.content); - fallback.append_content_to(&mut completed.content); - } - if presence.summary { - fallback.summary.clear(); - } else { - completed.summary = std::mem::take(&mut item.summary); - fallback.append_summary_to(&mut completed.summary); - } - *item = completed; + apply_reasoning_buffers(item, std::mem::take(buffers), *output_index); + item.apply_done(payload, &mut String::new()); return; } - if let Some((completed, _)) = completed { - self.upsert_completed_reasoning(output_index, completed); + if let Ok(completed) = ReasoningOutput::try_from(payload) { + self.upsert_completed_reasoning(*output_index, completed); } } @@ -770,12 +734,16 @@ impl ResponseAccumulator { self.completed.push((output_index, OutputItem::Reasoning(item))); } - fn in_flight_reasoning_fallback_mut(&mut self, item_id: &str, output_index: u32) -> Option<&mut ReasoningFallback> { + fn in_flight_reasoning_mut( + &mut self, + item_id: &str, + output_index: u32, + ) -> Option<(&mut ReasoningOutput, &mut ReasoningBuffers)> { let key = self.in_flight_reasoning_key(item_id, output_index)?; - let InFlight::Reasoning { fallback, .. } = &mut self.in_flight.get_mut(&key)?.item else { + let InFlight::Reasoning { item, buffers } = &mut self.in_flight.get_mut(&key)?.item else { return None; }; - Some(fallback) + Some((item, buffers)) } fn in_flight_reasoning_key(&self, item_id: &str, output_index: u32) -> Option { @@ -1464,7 +1432,7 @@ mod tests { } #[test] - fn reasoning_content_and_summary_fallbacks_keep_index_order() { + fn reasoning_content_and_summary_buffers_keep_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.delta","item_id":"rs_1","output_index":0,"content_index":1,"delta":"second"}"#.to_owned(), @@ -1498,7 +1466,7 @@ mod tests { } #[test] - fn completed_reasoning_uses_fallback_only_for_omitted_fields() { + fn completed_reasoning_uses_buffered_text_only_for_omitted_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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"fallback content"}"#.to_owned(), @@ -1589,7 +1557,7 @@ mod tests { } #[test] - fn malformed_completed_reasoning_retains_valid_fallbacks() { + fn malformed_completed_reasoning_retains_buffered_text() { 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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"fallback content"}"#.to_owned(), diff --git a/crates/agentic-server-core/src/types/io/output.rs b/crates/agentic-server-core/src/types/io/output.rs index 7a83f825..16439496 100644 --- a/crates/agentic-server-core/src/types/io/output.rs +++ b/crates/agentic-server-core/src/types/io/output.rs @@ -572,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(), + )), + } } } @@ -596,24 +613,52 @@ pub trait ApplyDone { impl ApplyDone for ReasoningOutput { fn apply_done(&mut self, payload: &EventPayload, buffer: &mut String) { match payload { - EventPayload::ReasoningTextDone { text, .. } => { + EventPayload::ReasoningTextDone { + text, content_index, .. + } => { let text = final_text(text, buffer); if !text.is_empty() { - self.content.push(ReasoningTextContent::new(text)); + insert_at_part_index(&mut self.content, *content_index, ReasoningTextContent::new(text)); } } - EventPayload::ReasoningSummaryTextDone { text, .. } => { + EventPayload::ReasoningSummaryTextDone { + text, summary_index, .. + } => { let text = final_text(text, buffer); if !text.is_empty() { - self.summary - .push(serde_json::json!({"type": "summary_text", "text": text})); + 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) @@ -889,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"); From 9535d3be01e2d45f7fb1f939bad5ddc10f8f2a87 Mon Sep 17 00:00:00 2001 From: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> Date: Fri, 28 Aug 2026 01:17:50 -0400 Subject: [PATCH 3/4] refactor: route reasoning done events through output item Signed-off-by: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> --- .../src/executor/accumulator.rs | 238 +++++------------- 1 file changed, 56 insertions(+), 182 deletions(-) diff --git a/crates/agentic-server-core/src/executor/accumulator.rs b/crates/agentic-server-core/src/executor/accumulator.rs index b15d852c..6bfdc8ba 100644 --- a/crates/agentic-server-core/src/executor/accumulator.rs +++ b/crates/agentic-server-core/src/executor/accumulator.rs @@ -7,7 +7,6 @@ //! runs on a blocking thread while the async task continues reading from the //! network — keeping the tokio executor thread free between chunk arrivals. -use std::collections::BTreeMap; use std::pin::Pin; use std::sync::mpsc; @@ -29,83 +28,17 @@ use crate::types::request_response::{IncompleteDetails, ResponsePayload}; use crate::utils::common::{deserialize_from_str, deserialize_from_value_opt}; use crate::utils::uuid7_str; -#[derive(Default)] -struct ReasoningBuffers { - content: BTreeMap, - summary: BTreeMap, -} - -impl ReasoningBuffers { - fn push_content_delta(&mut self, content_index: u32, delta: &str) { - self.content.entry(content_index).or_default().push_str(delta); - } - - fn take_content(&mut self, content_index: u32) -> String { - self.content.remove(&content_index).unwrap_or_default() - } - - fn push_summary_delta(&mut self, summary_index: u32, delta: &str) { - self.summary.entry(summary_index).or_default().push_str(delta); - } - - fn take_summary(&mut self, summary_index: u32) -> String { - self.summary.remove(&summary_index).unwrap_or_default() - } -} - -fn apply_reasoning_buffers(item: &mut ReasoningOutput, buffers: ReasoningBuffers, output_index: u32) { - for (content_index, mut buffer) in buffers.content { - let done = EventPayload::ReasoningTextDone { - text: String::new(), - item_id: item.id.clone(), - output_index, - content_index, - }; - item.apply_done(&done, &mut buffer); - } - - for (summary_index, mut buffer) in buffers.summary { - let done = EventPayload::ReasoningSummaryTextDone { - text: String::new(), - item_id: item.id.clone(), - output_index, - summary_index, - }; - item.apply_done(&done, &mut buffer); - } -} - /// Tracks a single output item currently being streamed, together with its /// accumulated text/arguments buffer. enum InFlight { - Message { - item: OutputMessage, - text: String, - }, - Reasoning { - item: ReasoningOutput, - buffers: ReasoningBuffers, - }, - FunctionCall { - item: FunctionToolCall, - arguments: String, - }, - CustomToolCall { - item: CustomToolCall, - input: String, - }, - WebSearchCall { - item: Option, - }, - McpCall { - item: McpCall, - }, - McpListTools { - item: McpListTools, - }, - Compaction { - item: CompactionItem, - }, + Message { item: OutputMessage, text: String }, + Reasoning { item: ReasoningOutput }, + FunctionCall { item: FunctionToolCall, arguments: String }, + CustomToolCall { item: CustomToolCall, input: String }, + WebSearchCall { item: Option }, + McpCall { item: McpCall }, + McpListTools { item: McpListTools }, + Compaction { item: CompactionItem }, } impl std::fmt::Debug for InFlight { @@ -124,12 +57,9 @@ impl std::fmt::Debug for InFlight { } impl InFlight { - fn finalize(self, output_index: u32) -> Option { + fn finalize(self) -> Option { match self { - Self::Reasoning { mut item, buffers } => { - apply_reasoning_buffers(&mut item, buffers, output_index); - 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; @@ -324,12 +254,11 @@ impl ResponseAccumulator { /// Finalizes all streaming items in upstream `output_index` order. pub(crate) fn finalize_all(&mut self) { - self.completed.extend(self.in_flight.drain(..).filter_map(|(_, entry)| { - entry - .item - .finalize(entry.output_index) - .map(|item| (entry.output_index, item)) - })); + self.completed.extend( + self.in_flight + .drain(..) + .filter_map(|(_, entry)| entry.item.finalize().map(|item| (entry.output_index, item))), + ); self.completed.sort_by_key(|(output_index, _)| *output_index); self.output .extend(self.completed.drain(..).map(|(_, output_item)| output_item)); @@ -423,9 +352,6 @@ impl ResponseAccumulator { /// frame (e.g. [`StreamTee`](future)) can call this directly without /// re-parsing from a raw line. pub(crate) fn process_event(&mut self, frame: &EventFrame) { - if self.route_reasoning_event(frame) { - return; - } match (&frame.event_type, &frame.payload) { (SSEEventType::ResponseCreated, EventPayload::Response { id, .. }) if !id.is_empty() => { self.response_id.clone_from(id); @@ -436,6 +362,22 @@ impl ResponseAccumulator { (SSEEventType::OutputItemDone, payload @ EventPayload::OutputItemDone { .. }) => { self.complete_output_item(payload); } + ( + 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()); + } + } ( SSEEventType::FunctionCallArgumentsDelta, EventPayload::FunctionCallArgsDelta { @@ -503,67 +445,6 @@ impl ResponseAccumulator { } } - fn route_reasoning_event(&mut self, frame: &EventFrame) -> bool { - match (&frame.event_type, &frame.payload) { - ( - SSEEventType::ReasoningTextDelta, - EventPayload::ReasoningTextDelta { - delta, - item_id, - output_index, - content_index, - }, - ) => { - if let Some((_, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { - buffers.push_content_delta(*content_index, delta); - } - } - ( - SSEEventType::ReasoningTextDone, - payload @ EventPayload::ReasoningTextDone { - item_id, - output_index, - content_index, - .. - }, - ) => { - if let Some((item, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { - let mut buffer = buffers.take_content(*content_index); - item.apply_done(payload, &mut buffer); - } - } - ( - SSEEventType::ReasoningSummaryTextDelta, - EventPayload::ReasoningSummaryTextDelta { - delta, - item_id, - output_index, - summary_index, - }, - ) => { - if let Some((_, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { - buffers.push_summary_delta(*summary_index, delta); - } - } - ( - SSEEventType::ReasoningSummaryTextDone, - payload @ EventPayload::ReasoningSummaryTextDone { - item_id, - output_index, - summary_index, - .. - }, - ) => { - if let Some((item, buffers)) = self.in_flight_reasoning_mut(item_id, *output_index) { - let mut buffer = buffers.take_summary(*summary_index); - item.apply_done(payload, &mut buffer); - } - } - _ => return false, - } - true - } - fn start_output_item(&mut self, payload: &EventPayload) { let EventPayload::OutputItemAdded { item_id, @@ -575,10 +456,9 @@ impl ResponseAccumulator { return; }; let item = match item_type { - SSEItemType::Reasoning => ReasoningOutput::try_from(payload).ok().map(|item| InFlight::Reasoning { - item, - buffers: ReasoningBuffers::default(), - }), + SSEItemType::Reasoning => ReasoningOutput::try_from(payload) + .ok() + .map(|item| InFlight::Reasoning { item }), SSEItemType::FunctionCall => FunctionToolCall::try_from(payload) .ok() .map(|item| InFlight::FunctionCall { @@ -707,10 +587,9 @@ impl ResponseAccumulator { return; }; entry.output_index = *output_index; - let InFlight::Reasoning { item, buffers } = &mut entry.item else { + let InFlight::Reasoning { item } = &mut entry.item else { return; }; - apply_reasoning_buffers(item, std::mem::take(buffers), *output_index); item.apply_done(payload, &mut String::new()); return; } @@ -734,22 +613,18 @@ impl ResponseAccumulator { self.completed.push((output_index, OutputItem::Reasoning(item))); } - fn in_flight_reasoning_mut( - &mut self, - item_id: &str, - output_index: u32, - ) -> Option<(&mut ReasoningOutput, &mut ReasoningBuffers)> { + 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, buffers } = &mut self.in_flight.get_mut(&key)?.item else { + let InFlight::Reasoning { item } = &mut self.in_flight.get_mut(&key)?.item else { return None; }; - Some((item, buffers)) + Some(item) } fn in_flight_reasoning_key(&self, item_id: &str, output_index: u32) -> Option { self.in_flight .get(item_id) - .filter(|entry| matches!(entry.item, InFlight::Reasoning { .. })) + .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)| { @@ -1432,13 +1307,12 @@ mod tests { } #[test] - fn reasoning_content_and_summary_buffers_keep_index_order() { + 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.delta","item_id":"rs_1","output_index":0,"content_index":1,"delta":"second"}"#.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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"first content"}"#.to_owned(), - r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":1,"delta":"second summary"}"#.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(), ]; @@ -1466,11 +1340,11 @@ mod tests { } #[test] - fn completed_reasoning_uses_buffered_text_only_for_omitted_fields() { + 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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"fallback content"}"#.to_owned(), - r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"fallback summary"}"#.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(), ]; @@ -1480,10 +1354,10 @@ mod tests { panic!("expected reasoning output"); }; - assert_eq!(reasoning.content[0].text, "fallback content"); + assert_eq!(reasoning.content[0].text, "completed content"); assert_eq!( reasoning.summary, - [serde_json::json!({"type": "summary_text", "text": "fallback summary"})] + [serde_json::json!({"type": "summary_text", "text": "completed summary"})] ); assert_eq!( reasoning.encrypted_content, @@ -1496,14 +1370,14 @@ mod tests { 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.delta","item_id":"rs_content_null","output_index":0,"content_index":0,"delta":"discarded content"}"#.to_owned(), - r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_content_null","output_index":0,"summary_index":0,"delta":"kept summary"}"#.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.delta","item_id":"rs_summary_empty","output_index":0,"content_index":0,"delta":"kept content"}"#.to_owned(), - r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_summary_empty","output_index":0,"summary_index":0,"delta":"discarded summary"}"#.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(), ]; @@ -1557,11 +1431,11 @@ mod tests { } #[test] - fn malformed_completed_reasoning_retains_buffered_text() { + 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.delta","item_id":"rs_1","output_index":0,"content_index":0,"delta":"fallback content"}"#.to_owned(), - r#"data: {"type":"response.reasoning_summary_text.delta","item_id":"rs_1","output_index":0,"summary_index":0,"delta":"fallback summary"}"#.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(), ]; @@ -1571,8 +1445,8 @@ mod tests { panic!("expected reasoning output"); }; - assert_eq!(reasoning.content[0].text, "fallback content"); - assert_eq!(reasoning.summary[0]["text"], "fallback summary"); + assert_eq!(reasoning.content[0].text, "completed content"); + assert_eq!(reasoning.summary[0]["text"], "completed summary"); assert!(reasoning.encrypted_content.is_none()); } From 2f6ba22de7f77b2573269a205965baa6018b8046 Mon Sep 17 00:00:00 2001 From: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> Date: Fri, 28 Aug 2026 03:57:25 -0400 Subject: [PATCH 4/4] refactor: use generic reasoning completion path Signed-off-by: StevenWang-CY <203932027+StevenWang-CY@users.noreply.github.com> --- .../src/executor/accumulator.rs | 81 ++++--------------- 1 file changed, 15 insertions(+), 66 deletions(-) diff --git a/crates/agentic-server-core/src/executor/accumulator.rs b/crates/agentic-server-core/src/executor/accumulator.rs index 6bfdc8ba..4c3d4289 100644 --- a/crates/agentic-server-core/src/executor/accumulator.rs +++ b/crates/agentic-server-core/src/executor/accumulator.rs @@ -360,7 +360,7 @@ impl ResponseAccumulator { self.start_output_item(payload); } (SSEEventType::OutputItemDone, payload @ EventPayload::OutputItemDone { .. }) => { - self.complete_output_item(payload); + self.complete_call_item(payload); } ( SSEEventType::ReasoningTextDone, @@ -515,7 +515,7 @@ impl ResponseAccumulator { self.usage = usage; } - fn complete_output_item(&mut self, payload: &EventPayload) { + fn complete_call_item(&mut self, payload: &EventPayload) { let EventPayload::OutputItemDone { item_id, item_type, @@ -526,14 +526,19 @@ impl ResponseAccumulator { else { return; }; - if *item_type == SSEItemType::Reasoning { - self.complete_reasoning_item(payload); - 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,46 +579,6 @@ impl ResponseAccumulator { } } - fn complete_reasoning_item(&mut self, payload: &EventPayload) { - let EventPayload::OutputItemDone { - item_id, output_index, .. - } = payload - else { - return; - }; - let in_flight_key = self.in_flight_reasoning_key(item_id, *output_index); - - if let Some(key) = in_flight_key { - let Some(entry) = self.in_flight.get_mut(&key) else { - return; - }; - entry.output_index = *output_index; - let InFlight::Reasoning { item } = &mut entry.item else { - return; - }; - item.apply_done(payload, &mut String::new()); - return; - } - - if let Ok(completed) = ReasoningOutput::try_from(payload) { - self.upsert_completed_reasoning(*output_index, completed); - } - } - - fn upsert_completed_reasoning(&mut self, output_index: u32, item: ReasoningOutput) { - if let Some((existing_index, existing)) = self.completed.iter_mut().find_map(|(existing_index, output_item)| { - let OutputItem::Reasoning(existing) = output_item else { - return None; - }; - (existing.id == item.id).then_some((existing_index, existing)) - }) { - *existing_index = output_index; - *existing = item; - return; - } - self.completed.push((output_index, OutputItem::Reasoning(item))); - } - 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 { @@ -1450,23 +1416,6 @@ mod tests { assert!(reasoning.encrypted_content.is_none()); } - #[test] - fn repeated_done_only_reasoning_is_upserted_by_id() { - let lines = [ - r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[{"type":"reasoning_text","text":"first"}],"summary":[]}}"#.to_owned(), - r#"data: {"type":"response.output_item.done","output_index":0,"item":{"id":"rs_1","type":"reasoning","content":[{"type":"reasoning_text","text":"final"}],"summary":[]}}"#.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); - let OutputItem::Reasoning(reasoning) = &acc.output[0] else { - panic!("expected reasoning output"); - }; - assert_eq!(reasoning.content[0].text, "final"); - } - #[test] fn test_accumulator_message_then_reasoning_preserves_order() { let lines = vec![