agent_base/llm/
openai_responses.rs

1use async_trait::async_trait;
2use eventsource_stream::Eventsource;
3use futures_core::Stream;
4use futures_util::StreamExt;
5use reqwest::Client;
6use serde_json::{Value, json};
7use std::collections::HashMap;
8use std::pin::Pin;
9
10use super::{
11    LlmCapabilities, LlmClientConfig, ReasoningConfig, StreamChunk, StreamClient, UsageInfo,
12};
13use crate::types::{AgentError, AgentResult, ChatMessage, ImageAttachment, ResponseFormat};
14
15// ── Function-call buffer for parallel streaming ──
16
17struct FunctionCallBuffer {
18    call_id: String,
19    name: String,
20    arguments: String,
21}
22
23// ── Client ──
24
25pub struct OpenAiResponsesClient {
26    api_key: String,
27    model: String,
28    base_url: String,
29    client: Client,
30}
31
32impl OpenAiResponsesClient {
33    pub fn new(api_key: String, model: String, base_url: Option<String>) -> Self {
34        Self::new_with_config(api_key, model, base_url, LlmClientConfig::default())
35    }
36
37    pub fn new_with_config(
38        api_key: String,
39        model: String,
40        base_url: Option<String>,
41        config: LlmClientConfig,
42    ) -> Self {
43        let client = Client::builder()
44            .connect_timeout(config.connect_timeout)
45            .read_timeout(config.request_timeout)
46            .pool_max_idle_per_host(config.pool_max_idle_per_host)
47            .pool_idle_timeout(config.pool_idle_timeout)
48            .build()
49            .unwrap_or_else(|e| {
50                tracing::warn!(error = %e, "Failed to build reqwest client, falling back to default");
51                Client::new()
52            });
53        Self {
54            api_key,
55            model,
56            base_url: base_url.unwrap_or_else(|| "https://api.openai.com/v1".to_string()),
57            client,
58        }
59    }
60
61    // ── Message conversion ──
62
63    fn convert_messages(messages: &[ChatMessage]) -> Vec<Value> {
64        messages
65            .iter()
66            .map(|msg| match msg {
67                ChatMessage::System { content, .. } => {
68                    json!({"role": "developer", "content": content})
69                }
70                ChatMessage::User {
71                    content, images, ..
72                } => {
73                    if images.is_empty() {
74                        json!({"role": "user", "content": content})
75                    } else {
76                        let mut items: Vec<Value> =
77                            vec![json!({"type": "input_text", "text": content})];
78                        for img in images {
79                            items.push(Self::image_to_input(img));
80                        }
81                        json!({"role": "user", "content": items})
82                    }
83                }
84                ChatMessage::Assistant {
85                    content,
86                    reasoning_content: _,
87                    tool_calls,
88                } => {
89                    let mut content_parts: Vec<Value> = vec![];
90                    if let Some(text) = content
91                        && !text.is_empty()
92                    {
93                        content_parts.push(json!({"type": "output_text", "text": text}));
94                    }
95                    if let Some(tc) = tool_calls {
96                        for t in tc {
97                            content_parts.push(json!({
98                                "type": "function_call",
99                                "call_id": t.id,
100                                "name": t.name,
101                                "arguments": t.arguments,
102                            }));
103                        }
104                    }
105                    json!({"type": "message", "role": "assistant", "content": content_parts})
106                }
107                ChatMessage::Tool {
108                    tool_call_id,
109                    content,
110                } => {
111                    json!({
112                        "type": "function_call_output",
113                        "call_id": tool_call_id,
114                        "output": content,
115                    })
116                }
117                ChatMessage::Custom { data, .. } => data.clone(),
118            })
119            .collect()
120    }
121
122    fn image_to_input(img: &ImageAttachment) -> Value {
123        match img {
124            ImageAttachment::Url { url, detail } => {
125                let mut obj = serde_json::Map::new();
126                obj.insert("type".to_string(), json!("input_image"));
127                obj.insert("image_url".to_string(), json!(url));
128                if let Some(d) = detail {
129                    obj.insert("detail".to_string(), json!(image_detail_str(d)));
130                }
131                Value::Object(obj)
132            }
133            ImageAttachment::Base64 {
134                data,
135                media_type,
136                detail,
137            } => {
138                let mime = media_type.as_deref().unwrap_or("image/jpeg");
139                let url = format!("data:{mime};base64,{data}");
140                let mut obj = serde_json::Map::new();
141                obj.insert("type".to_string(), json!("input_image"));
142                obj.insert("image_url".to_string(), json!(url));
143                if let Some(d) = detail {
144                    obj.insert("detail".to_string(), json!(image_detail_str(d)));
145                }
146                Value::Object(obj)
147            }
148        }
149    }
150
151    // ── SSE event processing ──
152
153    /// Process one SSE event, mutating the function-call buffers.
154    /// Returns zero or more `StreamChunk`s to emit.
155    fn process_event(
156        event_type: &str,
157        data: &Value,
158        buffers: &mut HashMap<String, FunctionCallBuffer>,
159        next_index: &mut usize,
160    ) -> AgentResult<Vec<StreamChunk>> {
161        match event_type {
162            "response.output_text.delta" => {
163                let delta = data
164                    .get("delta")
165                    .and_then(Value::as_str)
166                    .unwrap_or("")
167                    .to_string();
168                Ok(vec![StreamChunk::Text(delta)])
169            }
170            "response.output_text.done" => {
171                // No-op: avoid double-emitting the full text.
172                Ok(vec![])
173            }
174            "response.output_item.added" => {
175                let item = match data.get("item") {
176                    Some(i) => i,
177                    None => return Ok(vec![]),
178                };
179                if item.get("type").and_then(Value::as_str) == Some("function_call") {
180                    let call_id = item
181                        .get("call_id")
182                        .or_else(|| item.get("id"))
183                        .and_then(Value::as_str)
184                        .unwrap_or("")
185                        .to_string();
186                    let name = item
187                        .get("name")
188                        .and_then(Value::as_str)
189                        .unwrap_or("")
190                        .to_string();
191                    buffers.insert(
192                        call_id.clone(),
193                        FunctionCallBuffer {
194                            call_id,
195                            name,
196                            arguments: String::new(),
197                        },
198                    );
199                }
200                Ok(vec![])
201            }
202            "response.function_call_arguments.delta" => {
203                let call_id = data.get("call_id").and_then(Value::as_str).unwrap_or("");
204                let delta = data.get("delta").and_then(Value::as_str).unwrap_or("");
205                if let Some(buf) = buffers.get_mut(call_id) {
206                    buf.arguments.push_str(delta);
207                }
208                Ok(vec![])
209            }
210            "response.function_call_arguments.done" => {
211                let call_id = data.get("call_id").and_then(Value::as_str).unwrap_or("");
212                let done_args = data.get("arguments").and_then(Value::as_str).unwrap_or("");
213                if let Some(buf) = buffers.remove(call_id) {
214                    let args = if done_args.is_empty() {
215                        buf.arguments
216                    } else {
217                        done_args.to_string()
218                    };
219                    let idx = *next_index;
220                    *next_index += 1;
221                    Ok(vec![emit_tool_call(idx, &buf.call_id, &buf.name, &args)])
222                } else {
223                    let idx = *next_index;
224                    *next_index += 1;
225                    Ok(vec![emit_tool_call(idx, call_id, "", done_args)])
226                }
227            }
228            "response.reasoning_summary_text.delta" => {
229                let delta = data
230                    .get("delta")
231                    .and_then(Value::as_str)
232                    .unwrap_or("")
233                    .to_string();
234                Ok(vec![StreamChunk::Thought(delta)])
235            }
236            "response.completed" => {
237                let response = data.get("response");
238                let mut chunks = Vec::new();
239                if let Some(usage) = extract_usage(response) {
240                    chunks.push(StreamChunk::Usage(usage));
241                }
242                // Check if the response was actually incomplete despite being "completed".
243                let status = response
244                    .and_then(|r| r.get("status"))
245                    .and_then(Value::as_str);
246                let finish_reason = if status == Some("incomplete") {
247                    let reason = response
248                        .and_then(|r| r.get("incomplete_details"))
249                        .and_then(|d| d.get("reason"))
250                        .and_then(Value::as_str)
251                        .unwrap_or("unknown");
252                    Some(format!("incomplete:{reason}"))
253                } else {
254                    None
255                };
256                chunks.push(StreamChunk::Stop { finish_reason });
257                Ok(chunks)
258            }
259            "response.incomplete" => {
260                let reason = data
261                    .get("response")
262                    .and_then(|r| r.get("incomplete_details"))
263                    .and_then(|d| d.get("reason"))
264                    .and_then(Value::as_str)
265                    .unwrap_or("unknown");
266                Ok(vec![StreamChunk::Stop {
267                    finish_reason: Some(format!("incomplete:{reason}")),
268                }])
269            }
270            "response.failed" => {
271                let msg = data
272                    .get("error")
273                    .and_then(|e| e.get("message"))
274                    .and_then(Value::as_str)
275                    .unwrap_or("response failed");
276                Err(AgentError::llm(format!("Responses API: {msg}")))
277            }
278            // Ignored events
279            "response.created"
280            | "response.in_progress"
281            | "response.output_item.done"
282            | "response.content_part.added"
283            | "response.content_part.done" => Ok(vec![]),
284            _ => Ok(vec![]),
285        }
286    }
287}
288
289fn image_detail_str(d: &crate::types::ImageDetail) -> &'static str {
290    match d {
291        crate::types::ImageDetail::Low => "low",
292        crate::types::ImageDetail::High => "high",
293        crate::types::ImageDetail::Auto => "auto",
294    }
295}
296
297fn extract_usage(response: Option<&Value>) -> Option<UsageInfo> {
298    let r = response?;
299    let input = r
300        .get("usage")
301        .and_then(|u| u.get("input_tokens"))
302        .and_then(Value::as_u64)
303        .map(|v| v as u32);
304    let output = r
305        .get("usage")
306        .and_then(|u| u.get("output_tokens"))
307        .and_then(Value::as_u64)
308        .map(|v| v as u32);
309    let total = r
310        .get("usage")
311        .and_then(|u| u.get("total_tokens"))
312        .and_then(Value::as_u64)
313        .map(|v| v as u32);
314    if input.is_some() || output.is_some() || total.is_some() {
315        Some(UsageInfo {
316            prompt_tokens: input,
317            completion_tokens: output,
318            total_tokens: total,
319        })
320    } else {
321        None
322    }
323}
324
325/// Re-encode a function call into the OpenAI delta shape expected by the
326/// react-loop's turn parser.
327fn emit_tool_call(index: usize, call_id: &str, name: &str, arguments: &str) -> StreamChunk {
328    StreamChunk::ToolCall(json!({
329        "delta": {
330            "tool_calls": [{
331                "index": index,
332                "id": call_id,
333                "function": {
334                    "name": name,
335                    "arguments": arguments,
336                }
337            }]
338        }
339    }))
340}
341
342#[async_trait]
343impl StreamClient for OpenAiResponsesClient {
344    async fn stream(
345        &self,
346        messages: &[ChatMessage],
347        tools: &[Value],
348        reasoning: Option<&ReasoningConfig>,
349        _response_format: Option<&ResponseFormat>,
350    ) -> AgentResult<Pin<Box<dyn Stream<Item = AgentResult<StreamChunk>> + Send>>> {
351        let url = format!("{}/responses", self.base_url);
352        let input = Self::convert_messages(messages);
353
354        let mut body = json!({
355            "model": self.model,
356            "input": input,
357            "stream": true,
358        });
359
360        if !tools.is_empty()
361            && let Some(obj) = body.as_object_mut()
362        {
363            obj.insert("tools".to_string(), json!(tools));
364        }
365
366        // Reasoning config
367        if let Some(config) = reasoning
368            && (config.enabled == Some(true) || config.budget_tokens.is_some())
369        {
370            let mut reasoning_cfg = serde_json::Map::new();
371            reasoning_cfg.insert("summary".to_string(), json!("auto"));
372            if let Some(budget) = config.budget_tokens {
373                reasoning_cfg.insert("budget_tokens".to_string(), json!(budget));
374            }
375            if let Some(obj) = body.as_object_mut() {
376                obj.insert("reasoning".to_string(), Value::Object(reasoning_cfg));
377            }
378        }
379
380        tracing::debug!(
381            model = %self.model,
382            url = %url,
383            body = %serde_json::to_string_pretty(&body).unwrap_or_default(),
384            "Responses API stream request"
385        );
386
387        let response = self
388            .client
389            .post(&url)
390            .header("Authorization", format!("Bearer {}", self.api_key))
391            .header("Content-Type", "application/json")
392            .json(&body)
393            .send()
394            .await
395            .map_err(|e| AgentError::llm(format!("HTTP request failed: {e}")))?;
396
397        if !response.status().is_success() {
398            let status = response.status();
399            let err_text = response
400                .text()
401                .await
402                .map_err(|e| AgentError::llm(format!("Failed to read error response: {e}")))?;
403            tracing::warn!(%status, error = %err_text, "Responses API stream non-success");
404            return Err(AgentError::LlmApi { message: err_text });
405        }
406
407        // Stateful stream processing: function_call arguments arrive across
408        // multiple SSE events keyed by call_id.  We use `scan` to maintain
409        // a buffer HashMap + index counter, then `flat_map` to flatten the
410        // per-event Vec<StreamChunk> into the output stream.
411        let buffers: HashMap<String, FunctionCallBuffer> = HashMap::new();
412        let next_index: usize = 0;
413
414        let stream = response
415            .bytes_stream()
416            .eventsource()
417            .scan((buffers, next_index), |(buffers, next_index), event| {
418                let result = match event {
419                    Ok(ev) => {
420                        let data: Result<Value, _> = serde_json::from_str(&ev.data);
421                        match data {
422                            Ok(val) => {
423                                Self::process_event(ev.event.as_str(), &val, buffers, next_index)
424                            }
425                            Err(e) => Err(AgentError::json(format!("Responses API SSE JSON: {e}"))),
426                        }
427                    }
428                    Err(e) => Err(AgentError::LlmStream(format!("SSE Stream error: {e}"))),
429                };
430                // `scan` expects `Option<T>` — always continue (errors are
431                // propagated as `Err` chunks, not by terminating the stream).
432                std::future::ready(Some(result))
433            })
434            .flat_map(|result| match result {
435                Ok(chunks) => {
436                    futures_util::stream::iter(chunks.into_iter().map(Ok).collect::<Vec<_>>())
437                }
438                Err(e) => futures_util::stream::iter(vec![Err(e)]),
439            });
440
441        Ok(Box::pin(stream))
442    }
443
444    fn capabilities(&self) -> LlmCapabilities {
445        LlmCapabilities {
446            supports_streaming: true,
447            supports_tools: true,
448            supports_vision: true,
449            supports_thinking: true,
450            max_context_tokens: Some(128_000),
451            max_output_tokens: Some(16_384),
452        }
453    }
454
455    fn model_name(&self) -> &str {
456        &self.model
457    }
458}
459
460#[cfg(test)]
461mod tests {
462    use super::*;
463    use futures_util::TryStreamExt;
464    use wiremock::matchers::{method, path};
465    use wiremock::{Mock, MockServer, ResponseTemplate};
466
467    fn client(model: &str) -> OpenAiResponsesClient {
468        OpenAiResponsesClient::new("test-key".into(), model.into(), None)
469    }
470
471    // ── Message conversion ──
472
473    #[test]
474    fn convert_messages_system() {
475        let out = OpenAiResponsesClient::convert_messages(&[ChatMessage::system("be helpful")]);
476        assert_eq!(out.len(), 1);
477        assert_eq!(out[0]["role"], "developer");
478        assert_eq!(out[0]["content"], "be helpful");
479    }
480
481    #[test]
482    fn convert_messages_user_text_only() {
483        let out = OpenAiResponsesClient::convert_messages(&[ChatMessage::user("hi")]);
484        assert_eq!(out[0]["role"], "user");
485        assert_eq!(out[0]["content"], "hi");
486    }
487
488    #[test]
489    fn convert_messages_user_with_images() {
490        let out = OpenAiResponsesClient::convert_messages(&[ChatMessage::user_with_images(
491            "look",
492            vec![ImageAttachment::Url {
493                url: "http://x/a.png".into(),
494                detail: None,
495            }],
496        )]);
497        assert_eq!(out[0]["role"], "user");
498        assert_eq!(out[0]["content"][0]["type"], "input_text");
499        assert_eq!(out[0]["content"][0]["text"], "look");
500        assert_eq!(out[0]["content"][1]["type"], "input_image");
501    }
502
503    #[test]
504    fn convert_messages_assistant_text() {
505        let out = OpenAiResponsesClient::convert_messages(&[ChatMessage::assistant("hello")]);
506        assert_eq!(out[0]["type"], "message");
507        assert_eq!(out[0]["role"], "assistant");
508        assert_eq!(out[0]["content"][0]["type"], "output_text");
509        assert_eq!(out[0]["content"][0]["text"], "hello");
510    }
511
512    #[test]
513    fn convert_messages_assistant_tool_call() {
514        let out = OpenAiResponsesClient::convert_messages(&[ChatMessage::assistant_tool_call(
515            "call_1",
516            "echo",
517            "{\"x\":1}",
518        )]);
519        assert_eq!(out[0]["content"][0]["type"], "function_call");
520        assert_eq!(out[0]["content"][0]["call_id"], "call_1");
521        assert_eq!(out[0]["content"][0]["name"], "echo");
522        assert_eq!(out[0]["content"][0]["arguments"], "{\"x\":1}");
523    }
524
525    #[test]
526    fn convert_messages_tool_result() {
527        let out = OpenAiResponsesClient::convert_messages(&[ChatMessage::tool("call_1", "ok")]);
528        assert_eq!(out[0]["type"], "function_call_output");
529        assert_eq!(out[0]["call_id"], "call_1");
530        assert_eq!(out[0]["output"], "ok");
531    }
532
533    // ── SSE event processing ──
534
535    #[test]
536    fn process_event_text_delta() {
537        let mut buffers = HashMap::new();
538        let mut idx = 0;
539        let data = json!({"delta": "hello"});
540        let chunks = OpenAiResponsesClient::process_event(
541            "response.output_text.delta",
542            &data,
543            &mut buffers,
544            &mut idx,
545        )
546        .unwrap();
547        assert_eq!(chunks.len(), 1);
548        assert!(matches!(&chunks[0], StreamChunk::Text(t) if t == "hello"));
549    }
550
551    #[test]
552    fn process_event_text_done_noop() {
553        let mut buffers = HashMap::new();
554        let mut idx = 0;
555        let chunks = OpenAiResponsesClient::process_event(
556            "response.output_text.done",
557            &json!({}),
558            &mut buffers,
559            &mut idx,
560        )
561        .unwrap();
562        assert!(chunks.is_empty());
563    }
564
565    #[test]
566    fn process_event_function_call_lifecycle() {
567        let mut buffers = HashMap::new();
568        let mut idx = 0;
569
570        // Init
571        let item_data = json!({"item": {"type": "function_call", "call_id": "c1", "name": "echo"}});
572        let chunks = OpenAiResponsesClient::process_event(
573            "response.output_item.added",
574            &item_data,
575            &mut buffers,
576            &mut idx,
577        )
578        .unwrap();
579        assert!(chunks.is_empty());
580        assert!(buffers.contains_key("c1"));
581
582        // Append
583        let delta_data = json!({"call_id": "c1", "delta": "{\"x\":"});
584        OpenAiResponsesClient::process_event(
585            "response.function_call_arguments.delta",
586            &delta_data,
587            &mut buffers,
588            &mut idx,
589        )
590        .unwrap();
591        assert_eq!(buffers["c1"].arguments, "{\"x\":");
592
593        // Finalize
594        let done_data = json!({"call_id": "c1", "arguments": "{\"x\":1}"});
595        let chunks = OpenAiResponsesClient::process_event(
596            "response.function_call_arguments.done",
597            &done_data,
598            &mut buffers,
599            &mut idx,
600        )
601        .unwrap();
602        assert_eq!(chunks.len(), 1);
603        assert!(matches!(&chunks[0], StreamChunk::ToolCall(v)
604            if v["delta"]["tool_calls"][0]["function"]["name"] == "echo"
605            && v["delta"]["tool_calls"][0]["function"]["arguments"] == "{\"x\":1}"
606            && v["delta"]["tool_calls"][0]["index"] == 0
607        ));
608        assert!(buffers.is_empty());
609        assert_eq!(idx, 1);
610    }
611
612    #[test]
613    fn process_event_reasoning_summary() {
614        let mut buffers = HashMap::new();
615        let mut idx = 0;
616        let data = json!({"delta": "thinking..."});
617        let chunks = OpenAiResponsesClient::process_event(
618            "response.reasoning_summary_text.delta",
619            &data,
620            &mut buffers,
621            &mut idx,
622        )
623        .unwrap();
624        assert!(matches!(&chunks[0], StreamChunk::Thought(t) if t == "thinking..."));
625    }
626
627    #[test]
628    fn process_event_completed_with_usage() {
629        let mut buffers = HashMap::new();
630        let mut idx = 0;
631        let data = json!({"response": {"usage": {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30}}});
632        let chunks = OpenAiResponsesClient::process_event(
633            "response.completed",
634            &data,
635            &mut buffers,
636            &mut idx,
637        )
638        .unwrap();
639        assert_eq!(chunks.len(), 2);
640        assert!(matches!(&chunks[0], StreamChunk::Usage(u)
641                if u.prompt_tokens == Some(10)
642                && u.completion_tokens == Some(20)
643                && u.total_tokens == Some(30)));
644        assert!(matches!(
645            &chunks[1],
646            StreamChunk::Stop {
647                finish_reason: None
648            }
649        ));
650    }
651
652    #[test]
653    fn process_event_completed_no_usage() {
654        let mut buffers = HashMap::new();
655        let mut idx = 0;
656        let chunks = OpenAiResponsesClient::process_event(
657            "response.completed",
658            &json!({}),
659            &mut buffers,
660            &mut idx,
661        )
662        .unwrap();
663        assert_eq!(chunks.len(), 1);
664        assert!(matches!(
665            &chunks[0],
666            StreamChunk::Stop {
667                finish_reason: None
668            }
669        ));
670    }
671
672    #[test]
673    fn process_event_completed_with_incomplete_status() {
674        let mut buffers = HashMap::new();
675        let mut idx = 0;
676        let data = json!({"response": {"status": "incomplete", "incomplete_details": {"reason": "max_output_tokens"}}});
677        let chunks = OpenAiResponsesClient::process_event(
678            "response.completed",
679            &data,
680            &mut buffers,
681            &mut idx,
682        )
683        .unwrap();
684        assert!(matches!(
685            &chunks[0],
686            StreamChunk::Stop {
687                finish_reason: Some(r)
688            } if r == "incomplete:max_output_tokens"
689        ));
690    }
691
692    #[test]
693    fn process_event_incomplete() {
694        let mut buffers = HashMap::new();
695        let mut idx = 0;
696        let data = json!({"response": {"incomplete_details": {"reason": "max_output_tokens"}}});
697        let chunks = OpenAiResponsesClient::process_event(
698            "response.incomplete",
699            &data,
700            &mut buffers,
701            &mut idx,
702        )
703        .unwrap();
704        assert!(matches!(
705            &chunks[0],
706            StreamChunk::Stop {
707                finish_reason: Some(r)
708            } if r == "incomplete:max_output_tokens"
709        ));
710    }
711
712    #[test]
713    fn process_event_failed() {
714        let mut buffers = HashMap::new();
715        let mut idx = 0;
716        let data = json!({"error": {"message": "server error"}});
717        let result =
718            OpenAiResponsesClient::process_event("response.failed", &data, &mut buffers, &mut idx);
719        assert!(result.is_err());
720    }
721
722    #[test]
723    fn process_event_unknown_is_noop() {
724        let mut buffers = HashMap::new();
725        let mut idx = 0;
726        let chunks = OpenAiResponsesClient::process_event(
727            "some.unknown.event",
728            &json!({}),
729            &mut buffers,
730            &mut idx,
731        )
732        .unwrap();
733        assert!(chunks.is_empty());
734    }
735
736    // ── Parallel function calls ──
737
738    #[test]
739    fn parallel_function_calls_interleaved() {
740        let mut buffers = HashMap::new();
741        let mut idx = 0;
742
743        // Init two calls
744        OpenAiResponsesClient::process_event(
745            "response.output_item.added",
746            &json!({"item": {"type": "function_call", "call_id": "c1", "name": "a"}}),
747            &mut buffers,
748            &mut idx,
749        )
750        .unwrap();
751        OpenAiResponsesClient::process_event(
752            "response.output_item.added",
753            &json!({"item": {"type": "function_call", "call_id": "c2", "name": "b"}}),
754            &mut buffers,
755            &mut idx,
756        )
757        .unwrap();
758
759        // Interleaved deltas
760        OpenAiResponsesClient::process_event(
761            "response.function_call_arguments.delta",
762            &json!({"call_id": "c1", "delta": "{\"x"}),
763            &mut buffers,
764            &mut idx,
765        )
766        .unwrap();
767        OpenAiResponsesClient::process_event(
768            "response.function_call_arguments.delta",
769            &json!({"call_id": "c2", "delta": "{\"y"}),
770            &mut buffers,
771            &mut idx,
772        )
773        .unwrap();
774        OpenAiResponsesClient::process_event(
775            "response.function_call_arguments.delta",
776            &json!({"call_id": "c1", "delta": "\":1}"}),
777            &mut buffers,
778            &mut idx,
779        )
780        .unwrap();
781        OpenAiResponsesClient::process_event(
782            "response.function_call_arguments.delta",
783            &json!({"call_id": "c2", "delta": "\":2}"}),
784            &mut buffers,
785            &mut idx,
786        )
787        .unwrap();
788
789        // Finalize both
790        let chunks1 = OpenAiResponsesClient::process_event(
791            "response.function_call_arguments.done",
792            &json!({"call_id": "c1", "arguments": "{\"x\":1}"}),
793            &mut buffers,
794            &mut idx,
795        )
796        .unwrap();
797        let chunks2 = OpenAiResponsesClient::process_event(
798            "response.function_call_arguments.done",
799            &json!({"call_id": "c2", "arguments": "{\"y\":2}"}),
800            &mut buffers,
801            &mut idx,
802        )
803        .unwrap();
804
805        assert_eq!(chunks1.len(), 1);
806        assert_eq!(chunks2.len(), 1);
807        // Indices should be 0 and 1
808        assert!(
809            matches!(&chunks1[0], StreamChunk::ToolCall(v) if v["delta"]["tool_calls"][0]["index"] == 0)
810        );
811        assert!(
812            matches!(&chunks2[0], StreamChunk::ToolCall(v) if v["delta"]["tool_calls"][0]["index"] == 1)
813        );
814    }
815
816    // ── Capabilities and model name ──
817
818    #[test]
819    fn capabilities_and_model_name() {
820        let c = client("gpt-4o");
821        assert_eq!(c.model_name(), "gpt-4o");
822        let caps = c.capabilities();
823        assert!(caps.supports_streaming);
824        assert!(caps.supports_tools);
825        assert!(caps.supports_vision);
826        assert!(caps.supports_thinking);
827        assert_eq!(caps.max_context_tokens, Some(128_000));
828    }
829
830    // ── Mock HTTP ──
831
832    #[tokio::test]
833    async fn stream_parses_text_and_stop() {
834        let server = MockServer::start().await;
835        let sse = concat!(
836            "event: response.output_text.delta\n",
837            "data: {\"delta\":\"Hello\"}\n\n",
838            "event: response.output_text.delta\n",
839            "data: {\"delta\":\" world\"}\n\n",
840            "event: response.completed\n",
841            "data: {\"response\":{}}\n\n",
842        );
843        Mock::given(method("POST"))
844            .and(path("/responses"))
845            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
846            .mount(&server)
847            .await;
848
849        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
850        let stream = c
851            .stream(&[ChatMessage::user("hi")], &[], None, None)
852            .await
853            .unwrap();
854        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
855
856        assert!(matches!(&chunks[0], StreamChunk::Text(t) if t == "Hello"));
857        assert!(matches!(&chunks[1], StreamChunk::Text(t) if t == " world"));
858        assert!(matches!(
859            &chunks[2],
860            StreamChunk::Stop {
861                finish_reason: None
862            }
863        ));
864    }
865
866    #[tokio::test]
867    async fn stream_parses_usage_and_stop() {
868        let server = MockServer::start().await;
869        let sse = concat!(
870            "event: response.completed\n",
871            "data: {\"response\":{\"usage\":{\"input_tokens\":10,\"output_tokens\":20,\"total_tokens\":30}}}\n\n",
872        );
873        Mock::given(method("POST"))
874            .and(path("/responses"))
875            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
876            .mount(&server)
877            .await;
878
879        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
880        let stream = c
881            .stream(&[ChatMessage::user("hi")], &[], None, None)
882            .await
883            .unwrap();
884        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
885
886        assert!(matches!(&chunks[0], StreamChunk::Usage(u)
887                if u.prompt_tokens == Some(10)
888                && u.completion_tokens == Some(20)
889                && u.total_tokens == Some(30)));
890        assert!(matches!(
891            &chunks[1],
892            StreamChunk::Stop {
893                finish_reason: None
894            }
895        ));
896    }
897
898    #[tokio::test]
899    async fn stream_parses_incomplete() {
900        let server = MockServer::start().await;
901        let sse = concat!(
902            "event: response.incomplete\n",
903            "data: {\"response\":{\"incomplete_details\":{\"reason\":\"max_output_tokens\"}}}\n\n",
904        );
905        Mock::given(method("POST"))
906            .and(path("/responses"))
907            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
908            .mount(&server)
909            .await;
910
911        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
912        let stream = c
913            .stream(&[ChatMessage::user("hi")], &[], None, None)
914            .await
915            .unwrap();
916        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
917
918        assert!(matches!(
919            &chunks[0],
920            StreamChunk::Stop {
921                finish_reason: Some(r)
922            } if r == "incomplete:max_output_tokens"
923        ));
924    }
925
926    #[tokio::test]
927    async fn stream_returns_error_on_failed_event() {
928        let server = MockServer::start().await;
929        let sse = concat!(
930            "event: response.failed\n",
931            "data: {\"error\":{\"message\":\"server error\"}}\n\n",
932        );
933        Mock::given(method("POST"))
934            .and(path("/responses"))
935            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
936            .mount(&server)
937            .await;
938
939        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
940        let stream = c
941            .stream(&[ChatMessage::user("hi")], &[], None, None)
942            .await
943            .unwrap();
944        let result: Result<Vec<_>, _> = stream.try_collect().await;
945        assert!(result.is_err());
946    }
947
948    #[tokio::test]
949    async fn stream_returns_error_on_non_success() {
950        let server = MockServer::start().await;
951        Mock::given(method("POST"))
952            .and(path("/responses"))
953            .respond_with(ResponseTemplate::new(401).set_body_string("unauthorized"))
954            .mount(&server)
955            .await;
956
957        let c = OpenAiResponsesClient::new("bad".into(), "gpt-4o".into(), Some(server.uri()));
958        let result = c.stream(&[ChatMessage::user("hi")], &[], None, None).await;
959        assert!(result.is_err());
960    }
961
962    #[tokio::test]
963    async fn stream_parses_function_call() {
964        let server = MockServer::start().await;
965        let sse = concat!(
966            "event: response.output_item.added\n",
967            "data: {\"item\":{\"type\":\"function_call\",\"call_id\":\"c1\",\"name\":\"echo\"}}\n\n",
968            "event: response.function_call_arguments.delta\n",
969            "data: {\"call_id\":\"c1\",\"delta\":\"{\\\"x\\\":\"}\n\n",
970            "event: response.function_call_arguments.delta\n",
971            "data: {\"call_id\":\"c1\",\"delta\":\"1}\"}\n\n",
972            "event: response.function_call_arguments.done\n",
973            "data: {\"call_id\":\"c1\",\"arguments\":\"{\\\"x\\\":1}\"}\n\n",
974            "event: response.completed\n",
975            "data: {\"response\":{}}\n\n",
976        );
977        Mock::given(method("POST"))
978            .and(path("/responses"))
979            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
980            .mount(&server)
981            .await;
982
983        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
984        let stream = c
985            .stream(&[ChatMessage::user("hi")], &[], None, None)
986            .await
987            .unwrap();
988        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
989
990        assert!(matches!(&chunks[0], StreamChunk::ToolCall(v)
991            if v["delta"]["tool_calls"][0]["function"]["name"] == "echo"
992            && v["delta"]["tool_calls"][0]["function"]["arguments"] == "{\"x\":1}"
993        ));
994        assert!(matches!(
995            &chunks[1],
996            StreamChunk::Stop {
997                finish_reason: None
998            }
999        ));
1000    }
1001
1002    #[tokio::test]
1003    async fn stream_parses_reasoning_summary() {
1004        let server = MockServer::start().await;
1005        let sse = concat!(
1006            "event: response.reasoning_summary_text.delta\n",
1007            "data: {\"delta\":\"Let me think...\"}\n\n",
1008            "event: response.output_text.delta\n",
1009            "data: {\"delta\":\"Here is the answer\"}\n\n",
1010            "event: response.completed\n",
1011            "data: {\"response\":{}}\n\n",
1012        );
1013        Mock::given(method("POST"))
1014            .and(path("/responses"))
1015            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
1016            .mount(&server)
1017            .await;
1018
1019        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
1020        let stream = c
1021            .stream(&[ChatMessage::user("hi")], &[], None, None)
1022            .await
1023            .unwrap();
1024        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
1025
1026        assert!(matches!(&chunks[0], StreamChunk::Thought(t) if t == "Let me think..."));
1027        assert!(matches!(&chunks[1], StreamChunk::Text(t) if t == "Here is the answer"));
1028    }
1029
1030    // ── Phase 3 coverage gaps ──
1031
1032    /// Custom messages should pass through as-is to the Responses API.
1033    #[test]
1034    fn convert_messages_custom_passthrough() {
1035        let custom_data = json!({"type": "input_text", "text": "custom content"});
1036        let messages = vec![ChatMessage::Custom {
1037            role: "user".into(),
1038            data: custom_data.clone(),
1039        }];
1040        let result = OpenAiResponsesClient::convert_messages(&messages);
1041        assert_eq!(result.len(), 1);
1042        assert_eq!(result[0], custom_data);
1043    }
1044
1045    /// Tools should be included in the request body when provided.
1046    #[tokio::test]
1047    async fn stream_sends_tools_in_request_body() {
1048        let server = MockServer::start().await;
1049        let sse = concat!(
1050            "event: response.output_text.delta\n",
1051            "data: {\"delta\":\"ok\"}\n\n",
1052            "event: response.completed\n",
1053            "data: {\"response\":{}}\n\n",
1054        );
1055        Mock::given(method("POST"))
1056            .and(path("/responses"))
1057            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
1058            .expect(1)
1059            .mount(&server)
1060            .await;
1061
1062        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
1063        let tools = vec![json!({
1064            "type": "function",
1065            "name": "echo",
1066            "description": "Echo text",
1067            "parameters": {"type": "object", "properties": {}}
1068        })];
1069        let stream = c
1070            .stream(&[ChatMessage::user("hi")], &tools, None, None)
1071            .await
1072            .unwrap();
1073        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
1074        assert!(matches!(&chunks[0], StreamChunk::Text(t) if t == "ok"));
1075        // If we got here without error, tools were accepted by the request builder.
1076    }
1077
1078    /// Reasoning config should be included when enabled.
1079    #[tokio::test]
1080    async fn stream_sends_reasoning_config() {
1081        let server = MockServer::start().await;
1082        let sse = concat!(
1083            "event: response.reasoning_summary_text.delta\n",
1084            "data: {\"delta\":\"thinking\"}\n\n",
1085            "event: response.completed\n",
1086            "data: {\"response\":{}}\n\n",
1087        );
1088        Mock::given(method("POST"))
1089            .and(path("/responses"))
1090            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
1091            .expect(1)
1092            .mount(&server)
1093            .await;
1094
1095        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
1096        let reasoning = crate::ReasoningConfig {
1097            enabled: Some(true),
1098            budget_tokens: Some(4096),
1099            effort: None,
1100        };
1101        let stream = c
1102            .stream(&[ChatMessage::user("hi")], &[], Some(&reasoning), None)
1103            .await
1104            .unwrap();
1105        let chunks: Vec<StreamChunk> = stream.try_collect().await.unwrap();
1106        assert!(matches!(&chunks[0], StreamChunk::Thought(t) if t == "thinking"));
1107    }
1108
1109    /// Malformed JSON in an SSE data field should propagate as a stream error.
1110    #[tokio::test]
1111    async fn stream_errors_on_malformed_json() {
1112        let server = MockServer::start().await;
1113        let sse = concat!(
1114            "event: response.output_text.delta\n",
1115            "data: {not valid json}\n\n",
1116        );
1117        Mock::given(method("POST"))
1118            .and(path("/responses"))
1119            .respond_with(ResponseTemplate::new(200).set_body_raw(sse, "text/event-stream"))
1120            .mount(&server)
1121            .await;
1122
1123        let c = OpenAiResponsesClient::new("k".into(), "gpt-4o".into(), Some(server.uri()));
1124        let stream = c
1125            .stream(&[ChatMessage::user("hi")], &[], None, None)
1126            .await
1127            .unwrap();
1128        let result: Result<Vec<StreamChunk>, _> = stream.try_collect().await;
1129        assert!(
1130            result.is_err(),
1131            "malformed JSON should cause a stream error"
1132        );
1133        let err = result.unwrap_err().to_string();
1134        assert!(
1135            err.contains("Responses API SSE JSON"),
1136            "error should mention SSE JSON: {}",
1137            err
1138        );
1139    }
1140}
agent_base/llm/openai_responses.rs

agent_base/llm/
openai_responses.rs
