agentic_core/executor/
accumulator.rs

1//! Response accumulation and parsing utilities.
2//!
3//! Handles both streaming (SSE) and non-streaming JSON response formats,
4//! accumulating chunks into a unified `ResponsePayload` structure.
5//!
6//! Streaming path uses a channel + `spawn_blocking` so that SSE JSON parsing
7//! runs on a blocking thread while the async task continues reading from the
8//! network — keeping the tokio executor thread free between chunk arrivals.
9
10use std::pin::Pin;
11use std::sync::mpsc;
12
13use indexmap::IndexMap;
14
15use futures::{Stream, StreamExt};
16
17use crate::events::{EventFrame, EventPayload, SSEEventType, SSEItemType, normalize_sse_line};
18use crate::executor::error::{ExecutorError, ExecutorResult};
19use crate::types::event::{MessageStatus, ResponseStatus};
20use crate::types::io::{
21    ApplyDone, FunctionToolCall, OutputItem, OutputMessage, OutputTextContent, ReasoningOutput, ReasoningTextContent,
22    ResponseUsage,
23};
24use crate::types::request_response::{IncompleteDetails, ResponsePayload};
25use crate::utils::common::{deserialize_from_str, deserialize_from_value_opt};
26use crate::utils::uuid7_str;
27
28/// Tracks a single output item currently being streamed, together with its
29/// accumulated text/arguments buffer.
30enum InFlight {
31    Message { item: OutputMessage, text: String },
32    Reasoning { item: ReasoningOutput, text: String },
33    FunctionCall { item: FunctionToolCall, arguments: String },
34}
35
36impl std::fmt::Debug for InFlight {
37    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
38        match self {
39            Self::Message { .. } => write!(f, "InFlight::Message {{ .. }}"),
40            Self::Reasoning { .. } => write!(f, "InFlight::Reasoning {{ .. }}"),
41            Self::FunctionCall { .. } => write!(f, "InFlight::FunctionCall {{ .. }}"),
42        }
43    }
44}
45
46impl InFlight {
47    fn finalize(self, output: &mut Vec<OutputItem>) {
48        match self {
49            Self::Reasoning { mut item, text } => {
50                if !text.is_empty() {
51                    item.content.push(ReasoningTextContent::new(text));
52                }
53                output.push(OutputItem::Reasoning(item));
54            }
55            Self::FunctionCall { mut item, arguments } => {
56                if !arguments.is_empty() && item.arguments.is_empty() {
57                    item.arguments = arguments;
58                }
59                item.status = MessageStatus::Completed;
60                output.push(OutputItem::FunctionCall(item));
61            }
62            Self::Message { mut item, text } => {
63                if !text.is_empty() {
64                    item.content.push(OutputTextContent::new(text));
65                }
66                item.status = MessageStatus::Completed;
67                output.push(OutputItem::Message(item));
68            }
69        }
70    }
71}
72
73/// Accumulates LLM response chunks from streaming or non-streaming sources.
74#[derive(Debug)]
75pub struct ResponseAccumulator {
76    response_id: String,
77    conversation_id: Option<String>,
78    output: Vec<OutputItem>,
79    usage: Option<ResponseUsage>,
80    status: ResponseStatus,
81    incomplete_details: Option<IncompleteDetails>,
82    /// In-flight output items keyed by `item_id`, in insertion order.
83    in_flight: IndexMap<String, InFlight>,
84}
85
86impl ResponseAccumulator {
87    /// Creates a new response accumulator.
88    #[must_use]
89    pub fn new(response_id: String, conversation_id: Option<String>) -> Self {
90        Self {
91            response_id,
92            conversation_id,
93            output: Vec::new(),
94            usage: None,
95            status: ResponseStatus::InProgress,
96            incomplete_details: None,
97            in_flight: IndexMap::new(),
98        }
99    }
100
101    /// Parses a non-streaming JSON response body.
102    ///
103    /// # Errors
104    /// Returns `ExecutorError::ParseError` if JSON parsing fails or required fields are missing.
105    pub fn from_json(body: &str, conversation_id: Option<&str>) -> ExecutorResult<Self> {
106        let mut json: serde_json::Value = deserialize_from_str(body).map_err(ExecutorError::JsonError)?;
107
108        let response_id = json["id"]
109            .as_str()
110            .ok_or_else(|| ExecutorError::ParseError("missing 'id' field in response".into()))?
111            .to_string();
112
113        let output = deserialize_from_value_opt::<Vec<serde_json::Value>>(json["output"].take())
114            .map(|items| {
115                let mut out = Vec::with_capacity(items.len());
116                out.extend(items.into_iter().filter_map(deserialize_from_value_opt::<OutputItem>));
117                out
118            })
119            .unwrap_or_default();
120
121        let status = json["status"]
122            .as_str()
123            .map_or(ResponseStatus::Completed, |s| s.parse().unwrap_or_default());
124
125        let usage = deserialize_from_value_opt::<ResponseUsage>(json["usage"].take());
126
127        Ok(Self {
128            response_id,
129            conversation_id: conversation_id.map(str::to_string),
130            output,
131            usage,
132            status,
133            incomplete_details: None,
134            in_flight: IndexMap::new(),
135        })
136    }
137
138    /// Accumulates an async stream of raw SSE lines with parallel processing.
139    ///
140    /// The async task feeds raw SSE lines through a channel while a `spawn_blocking`
141    /// worker handles JSON parsing on a blocking thread — keeping the tokio executor
142    /// free between chunk arrivals.
143    ///
144    /// # Errors
145    /// Returns `ExecutorError::ParseError` if chunk parsing fails, or
146    /// `ExecutorError::StreamError` if the stream or worker encounters an error.
147    pub async fn from_stream(
148        mut stream: Pin<Box<dyn Stream<Item = Result<String, ExecutorError>> + Send>>,
149        conversation_id: Option<&str>,
150    ) -> ExecutorResult<Self> {
151        let (tx, rx) = mpsc::channel::<String>();
152        // Convert to owned here — spawn_blocking closure must be 'static.
153        let conv_id_owned = conversation_id.map(str::to_string);
154
155        // Spawn blocking task: JSON parsing is CPU-bound, runs off the async executor.
156        let worker_handle = tokio::task::spawn_blocking(move || Self::process_stream_chunks(rx, conv_id_owned));
157
158        // Feed raw SSE lines from the async stream to the blocking worker.
159        while let Some(chunk_result) = stream.next().await {
160            match chunk_result {
161                Ok(chunk) => {
162                    if tx.send(chunk).is_err() {
163                        break;
164                    }
165                }
166                Err(e) => return Err(e),
167            }
168        }
169
170        // Signal EOF to worker.
171        drop(tx);
172
173        // Properly async join — does not block the tokio executor thread.
174        worker_handle
175            .await
176            .map_err(|_| ExecutorError::StreamError("Worker thread panicked".into()))
177    }
178
179    /// Worker function that processes SSE lines from the channel (runs on blocking thread).
180    fn process_stream_chunks(rx: mpsc::Receiver<String>, conversation_id: Option<String>) -> Self {
181        let mut acc = Self::new(uuid7_str("resp_"), conversation_id);
182        for line in rx {
183            acc.process_sse_line(&line);
184        }
185        acc.finish_stream();
186        acc
187    }
188
189    /// Processes pre-collected raw SSE lines synchronously.
190    ///
191    /// Useful when lines have already been buffered (e.g. replaying a recorded stream).
192    /// Prefer [`from_stream`](Self::from_stream) for live async streams.
193    /// Line parse errors are silently skipped — this function is infallible.
194    #[must_use]
195    pub fn from_sse_lines(lines: impl IntoIterator<Item = String>, conversation_id: Option<&str>) -> Self {
196        let mut acc = Self::new(uuid7_str("resp_"), conversation_id.map(str::to_string));
197        for line in lines {
198            acc.process_sse_line(&line);
199        }
200        acc.finalize_all();
201        acc
202    }
203
204    /// Finalizes all in-flight items in insertion order, pushing them to `output`.
205    pub(crate) fn finalize_all(&mut self) {
206        for (_, entry) in self.in_flight.drain(..) {
207            entry.finalize(&mut self.output);
208        }
209    }
210
211    pub(crate) fn process_sse_line(&mut self, line: &str) {
212        if let Some(frame) = normalize_sse_line(line) {
213            self.process_event(&frame);
214        }
215    }
216
217    pub(crate) fn finish_stream(&mut self) {
218        self.finalize_all();
219        if self.status == ResponseStatus::InProgress {
220            self.status = ResponseStatus::Completed;
221        }
222    }
223
224    /// Processes a typed [`EventFrame`], updating accumulator state.
225    ///
226    /// This is the core state machine — callers that already have a normalized
227    /// frame (e.g. [`StreamTee`](future)) can call this directly without
228    /// re-parsing from a raw line.
229    pub(crate) fn process_event(&mut self, frame: &EventFrame) {
230        match (&frame.event_type, &frame.payload) {
231            (SSEEventType::ResponseCreated, EventPayload::Response { id, .. }) if !id.is_empty() => {
232                self.response_id.clone_from(id);
233            }
234            (SSEEventType::OutputItemAdded, payload @ EventPayload::OutputItemAdded { item_id, item_type, .. }) => {
235                let entry = match item_type {
236                    SSEItemType::Reasoning => ReasoningOutput::try_from(payload).ok().map(|item| InFlight::Reasoning {
237                        item,
238                        text: String::with_capacity(256),
239                    }),
240                    SSEItemType::FunctionCall => {
241                        FunctionToolCall::try_from(payload)
242                            .ok()
243                            .map(|item| InFlight::FunctionCall {
244                                item,
245                                arguments: String::with_capacity(128),
246                            })
247                    }
248                    SSEItemType::Message => OutputMessage::try_from(payload).ok().map(|item| InFlight::Message {
249                        item,
250                        text: String::with_capacity(256),
251                    }),
252                };
253                if let Some(inflight) = entry {
254                    self.in_flight.insert(item_id.clone(), inflight);
255                }
256            }
257            (SSEEventType::ReasoningTextDelta, EventPayload::ReasoningDelta { delta, item_id }) => {
258                if let Some(InFlight::Reasoning { text, .. }) = self.in_flight.get_mut(item_id) {
259                    text.push_str(delta);
260                }
261            }
262            (SSEEventType::ReasoningTextDone, EventPayload::ReasoningDone { item_id, .. }) => {
263                if let Some(InFlight::Reasoning { item, text }) = self.in_flight.get_mut(item_id) {
264                    item.apply_done(&frame.payload, text);
265                }
266            }
267            (SSEEventType::FunctionCallArgumentsDelta, EventPayload::FunctionCallArgsDelta { delta, item_id, .. }) => {
268                if let Some(InFlight::FunctionCall { arguments, .. }) = self.in_flight.get_mut(item_id) {
269                    arguments.push_str(delta);
270                }
271            }
272            (SSEEventType::FunctionCallArgumentsDone, EventPayload::FunctionCallArgsDone { item_id, .. }) => {
273                if let Some(InFlight::FunctionCall { item, arguments }) = self.in_flight.get_mut(item_id) {
274                    item.apply_done(&frame.payload, arguments);
275                }
276            }
277            (SSEEventType::OutputTextDelta, EventPayload::TextDelta { delta, item_id, .. }) => {
278                if let Some(InFlight::Message { text, .. }) = self.in_flight.get_mut(item_id) {
279                    text.push_str(delta);
280                }
281            }
282            (SSEEventType::ResponseCompleted, EventPayload::Response { usage, .. }) => {
283                self.finalize_all();
284                self.status = ResponseStatus::Completed;
285                self.usage = *usage;
286            }
287            (SSEEventType::ResponseFailed, EventPayload::Response { usage, .. }) => {
288                self.finalize_all();
289                self.status = ResponseStatus::Error;
290                self.usage = *usage;
291            }
292            (SSEEventType::ResponseIncomplete, EventPayload::Response { usage, .. }) => {
293                self.finalize_all();
294                self.status = ResponseStatus::Incomplete;
295                self.usage = *usage;
296            }
297            _ => {}
298        }
299    }
300
301    /// Marks the response as incomplete due to an error or interruption.
302    pub fn mark_incomplete(&mut self, reason: impl Into<String>) {
303        self.status = ResponseStatus::Incomplete;
304        self.incomplete_details = Some(IncompleteDetails {
305            reason: Some(reason.into()),
306        });
307    }
308
309    /// Finalizes the accumulator into a `ResponsePayload`.
310    ///
311    /// The caller supplies fields that come from the original request, not from
312    /// the LLM response stream.
313    #[must_use]
314    pub fn finalize(
315        self,
316        model: &str,
317        previous_response_id: Option<&str>,
318        instructions: Option<&str>,
319    ) -> ResponsePayload {
320        ResponsePayload {
321            id: self.response_id,
322            object: "response".to_string(),
323            created_at: chrono::Utc::now().timestamp(),
324            model: model.to_string(),
325            status: self.status.as_str().to_string(),
326            output: self.output,
327            usage: self.usage,
328            incomplete_details: self.incomplete_details,
329            error: None,
330            previous_response_id: previous_response_id.map(str::to_string),
331            conversation_id: self.conversation_id,
332            instructions: instructions.map(str::to_string),
333        }
334    }
335}
336
337#[cfg(test)]
338mod tests {
339    use super::*;
340
341    #[test]
342    fn test_accumulator_new() {
343        let acc = ResponseAccumulator::new("resp_123".into(), Some("conv_456".into()));
344        assert_eq!(acc.response_id, "resp_123");
345        assert_eq!(acc.conversation_id, Some("conv_456".into()));
346        assert_eq!(acc.status, ResponseStatus::InProgress);
347    }
348
349    #[test]
350    fn test_accumulator_mark_incomplete() {
351        let mut acc = ResponseAccumulator::new("resp_123".into(), None);
352        acc.mark_incomplete("Stream interrupted");
353        assert_eq!(acc.status, ResponseStatus::Incomplete);
354        assert!(acc.incomplete_details.is_some());
355    }
356
357    #[test]
358    fn test_accumulator_finalize() {
359        let acc = ResponseAccumulator::new("resp_123".into(), Some("conv_456".into()));
360        let payload = acc.finalize("gpt-4o", Some("resp_prev"), Some("be helpful"));
361        assert_eq!(payload.id, "resp_123");
362        assert_eq!(payload.model, "gpt-4o");
363        assert_eq!(payload.conversation_id, Some("conv_456".into()));
364        assert_eq!(payload.previous_response_id, Some("resp_prev".into()));
365        assert_eq!(payload.instructions, Some("be helpful".into()));
366        assert_eq!(payload.status, ResponseStatus::InProgress.as_str());
367    }
368
369    #[test]
370    fn test_accumulator_from_sse_lines_empty() {
371        let acc = ResponseAccumulator::from_sse_lines(vec![], None);
372        assert_eq!(acc.status, ResponseStatus::InProgress);
373        assert!(acc.output.is_empty());
374    }
375
376    #[test]
377    fn test_accumulator_text_delta_assigned_to_message() {
378        let lines = vec![
379            r#"data: {"type":"response.created","response":{"id":"resp_abc"}}"#.to_string(),
380            r#"data: {"type":"response.output_item.added","item":{"id":"msg_1"}}"#.to_string(),
381            r#"data: {"type":"response.output_text.delta","delta":"Hello","item_id":"msg_1"}"#.to_string(),
382            r#"data: {"type":"response.output_text.delta","delta":" world","item_id":"msg_1"}"#.to_string(),
383            r#"data: {"type":"response.done","response":{"usage":{"input_tokens":5,"output_tokens":2,"total_tokens":7}}}"#.to_string(),
384        ];
385
386        let acc = ResponseAccumulator::from_sse_lines(lines, None);
387        assert_eq!(acc.status, ResponseStatus::Completed);
388        assert_eq!(acc.output.len(), 1);
389
390        if let OutputItem::Message(msg) = &acc.output[0] {
391            assert_eq!(msg.content.len(), 1);
392            assert_eq!(msg.content[0].text, "Hello world");
393        } else {
394            panic!("expected OutputItem::Message");
395        }
396
397        assert!(acc.usage.is_some());
398        let usage = acc.usage.unwrap();
399        assert_eq!(usage.total_tokens, 7);
400    }
401
402    #[test]
403    fn test_message_status_enum() {
404        assert_eq!(MessageStatus::Completed.as_str(), "completed");
405        assert_eq!(MessageStatus::InProgress.as_str(), "in_progress");
406    }
407
408    #[test]
409    fn test_process_event_response_created_sets_id() {
410        let mut acc = ResponseAccumulator::new("resp_old".into(), None);
411        let frame = EventFrame {
412            event_type: SSEEventType::ResponseCreated,
413            payload: EventPayload::Response {
414                id: "resp_new".into(),
415                status: "in_progress".into(),
416                usage: None,
417            },
418            sequence_number: Some(0),
419        };
420        acc.process_event(&frame);
421        assert_eq!(acc.response_id, "resp_new");
422    }
423
424    #[test]
425    fn test_process_event_response_created_empty_id_no_overwrite() {
426        let mut acc = ResponseAccumulator::new("resp_keep".into(), None);
427        let frame = EventFrame {
428            event_type: SSEEventType::ResponseCreated,
429            payload: EventPayload::Response {
430                id: String::new(),
431                status: "in_progress".into(),
432                usage: None,
433            },
434            sequence_number: Some(0),
435        };
436        acc.process_event(&frame);
437        assert_eq!(acc.response_id, "resp_keep");
438    }
439
440    #[test]
441    fn test_process_event_text_delta_accumulates() {
442        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
443
444        acc.process_event(&EventFrame {
445            event_type: SSEEventType::OutputItemAdded,
446            payload: EventPayload::OutputItemAdded {
447                item_id: "msg_1".into(),
448                item_type: "message".into(),
449                output_index: 0,
450                name: None,
451                namespace: None,
452                call_id: None,
453            },
454            sequence_number: Some(1),
455        });
456
457        acc.process_event(&EventFrame {
458            event_type: SSEEventType::OutputTextDelta,
459            payload: EventPayload::TextDelta {
460                delta: "Hello".into(),
461                item_id: "msg_1".into(),
462                output_index: 0,
463                content_index: 0,
464            },
465            sequence_number: Some(2),
466        });
467        acc.process_event(&EventFrame {
468            event_type: SSEEventType::OutputTextDelta,
469            payload: EventPayload::TextDelta {
470                delta: " world".into(),
471                item_id: "msg_1".into(),
472                output_index: 0,
473                content_index: 0,
474            },
475            sequence_number: Some(3),
476        });
477
478        acc.process_event(&EventFrame {
479            event_type: SSEEventType::ResponseCompleted,
480            payload: EventPayload::Response {
481                id: "resp_1".into(),
482                status: "completed".into(),
483                usage: None,
484            },
485            sequence_number: Some(4),
486        });
487
488        assert_eq!(acc.status, ResponseStatus::Completed);
489        assert_eq!(acc.output.len(), 1);
490        if let OutputItem::Message(msg) = &acc.output[0] {
491            assert_eq!(msg.content[0].text, "Hello world");
492        } else {
493            panic!("expected Message");
494        }
495    }
496
497    #[test]
498    fn test_process_event_completed_with_usage() {
499        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
500        let frame = EventFrame {
501            event_type: SSEEventType::ResponseCompleted,
502            payload: EventPayload::Response {
503                id: "resp_1".into(),
504                status: "completed".into(),
505                usage: Some(ResponseUsage {
506                    input_tokens: 10,
507                    output_tokens: 5,
508                    total_tokens: 15,
509                    ..Default::default()
510                }),
511            },
512            sequence_number: Some(9),
513        };
514        acc.process_event(&frame);
515        assert_eq!(acc.status, ResponseStatus::Completed);
516        assert!(acc.usage.is_some());
517        assert_eq!(acc.usage.unwrap().total_tokens, 15);
518    }
519
520    #[test]
521    fn test_process_event_failed_sets_error_status() {
522        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
523        acc.process_event(&EventFrame {
524            event_type: SSEEventType::ResponseFailed,
525            payload: EventPayload::Response {
526                id: "resp_1".into(),
527                status: "failed".into(),
528                usage: None,
529            },
530            sequence_number: Some(4),
531        });
532        assert_eq!(acc.status, ResponseStatus::Error);
533    }
534
535    #[test]
536    fn test_process_event_incomplete_sets_incomplete_status() {
537        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
538        acc.process_event(&EventFrame {
539            event_type: SSEEventType::ResponseIncomplete,
540            payload: EventPayload::Response {
541                id: "resp_1".into(),
542                status: "incomplete".into(),
543                usage: None,
544            },
545            sequence_number: Some(4),
546        });
547        assert_eq!(acc.status, ResponseStatus::Incomplete);
548    }
549
550    #[test]
551    fn test_process_event_unknown_payload_ignored() {
552        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
553        let frame = EventFrame {
554            event_type: SSEEventType::ContentPartAdded,
555            payload: EventPayload::Raw(serde_json::json!({"type": "response.content_part.added"})),
556            sequence_number: Some(3),
557        };
558        acc.process_event(&frame);
559        assert_eq!(acc.response_id, "resp_1");
560        assert_eq!(acc.status, ResponseStatus::InProgress);
561        assert!(acc.output.is_empty());
562    }
563
564    #[test]
565    fn test_accumulator_reasoning_and_message_from_sse() {
566        let lines = vec![
567            r#"data: {"type":"response.created","response":{"id":"resp_abc"}}"#.to_string(),
568            r#"data: {"type":"response.output_item.added","item":{"id":"rs_1","type":"reasoning","summary":[]}}"#.to_string(),
569            r#"data: {"type":"response.reasoning_text.delta","delta":"Let me ","item_id":"rs_1"}"#.to_string(),
570            r#"data: {"type":"response.reasoning_text.delta","delta":"think.","item_id":"rs_1"}"#.to_string(),
571            r#"data: {"type":"response.reasoning_text.done","text":"Let me think.","item_id":"rs_1"}"#.to_string(),
572            r#"data: {"type":"response.output_item.added","item":{"id":"msg_1","type":"message"}}"#.to_string(),
573            r#"data: {"type":"response.output_text.delta","delta":"Hello","item_id":"msg_1"}"#.to_string(),
574            r#"data: {"type":"response.done","response":{"usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}"#.to_string(),
575        ];
576
577        let acc = ResponseAccumulator::from_sse_lines(lines, None);
578        assert_eq!(acc.status, ResponseStatus::Completed);
579        assert_eq!(acc.output.len(), 2);
580
581        if let OutputItem::Reasoning(r) = &acc.output[0] {
582            assert_eq!(r.id, "rs_1");
583            assert_eq!(r.content.len(), 1);
584            assert_eq!(r.content[0].text, "Let me think.");
585        } else {
586            panic!("expected OutputItem::Reasoning, got {:?}", acc.output[0]);
587        }
588
589        if let OutputItem::Message(msg) = &acc.output[1] {
590            assert_eq!(msg.id, "msg_1");
591            assert_eq!(msg.content[0].text, "Hello");
592        } else {
593            panic!("expected OutputItem::Message");
594        }
595    }
596
597    #[test]
598    fn test_accumulator_message_then_reasoning_preserves_order() {
599        let lines = vec![
600            r#"data: {"type":"response.created","response":{"id":"resp_abc"}}"#.to_string(),
601            r#"data: {"type":"response.output_item.added","item":{"id":"msg_1","type":"message"}}"#.to_string(),
602            r#"data: {"type":"response.output_text.delta","delta":"Hello","item_id":"msg_1"}"#.to_string(),
603            r#"data: {"type":"response.output_item.added","item":{"id":"rs_1","type":"reasoning","summary":[]}}"#.to_string(),
604            r#"data: {"type":"response.reasoning_text.done","text":"thinking...","item_id":"rs_1"}"#.to_string(),
605            r#"data: {"type":"response.done","response":{"usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}"#.to_string(),
606        ];
607
608        let acc = ResponseAccumulator::from_sse_lines(lines, None);
609        assert_eq!(acc.output.len(), 2);
610        assert!(matches!(acc.output[0], OutputItem::Message(_)));
611        assert!(matches!(acc.output[1], OutputItem::Reasoning(_)));
612    }
613
614    #[test]
615    fn test_accumulator_reasoning_done_without_delta_uses_text() {
616        let lines = vec![
617            r#"data: {"type":"response.output_item.added","item":{"id":"rs_1","type":"reasoning","summary":[]}}"#.to_string(),
618            r#"data: {"type":"response.reasoning_text.done","text":"done only","item_id":"rs_1"}"#.to_string(),
619            r#"data: {"type":"response.done","response":{"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}"#.to_string(),
620        ];
621
622        let acc = ResponseAccumulator::from_sse_lines(lines, None);
623        if let OutputItem::Reasoning(reasoning) = &acc.output[0] {
624            assert_eq!(reasoning.content.len(), 1);
625            assert_eq!(reasoning.content[0].text, "done only");
626        } else {
627            panic!("expected reasoning output");
628        }
629    }
630
631    #[test]
632    fn test_accumulator_reasoning_from_json() {
633        let body = serde_json::json!({
634            "id": "resp_xyz",
635            "status": "completed",
636            "output": [
637                {
638                    "id": "rs_1",
639                    "type": "reasoning",
640                    "summary": [],
641                    "content": [{"text": "thinking...", "type": "reasoning_text"}],
642                    "encrypted_content": null,
643                    "status": null
644                },
645                {
646                    "id": "msg_1",
647                    "type": "message",
648                    "role": "assistant",
649                    "status": "completed",
650                    "content": [{"type": "output_text", "text": "answer", "annotations": []}]
651                }
652            ],
653            "usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15}
654        });
655
656        let acc = ResponseAccumulator::from_json(&body.to_string(), None).unwrap();
657        assert_eq!(acc.output.len(), 2);
658        assert!(matches!(acc.output[0], OutputItem::Reasoning(_)));
659        assert!(matches!(acc.output[1], OutputItem::Message(_)));
660    }
661
662    #[test]
663    fn test_function_call_accumulation_basic() {
664        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
665
666        acc.process_event(&EventFrame {
667            event_type: SSEEventType::OutputItemAdded,
668            payload: EventPayload::OutputItemAdded {
669                item_id: "fc_1".into(),
670                item_type: "function_call".into(),
671                output_index: 0,
672                name: Some("get_weather".into()),
673                namespace: Some("mcp__weather".into()),
674                call_id: Some("call_abc".into()),
675            },
676            sequence_number: Some(1),
677        });
678
679        acc.process_event(&EventFrame {
680            event_type: SSEEventType::FunctionCallArgumentsDelta,
681            payload: EventPayload::FunctionCallArgsDelta {
682                delta: r#"{"location""#.into(),
683                call_id: Some("call_abc".into()),
684                item_id: "fc_1".into(),
685                output_index: 0,
686            },
687            sequence_number: Some(2),
688        });
689
690        acc.process_event(&EventFrame {
691            event_type: SSEEventType::FunctionCallArgumentsDelta,
692            payload: EventPayload::FunctionCallArgsDelta {
693                delta: r#":"Paris"}"#.into(),
694                call_id: Some("call_abc".into()),
695                item_id: "fc_1".into(),
696                output_index: 0,
697            },
698            sequence_number: Some(3),
699        });
700
701        acc.process_event(&EventFrame {
702            event_type: SSEEventType::FunctionCallArgumentsDone,
703            payload: EventPayload::FunctionCallArgsDone {
704                arguments: r#"{"location":"Paris"}"#.into(),
705                call_id: Some("call_abc".into()),
706                item_id: "fc_1".into(),
707                name: "get_weather".into(),
708                output_index: 0,
709            },
710            sequence_number: Some(4),
711        });
712
713        acc.process_event(&EventFrame {
714            event_type: SSEEventType::ResponseCompleted,
715            payload: EventPayload::Response {
716                id: "resp_1".into(),
717                status: "completed".into(),
718                usage: None,
719            },
720            sequence_number: Some(5),
721        });
722
723        assert_eq!(acc.status, ResponseStatus::Completed);
724        assert_eq!(acc.output.len(), 1);
725        if let OutputItem::FunctionCall(fc) = &acc.output[0] {
726            assert_eq!(fc.id, "fc_1");
727            assert_eq!(fc.call_id, "call_abc");
728            assert_eq!(fc.name, "get_weather");
729            assert_eq!(fc.namespace.as_deref(), Some("mcp__weather"));
730            assert_eq!(fc.arguments, r#"{"location":"Paris"}"#);
731            assert_eq!(fc.status, MessageStatus::Completed);
732        } else {
733            panic!("expected FunctionCall");
734        }
735    }
736
737    #[test]
738    fn test_function_call_done_uses_deltas_when_arguments_empty() {
739        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
740
741        acc.process_event(&EventFrame {
742            event_type: SSEEventType::OutputItemAdded,
743            payload: EventPayload::OutputItemAdded {
744                item_id: "fc_1".into(),
745                item_type: "function_call".into(),
746                output_index: 0,
747                name: Some("search".into()),
748                namespace: None,
749                call_id: Some("call_1".into()),
750            },
751            sequence_number: Some(1),
752        });
753
754        acc.process_event(&EventFrame {
755            event_type: SSEEventType::FunctionCallArgumentsDelta,
756            payload: EventPayload::FunctionCallArgsDelta {
757                delta: r#"{"q":"rust"}"#.into(),
758                call_id: Some("call_1".into()),
759                item_id: "fc_1".into(),
760                output_index: 0,
761            },
762            sequence_number: Some(2),
763        });
764
765        acc.process_event(&EventFrame {
766            event_type: SSEEventType::FunctionCallArgumentsDone,
767            payload: EventPayload::FunctionCallArgsDone {
768                arguments: String::new(),
769                call_id: Some("call_1".into()),
770                item_id: "fc_1".into(),
771                name: "search".into(),
772                output_index: 0,
773            },
774            sequence_number: Some(3),
775        });
776
777        acc.finalize_all();
778        assert_eq!(acc.output.len(), 1);
779        if let OutputItem::FunctionCall(fc) = &acc.output[0] {
780            assert_eq!(fc.arguments, r#"{"q":"rust"}"#);
781        } else {
782            panic!("expected FunctionCall");
783        }
784    }
785
786    #[test]
787    fn test_function_call_multiple_parallel() {
788        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
789
790        acc.process_event(&EventFrame {
791            event_type: SSEEventType::OutputItemAdded,
792            payload: EventPayload::OutputItemAdded {
793                item_id: "fc_1".into(),
794                item_type: "function_call".into(),
795                output_index: 0,
796                name: Some("get_weather".into()),
797                namespace: None,
798                call_id: Some("call_1".into()),
799            },
800            sequence_number: Some(1),
801        });
802        acc.process_event(&EventFrame {
803            event_type: SSEEventType::FunctionCallArgumentsDone,
804            payload: EventPayload::FunctionCallArgsDone {
805                arguments: r#"{"city":"NYC"}"#.into(),
806                call_id: Some("call_1".into()),
807                item_id: "fc_1".into(),
808                name: "get_weather".into(),
809                output_index: 0,
810            },
811            sequence_number: Some(2),
812        });
813
814        acc.process_event(&EventFrame {
815            event_type: SSEEventType::OutputItemAdded,
816            payload: EventPayload::OutputItemAdded {
817                item_id: "fc_2".into(),
818                item_type: "function_call".into(),
819                output_index: 1,
820                name: Some("get_time".into()),
821                namespace: None,
822                call_id: Some("call_2".into()),
823            },
824            sequence_number: Some(3),
825        });
826        acc.process_event(&EventFrame {
827            event_type: SSEEventType::FunctionCallArgumentsDone,
828            payload: EventPayload::FunctionCallArgsDone {
829                arguments: r#"{"tz":"EST"}"#.into(),
830                call_id: Some("call_2".into()),
831                item_id: "fc_2".into(),
832                name: "get_time".into(),
833                output_index: 1,
834            },
835            sequence_number: Some(4),
836        });
837
838        acc.process_event(&EventFrame {
839            event_type: SSEEventType::ResponseCompleted,
840            payload: EventPayload::Response {
841                id: "resp_1".into(),
842                status: "completed".into(),
843                usage: None,
844            },
845            sequence_number: Some(5),
846        });
847
848        assert_eq!(acc.output.len(), 2);
849        assert!(matches!(&acc.output[0], OutputItem::FunctionCall(fc) if fc.name == "get_weather"));
850        assert!(matches!(&acc.output[1], OutputItem::FunctionCall(fc) if fc.name == "get_time"));
851    }
852
853    #[test]
854    fn test_function_call_interleaved_with_message() {
855        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
856
857        acc.process_event(&EventFrame {
858            event_type: SSEEventType::OutputItemAdded,
859            payload: EventPayload::OutputItemAdded {
860                item_id: "msg_1".into(),
861                item_type: "message".into(),
862                output_index: 0,
863                name: None,
864                namespace: None,
865                call_id: None,
866            },
867            sequence_number: Some(1),
868        });
869        acc.process_event(&EventFrame {
870            event_type: SSEEventType::OutputTextDelta,
871            payload: EventPayload::TextDelta {
872                delta: "Let me check".into(),
873                item_id: "msg_1".into(),
874                output_index: 0,
875                content_index: 0,
876            },
877            sequence_number: Some(2),
878        });
879
880        acc.process_event(&EventFrame {
881            event_type: SSEEventType::OutputItemAdded,
882            payload: EventPayload::OutputItemAdded {
883                item_id: "fc_1".into(),
884                item_type: "function_call".into(),
885                output_index: 1,
886                name: Some("lookup".into()),
887                namespace: None,
888                call_id: Some("call_x".into()),
889            },
890            sequence_number: Some(3),
891        });
892        acc.process_event(&EventFrame {
893            event_type: SSEEventType::FunctionCallArgumentsDone,
894            payload: EventPayload::FunctionCallArgsDone {
895                arguments: "{}".into(),
896                call_id: Some("call_x".into()),
897                item_id: "fc_1".into(),
898                name: "lookup".into(),
899                output_index: 1,
900            },
901            sequence_number: Some(4),
902        });
903
904        acc.process_event(&EventFrame {
905            event_type: SSEEventType::ResponseCompleted,
906            payload: EventPayload::Response {
907                id: "resp_1".into(),
908                status: "completed".into(),
909                usage: None,
910            },
911            sequence_number: Some(5),
912        });
913
914        assert_eq!(acc.output.len(), 2);
915        assert!(matches!(&acc.output[0], OutputItem::Message(m) if m.content[0].text == "Let me check"));
916        assert!(matches!(&acc.output[1], OutputItem::FunctionCall(fc) if fc.name == "lookup"));
917    }
918
919    #[test]
920    fn test_function_call_done_updates_metadata() {
921        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
922
923        acc.process_event(&EventFrame {
924            event_type: SSEEventType::OutputItemAdded,
925            payload: EventPayload::OutputItemAdded {
926                item_id: "fc_1".into(),
927                item_type: "function_call".into(),
928                output_index: 0,
929                name: Some("old_name".into()),
930                namespace: None,
931                call_id: Some("old_call".into()),
932            },
933            sequence_number: Some(1),
934        });
935
936        acc.process_event(&EventFrame {
937            event_type: SSEEventType::FunctionCallArgumentsDone,
938            payload: EventPayload::FunctionCallArgsDone {
939                arguments: "{}".into(),
940                call_id: Some("new_call".into()),
941                item_id: "fc_1".into(),
942                name: "new_name".into(),
943                output_index: 0,
944            },
945            sequence_number: Some(2),
946        });
947
948        acc.finalize_all();
949        if let OutputItem::FunctionCall(fc) = &acc.output[0] {
950            assert_eq!(fc.call_id, "new_call");
951            assert_eq!(fc.name, "new_name");
952        } else {
953            panic!("expected FunctionCall");
954        }
955    }
956
957    #[test]
958    fn test_function_call_empty_item_id_generates_uuid() {
959        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
960
961        acc.process_event(&EventFrame {
962            event_type: SSEEventType::OutputItemAdded,
963            payload: EventPayload::OutputItemAdded {
964                item_id: String::new(),
965                item_type: "function_call".into(),
966                output_index: 0,
967                name: Some("tool".into()),
968                namespace: None,
969                call_id: Some("c1".into()),
970            },
971            sequence_number: Some(1),
972        });
973
974        acc.process_event(&EventFrame {
975            event_type: SSEEventType::FunctionCallArgumentsDone,
976            payload: EventPayload::FunctionCallArgsDone {
977                arguments: "{}".into(),
978                call_id: Some("c1".into()),
979                item_id: String::new(),
980                name: "tool".into(),
981                output_index: 0,
982            },
983            sequence_number: Some(2),
984        });
985
986        acc.finalize_all();
987        if let OutputItem::FunctionCall(fc) = &acc.output[0] {
988            assert!(fc.id.starts_with("fc_"), "expected fc_ prefix, got: {}", fc.id);
989        } else {
990            panic!("expected FunctionCall");
991        }
992    }
993
994    /// Orphaned delta (no active function call for this `item_id`) is silently dropped.
995    #[test]
996    fn test_function_call_orphaned_delta_safe() {
997        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
998
999        acc.process_event(&EventFrame {
1000            event_type: SSEEventType::FunctionCallArgumentsDelta,
1001            payload: EventPayload::FunctionCallArgsDelta {
1002                delta: "orphan".into(),
1003                call_id: None,
1004                item_id: String::new(),
1005                output_index: 0,
1006            },
1007            sequence_number: Some(1),
1008        });
1009
1010        assert!(acc.output.is_empty());
1011        assert!(acc.in_flight.is_empty());
1012    }
1013
1014    #[test]
1015    fn test_function_call_finalized_on_response_completed() {
1016        let mut acc = ResponseAccumulator::new("resp_1".into(), None);
1017
1018        acc.process_event(&EventFrame {
1019            event_type: SSEEventType::OutputItemAdded,
1020            payload: EventPayload::OutputItemAdded {
1021                item_id: "fc_1".into(),
1022                item_type: "function_call".into(),
1023                output_index: 0,
1024                name: Some("partial".into()),
1025                namespace: None,
1026                call_id: Some("c1".into()),
1027            },
1028            sequence_number: Some(1),
1029        });
1030        acc.process_event(&EventFrame {
1031            event_type: SSEEventType::FunctionCallArgumentsDelta,
1032            payload: EventPayload::FunctionCallArgsDelta {
1033                delta: r#"{"x":1}"#.into(),
1034                call_id: Some("c1".into()),
1035                item_id: "fc_1".into(),
1036                output_index: 0,
1037            },
1038            sequence_number: Some(2),
1039        });
1040
1041        acc.process_event(&EventFrame {
1042            event_type: SSEEventType::ResponseCompleted,
1043            payload: EventPayload::Response {
1044                id: "resp_1".into(),
1045                status: "completed".into(),
1046                usage: None,
1047            },
1048            sequence_number: Some(3),
1049        });
1050
1051        assert_eq!(acc.output.len(), 1);
1052        if let OutputItem::FunctionCall(fc) = &acc.output[0] {
1053            assert_eq!(fc.arguments, r#"{"x":1}"#);
1054            assert_eq!(fc.status, MessageStatus::Completed);
1055        } else {
1056            panic!("expected FunctionCall");
1057        }
1058    }
1059
1060    #[test]
1061    fn test_function_call_from_sse_lines() {
1062        let lines = vec![
1063            r#"data: {"type":"response.created","response":{"id":"resp_fc"}}"#.to_string(),
1064            r#"data: {"type":"response.output_item.added","item":{"id":"fc_1","type":"function_call","name":"get_weather","call_id":"call_abc"}}"#.to_string(),
1065            r#"data: {"type":"response.function_call_arguments.delta","delta":"{\"city\":","item_id":"fc_1"}"#.to_string(),
1066            r#"data: {"type":"response.function_call_arguments.delta","delta":"\"SF\"}}","item_id":"fc_1"}"#.to_string(),
1067            r#"data: {"type":"response.function_call_arguments.done","arguments":"{\"city\":\"SF\"}","call_id":"call_abc","name":"get_weather","item_id":"fc_1"}"#.to_string(),
1068            r#"data: {"type":"response.done","response":{"id":"resp_fc","usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}"#.to_string(),
1069        ];
1070
1071        let acc = ResponseAccumulator::from_sse_lines(lines, Some("conv_1"));
1072        assert_eq!(acc.status, ResponseStatus::Completed);
1073        assert_eq!(acc.output.len(), 1);
1074
1075        if let OutputItem::FunctionCall(fc) = &acc.output[0] {
1076            assert_eq!(fc.name, "get_weather");
1077            assert_eq!(fc.arguments, r#"{"city":"SF"}"#);
1078            assert_eq!(fc.call_id, "call_abc");
1079        } else {
1080            panic!("expected FunctionCall");
1081        }
1082
1083        assert_eq!(acc.usage.unwrap().total_tokens, 15);
1084    }
1085}
agentic_core/executor/accumulator.rs

agentic_core/executor/
accumulator.rs
