Vakyartha documentation

1//! OpenAI Responses API adapter (`POST /v1/responses`) — the native wire
2//! format for GPT-5.x-class models. Distinct from chat-completions: typed
3//! content parts, `function_call` items, and event-name-keyed SSE deltas.
4
5use futures::StreamExt;
6use serde_json::Value;
7use tokio_util::sync::CancellationToken;
8
9use crate::Provider;
10use crate::error::LlmError;
11use crate::gate::ProviderGate;
12use crate::sse::SseDecoder;
13use crate::stream::{EventStream, StreamEvent, channel};
14use crate::types::{
15 AssistantMessage, ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition, Usage,
16};
17
18pub const OPENAI_RESPONSES_DEFAULT_BASE_URL: &str = "https://api.openai.com/v1";
19
20#[derive(Debug, Clone, Default)]
21pub struct OpenAiResponsesConfig {
22 pub api_key: String,
23 pub base_url: String,
24 /// When true, and the request carries `ChatRequest::cache`, send
25 /// `prompt_cache_key` so the provider can route repeat traffic to the
26 /// same cache-warm backend.
27 pub cache_key: bool,
28 /// When true, also send OpenRouter's `session_id` alongside
29 /// `prompt_cache_key` (for the `openrouter-responses` route).
30 pub openrouter: bool,
31}
32
33#[derive(Clone)]
34pub struct OpenAiResponsesProvider {
35 http: reqwest::Client,
36 config: OpenAiResponsesConfig,
37 gate: ProviderGate,
38}
39
40impl OpenAiResponsesProvider {
41 pub fn new(config: OpenAiResponsesConfig) -> Result<Self, LlmError> {
42 let http = reqwest::Client::builder()
43 .connect_timeout(std::time::Duration::from_secs(30))
44 .build()
45 .map_err(|e| LlmError::Network(e.to_string()))?;
46 Ok(OpenAiResponsesProvider {
47 gate: ProviderGate::new(&config.base_url, &config.api_key),
48 http,
49 config,
50 })
51 }
52}
53
54/// The messages an incremental chained request must send: everything after
55/// the last assistant message. `previous_response_id` already carries the
56/// server's record of that assistant turn and everything before it, so
57/// replaying it here would duplicate history the provider already has.
58fn messages_since_last_assistant(messages: &[Message]) -> &[Message] {
59 match messages.iter().rposition(|m| m.role == Role::Assistant) {
60 Some(idx) => &messages[idx + 1..],
61 None => messages,
62 }
63}
64
65pub fn build_body(
66 config: &OpenAiResponsesConfig,
67 request: &ChatRequest,
68) -> Result<Value, LlmError> {
69 let messages: &[Message] = match &request.previous_response_id {
70 Some(_) => messages_since_last_assistant(&request.messages),
71 None => &request.messages,
72 };
73 let mut input: Vec<Value> = Vec::with_capacity(messages.len());
74 for m in messages {
75 append_input_item(&mut input, m)?;
76 }
77
78 // `previous_response_id` only resolves against a response the provider
79 // actually retained, so a request that is (or may become) a chain link
80 // must opt into `store`. A plain one-shot request with neither cache
81 // hints nor a chain to continue keeps the old `store: false` default.
82 let store = request.cache.is_some() || request.previous_response_id.is_some();
83 let mut body = serde_json::json!({
84 "model": request.model,
85 "input": input,
86 "stream": true,
87 "store": store,
88 });
89 if let Some(system) = &request.system {
90 body["instructions"] = Value::String(system.clone());
91 }
92 if let Some(previous) = &request.previous_response_id {
93 body["previous_response_id"] = Value::String(previous.clone());
94 }
95 if let Some(cache) = &request.cache {
96 if config.cache_key {
97 body["prompt_cache_key"] = serde_json::json!(cache.session_key);
98 }
99 if config.openrouter {
100 body["session_id"] = serde_json::json!(cache.session_key);
101 }
102 }
103 if !request.tools.is_empty() {
104 let tools: Vec<Value> = request
105 .tools
106 .iter()
107 .map(|t: &ToolDefinition| {
108 serde_json::json!({
109 "type": "function",
110 "name": t.name,
111 "description": t.description,
112 "parameters": t.parameters,
113 })
114 })
115 .collect();
116 body["tools"] = Value::Array(tools);
117 }
118 Ok(body)
119}
120
121fn append_input_item(out: &mut Vec<Value>, m: &Message) -> Result<(), LlmError> {
122 match m.role {
123 Role::User => {
124 let mut text = String::new();
125 let mut outputs: Vec<(String, String)> = Vec::new();
126 let mut images: Vec<&crate::types::ImageSource> = Vec::new();
127 for b in &m.content {
128 match b {
129 ContentBlock::Text { text: t } => {
130 if !text.is_empty() {
131 text.push('\n');
132 }
133 text.push_str(t);
134 }
135 ContentBlock::Image { source } => images.push(source),
136 ContentBlock::ToolResult {
137 tool_use_id,
138 content,
139 ..
140 } => outputs.push((tool_use_id.clone(), content.clone())),
141 ContentBlock::ToolUse { .. } => {
142 return Err(LlmError::InvalidRequest(
143 "tool_use blocks must appear in assistant messages".into(),
144 ));
145 }
146 ContentBlock::Thinking { .. } => {}
147 // Only the Anthropic adapter understands server-side
148 // tool search; every other adapter skips this opaque
149 // block entirely (docs/design/68 §5/§12).
150 ContentBlock::Provider { .. } => {}
151 }
152 }
153 for (call_id, output) in outputs {
154 out.push(serde_json::json!({
155 "type": "function_call_output",
156 "call_id": call_id,
157 "output": output,
158 }));
159 }
160 if !text.is_empty() || !images.is_empty() {
161 let mut content: Vec<Value> = Vec::new();
162 if !text.is_empty() {
163 content.push(serde_json::json!({"type": "input_text", "text": text}));
164 }
165 for img in images {
166 content.push(serde_json::json!({
167 "type": "input_image",
168 "image_url": format!("data:{};base64,{}", img.media_type, img.data),
169 }));
170 }
171 out.push(serde_json::json!({
172 "role": "user",
173 "content": content,
174 }));
175 }
176 }
177 Role::Assistant => {
178 let mut text = String::new();
179 for b in &m.content {
180 match b {
181 ContentBlock::Text { text: t } => {
182 if !text.is_empty() {
183 text.push('\n');
184 }
185 text.push_str(t);
186 }
187 ContentBlock::ToolUse {
188 id,
189 name,
190 input: args,
191 } => {
192 out.push(serde_json::json!({
193 "type": "function_call",
194 "call_id": id,
195 "name": name,
196 "arguments": serde_json::to_string(args)
197 .map_err(|e| LlmError::Parse(e.to_string()))?,
198 }));
199 }
200 ContentBlock::Thinking { .. }
201 | ContentBlock::ToolResult { .. }
202 | ContentBlock::Image { .. }
203 | ContentBlock::Provider { .. } => {}
204 }
205 }
206 if !text.is_empty() {
207 out.push(serde_json::json!({
208 "role": "assistant",
209 "content": [{"type": "output_text", "text": text}],
210 }));
211 }
212 }
213 }
214 Ok(())
215}
216
217fn map_status_error(status: u16, body: &str, retry_after: Option<u64>) -> LlmError {
218 let message = serde_json::from_str::<Value>(body)
219 .ok()
220 .and_then(|v| {
221 v.pointer("/error/message")
222 .and_then(|m| m.as_str().map(String::from))
223 })
224 .unwrap_or_else(|| body.chars().take(500).collect());
225 match status {
226 401 | 403 => LlmError::Auth(message),
227 400 => LlmError::classify_400(message),
228 404 | 413 | 422 => LlmError::InvalidRequest(message),
229 429 => LlmError::RateLimit {
230 message,
231 retry_after_secs: retry_after,
232 },
233 503 | 529 => LlmError::Overloaded(message),
234 _ => LlmError::Api { status, message },
235 }
236}
237
238struct Accumulator {
239 message: AssistantMessage,
240 saw_completed: bool,
241 /// call_id → position of the ToolUse block in `message.content`.
242 tool_pos: std::collections::HashMap<String, usize>,
243 raw_json: std::collections::HashMap<String, String>,
244}
245
246impl Accumulator {
247 fn new(model: &str) -> Self {
248 Accumulator {
249 message: AssistantMessage::empty(model),
250 saw_completed: false,
251 tool_pos: std::collections::HashMap::new(),
252 raw_json: std::collections::HashMap::new(),
253 }
254 }
255
256 fn convert(&mut self, data: &str) -> Result<Option<StreamEvent>, LlmError> {
257 let v: Value =
258 serde_json::from_str(data).map_err(|e| LlmError::Parse(format!("bad chunk: {e}")))?;
259 let kind = v
260 .get("type")
261 .and_then(|t| t.as_str())
262 .ok_or_else(|| LlmError::Parse("responses event missing type".into()))?
263 .to_string();
264
265 match kind.as_str() {
266 "response.output_text.delta" => {
267 let Some(text) = v.get("delta").and_then(|d| d.as_str()) else {
268 return Ok(None);
269 };
270 append_text_block(&mut self.message.content, text);
271 Ok(Some(StreamEvent::TextDelta {
272 delta: text.to_string(),
273 partial: self.message.clone(),
274 }))
275 }
276 "response.output_item.added" => {
277 let item = v.get("item").cloned().unwrap_or(Value::Null);
278 if item.get("type").and_then(|t| t.as_str()) == Some("function_call") {
279 // Any emitted function_call means the model wants tools.
280 self.message.stop_reason = StopReason::ToolUse;
281 let call_id = item
282 .get("call_id")
283 .and_then(|c| c.as_str())
284 .unwrap_or_default()
285 .to_string();
286 let name = item
287 .get("name")
288 .and_then(|n| n.as_str())
289 .unwrap_or_default()
290 .to_string();
291 let pos = self.message.content.len();
292 self.message.content.push(ContentBlock::ToolUse {
293 id: call_id.clone(),
294 name: name.clone(),
295 input: Value::Object(Default::default()),
296 });
297 if let Some(item_id) = item.get("id").and_then(|i| i.as_str()) {
298 self.tool_pos.insert(item_id.to_string(), pos);
299 }
300 if !call_id.is_empty() {
301 self.tool_pos.insert(call_id.clone(), pos);
302 }
303 return Ok(Some(StreamEvent::ToolUseStart {
304 index: pos,
305 id: call_id,
306 name,
307 partial: self.message.clone(),
308 }));
309 }
310 Ok(None)
311 }
312 "response.function_call_arguments.delta" => {
313 let Some(delta) = v.get("delta").and_then(|d| d.as_str()) else {
314 return Ok(None);
315 };
316 let call_id = v
317 .get("item_id")
318 .or_else(|| v.get("call_id"))
319 .and_then(|c| c.as_str())
320 .unwrap_or_default();
321 // The added-item's `id` is used as item_id; map through both.
322 let pos = self.resolve_tool_pos(call_id);
323 let Some(pos) = pos else {
324 return Ok(None);
325 };
326 let raw = self.raw_json.entry(call_id.to_string()).or_default();
327 raw.push_str(delta);
328 let parsed: Value =
329 serde_json::from_str(raw).unwrap_or(Value::Object(Default::default()));
330 if let ContentBlock::ToolUse { input, .. } = &mut self.message.content[pos] {
331 *input = parsed;
332 }
333 Ok(Some(StreamEvent::ToolInputDelta {
334 index: pos,
335 delta: delta.to_string(),
336 partial: self.message.clone(),
337 }))
338 }
339 "response.completed" | "response.incomplete" => {
340 if let Some(id) = v.pointer("/response/id").and_then(|i| i.as_str()) {
341 self.message.response_id = Some(id.to_string());
342 }
343 if let Some(usage) = v.pointer("/response/usage") {
344 // Same normalization as the chat adapter: Responses'
345 // `input_tokens` already includes
346 // `input_tokens_details.cached_tokens`, so the cached
347 // share must be subtracted to get the non-cached
348 // remainder every adapter reports as `input_tokens`
349 // (docs/design/68-context-engine.md §1).
350 let raw_input = usage
351 .get("input_tokens")
352 .and_then(|x| x.as_u64())
353 .unwrap_or(0);
354 let cached_tokens = usage
355 .pointer("/input_tokens_details/cached_tokens")
356 .and_then(|x| x.as_u64())
357 .unwrap_or(0);
358 self.message.usage = Usage {
359 input_tokens: raw_input.saturating_sub(cached_tokens),
360 output_tokens: usage
361 .get("output_tokens")
362 .and_then(|x| x.as_u64())
363 .unwrap_or(0),
364 cache_read_input_tokens: (cached_tokens > 0).then_some(cached_tokens),
365 cache_creation_input_tokens: None,
366 ..Default::default()
367 };
368 }
369 if kind == "response.incomplete" {
370 self.message.stop_reason = StopReason::MaxTokens;
371 }
372 self.saw_completed = true;
373 Ok(Some(StreamEvent::End {
374 message: self.message.clone(),
375 }))
376 }
377 _ => Ok(None),
378 }
379 }
380
381 fn resolve_tool_pos(&self, call_or_item_id: &str) -> Option<usize> {
382 if let Some(pos) = self.tool_pos.get(call_or_item_id) {
383 return Some(*pos);
384 }
385 // item ids look like "fc_..."; the block id stores the call_id. Fall
386 // back to the most recent tool block when only one exists.
387 if self.tool_pos.len() == 1 {
388 return self.tool_pos.values().next().copied();
389 }
390 None
391 }
392}
393
394fn append_text_block(content: &mut Vec<ContentBlock>, text: &str) {
395 if let Some(ContentBlock::Text { text: last }) = content.last_mut() {
396 last.push_str(text);
397 return;
398 }
399 content.push(ContentBlock::text(text));
400}
401
402#[async_trait::async_trait]
403impl Provider for OpenAiResponsesProvider {
404 fn name(&self) -> &str {
405 "openai-responses"
406 }
407
408 fn circuit_key(&self) -> String {
409 crate::gate::route_identity(self.name(), &self.config.base_url, &self.config.api_key)
410 }
411
412 async fn stream(
413 &self,
414 request: ChatRequest,
415 cancel: CancellationToken,
416 ) -> Result<EventStream, LlmError> {
417 let provider_permit = self.gate.acquire(&cancel).await?;
418 let url = format!("{}/responses", self.config.base_url.trim_end_matches('/'));
419 let body = build_body(&self.config, &request)?;
420 let send_fut = self
421 .http
422 .post(&url)
423 .bearer_auth(&self.config.api_key)
424 .json(&body)
425 .send();
426 let response = tokio::select! {
427 _ = cancel.cancelled() => return Err(LlmError::Aborted { partial: None }),
428 r = send_fut => match r {
429 Ok(r) => r,
430 Err(e) => return Err(LlmError::Network(e.to_string())),
431 },
432 };
433
434 let status = response.status();
435 if !status.is_success() {
436 let retry_after = response
437 .headers()
438 .get("retry-after")
439 .and_then(|value| value.to_str().ok())
440 .and_then(|value| value.parse::<u64>().ok());
441 let text = response.text().await.unwrap_or_default();
442 return Err(map_status_error(status.as_u16(), &text, retry_after));
443 }
444
445 let model = request.model.clone();
446 let (mut sink, stream_rx) = channel(256);
447 let mut byte_stream = response.bytes_stream();
448 let mut decoder = SseDecoder::new();
449 let mut acc = Accumulator::new(&model);
450
451 tokio::spawn(async move {
452 loop {
453 tokio::select! {
454 _ = cancel.cancelled() => {
455 let partial = (!acc.message.content.is_empty()).then(|| Box::new(acc.message.clone()));
456 sink.close_error(LlmError::Aborted { partial }).await;
457 return;
458 }
459 chunk = byte_stream.next() => {
460 match chunk {
461 Some(Ok(bytes)) => {
462 decoder.push(&bytes);
463 while let Some(frame) = decoder.next_frame() {
464 match acc.convert(&frame.data) {
465 Ok(Some(event)) => sink.push(event),
466 Ok(None) => {}
467 Err(e) => {
468 sink.close_error(e).await;
469 return;
470 }
471 }
472 }
473 }
474 Some(Err(e)) => {
475 sink.close_error(LlmError::Network(e.to_string())).await;
476 return;
477 }
478 None => {
479 if acc.saw_completed {
480 sink.close_message(acc.message.clone()).await;
481 } else {
482 sink.close_error(LlmError::Parse(
483 "stream closed before response.completed".into(),
484 )).await;
485 }
486 return;
487 }
488 }
489 }
490 }
491 }
492 });
493
494 Ok(stream_rx.with_guard(provider_permit))
495 }
496}
497
498#[cfg(test)]
499mod build_body_tests {
500 #![allow(clippy::unwrap_used, clippy::expect_used)]
501 use super::*;
502 use crate::types::CacheHints;
503
504 fn config() -> OpenAiResponsesConfig {
505 OpenAiResponsesConfig::default()
506 }
507
508 #[test]
509 fn cache_key_on_sends_prompt_cache_key() {
510 let mut req = ChatRequest::new("gpt-5.6");
511 req.messages = vec![Message::user_text("hi")];
512 req.cache = Some(CacheHints {
513 session_key: "sess-1".into(),
514 breakpoints: Vec::new(),
515 });
516 let cfg = OpenAiResponsesConfig {
517 cache_key: true,
518 ..config()
519 };
520 let body = build_body(&cfg, &req).unwrap();
521 assert_eq!(body["prompt_cache_key"], "sess-1");
522 assert!(body.get("session_id").is_none());
523 // Cache hints imply this response might be chained from later.
524 assert_eq!(body["store"], true);
525 }
526
527 #[test]
528 fn openrouter_flag_adds_session_id() {
529 let mut req = ChatRequest::new("gpt-5.6");
530 req.messages = vec![Message::user_text("hi")];
531 req.cache = Some(CacheHints {
532 session_key: "sess-1".into(),
533 breakpoints: Vec::new(),
534 });
535 let cfg = OpenAiResponsesConfig {
536 cache_key: true,
537 openrouter: true,
538 ..config()
539 };
540 let body = build_body(&cfg, &req).unwrap();
541 assert_eq!(body["session_id"], "sess-1");
542 }
543
544 #[test]
545 fn no_cache_hints_keeps_store_false() {
546 let mut req = ChatRequest::new("gpt-5.6");
547 req.messages = vec![Message::user_text("hi")];
548 let body = build_body(&config(), &req).unwrap();
549 assert_eq!(body["store"], false);
550 assert!(body.get("prompt_cache_key").is_none());
551 }
552
553 #[test]
554 fn previous_response_id_chains_only_messages_after_the_last_assistant_turn() {
555 let mut req = ChatRequest::new("gpt-5.6");
556 req.messages = vec![
557 Message::user_text("first"),
558 Message::assistant(vec![ContentBlock::ToolUse {
559 id: "t1".into(),
560 name: "search".into(),
561 input: serde_json::json!({}),
562 }]),
563 Message {
564 role: Role::User,
565 content: vec![ContentBlock::tool_result("t1", "result")],
566 },
567 ];
568 req.previous_response_id = Some("resp_abc".into());
569 let body = build_body(&config(), &req).unwrap();
570 assert_eq!(body["previous_response_id"], "resp_abc");
571 let input = body["input"].as_array().unwrap();
572 // Only the tool-result message (after the last assistant turn) is
573 // sent; the earlier user/assistant exchange is already server-side.
574 assert_eq!(input.len(), 1);
575 assert_eq!(input[0]["type"], "function_call_output");
576 }
577
578 #[test]
579 fn no_previous_response_id_sends_full_history() {
580 let mut req = ChatRequest::new("gpt-5.6");
581 req.messages = vec![
582 Message::user_text("first"),
583 Message::assistant(vec![ContentBlock::text("answer")]),
584 ];
585 let body = build_body(&config(), &req).unwrap();
586 assert_eq!(body["input"].as_array().unwrap().len(), 2);
587 assert!(body.get("previous_response_id").is_none());
588 }
589}
590

1//! OpenAI Responses API adapter (`POST /v1/responses`) — the native wire
2//! format for GPT-5.x-class models. Distinct from chat-completions: typed
3//! content parts, `function_call` items, and event-name-keyed SSE deltas.
4
5use futures::StreamExt;
6use serde_json::Value;
7use tokio_util::sync::CancellationToken;
8
9use crate::Provider;
10use crate::error::LlmError;
11use crate::gate::ProviderGate;
12use crate::sse::SseDecoder;
13use crate::stream::{EventStream, StreamEvent, channel};
14use crate::types::{
15 AssistantMessage, ChatRequest, ContentBlock, Message, Role, StopReason, ToolDefinition, Usage,
16};
17
18pub const OPENAI_RESPONSES_DEFAULT_BASE_URL: &str = "https://api.openai.com/v1";
19
20#[derive(Debug, Clone, Default)]
21pub struct OpenAiResponsesConfig {
22 pub api_key: String,
23 pub base_url: String,
24 /// When true, and the request carries `ChatRequest::cache`, send
25 /// `prompt_cache_key` so the provider can route repeat traffic to the
26 /// same cache-warm backend.
27 pub cache_key: bool,
28 /// When true, also send OpenRouter's `session_id` alongside
29 /// `prompt_cache_key` (for the `openrouter-responses` route).
30 pub openrouter: bool,
31}
32
33#[derive(Clone)]
34pub struct OpenAiResponsesProvider {
35 http: reqwest::Client,
36 config: OpenAiResponsesConfig,
37 gate: ProviderGate,
38}
39
40impl OpenAiResponsesProvider {
41 pub fn new(config: OpenAiResponsesConfig) -> Result<Self, LlmError> {
42 let http = reqwest::Client::builder()
43 .connect_timeout(std::time::Duration::from_secs(30))
44 .build()
45 .map_err(|e| LlmError::Network(e.to_string()))?;
46 Ok(OpenAiResponsesProvider {
47 gate: ProviderGate::new(&config.base_url, &config.api_key),
48 http,
49 config,
50 })
51 }
52}
53
54/// The messages an incremental chained request must send: everything after
55/// the last assistant message. `previous_response_id` already carries the
56/// server's record of that assistant turn and everything before it, so
57/// replaying it here would duplicate history the provider already has.
58fn messages_since_last_assistant(messages: &[Message]) -> &[Message] {
59 match messages.iter().rposition(|m| m.role == Role::Assistant) {
60 Some(idx) => &messages[idx + 1..],
61 None => messages,
62 }
63}
64
65pub fn build_body(
66 config: &OpenAiResponsesConfig,
67 request: &ChatRequest,
68) -> Result<Value, LlmError> {
69 let messages: &[Message] = match &request.previous_response_id {
70 Some(_) => messages_since_last_assistant(&request.messages),
71 None => &request.messages,
72 };
73 let mut input: Vec<Value> = Vec::with_capacity(messages.len());
74 for m in messages {
75 append_input_item(&mut input, m)?;
76 }
77
78 // `previous_response_id` only resolves against a response the provider
79 // actually retained, so a request that is (or may become) a chain link
80 // must opt into `store`. A plain one-shot request with neither cache
81 // hints nor a chain to continue keeps the old `store: false` default.
82 let store = request.cache.is_some() || request.previous_response_id.is_some();
83 let mut body = serde_json::json!({
84 "model": request.model,
85 "input": input,
86 "stream": true,
87 "store": store,
88 });
89 if let Some(system) = &request.system {
90 body["instructions"] = Value::String(system.clone());
91 }
92 if let Some(previous) = &request.previous_response_id {
93 body["previous_response_id"] = Value::String(previous.clone());
94 }
95 if let Some(cache) = &request.cache {
96 if config.cache_key {
97 body["prompt_cache_key"] = serde_json::json!(cache.session_key);
98 }
99 if config.openrouter {
100 body["session_id"] = serde_json::json!(cache.session_key);
101 }
102 }
103 if !request.tools.is_empty() {
104 let tools: Vec<Value> = request
105 .tools
106 .iter()
107 .map(|t: &ToolDefinition| {
108 serde_json::json!({
109 "type": "function",
110 "name": t.name,
111 "description": t.description,
112 "parameters": t.parameters,
113 })
114 })
115 .collect();
116 body["tools"] = Value::Array(tools);
117 }
118 Ok(body)
119}
120
121fn append_input_item(out: &mut Vec<Value>, m: &Message) -> Result<(), LlmError> {
122 match m.role {
123 Role::User => {
124 let mut text = String::new();
125 let mut outputs: Vec<(String, String)> = Vec::new();
126 let mut images: Vec<&crate::types::ImageSource> = Vec::new();
127 for b in &m.content {
128 match b {
129 ContentBlock::Text { text: t } => {
130 if !text.is_empty() {
131 text.push('\n');
132 }
133 text.push_str(t);
134 }
135 ContentBlock::Image { source } => images.push(source),
136 ContentBlock::ToolResult {
137 tool_use_id,
138 content,
139 ..
140 } => outputs.push((tool_use_id.clone(), content.clone())),
141 ContentBlock::ToolUse { .. } => {
142 return Err(LlmError::InvalidRequest(
143 "tool_use blocks must appear in assistant messages".into(),
144 ));
145 }
146 ContentBlock::Thinking { .. } => {}
147 // Only the Anthropic adapter understands server-side
148 // tool search; every other adapter skips this opaque
149 // block entirely (docs/design/68 §5/§12).
150 ContentBlock::Provider { .. } => {}
151 }
152 }
153 for (call_id, output) in outputs {
154 out.push(serde_json::json!({
155 "type": "function_call_output",
156 "call_id": call_id,
157 "output": output,
158 }));
159 }
160 if !text.is_empty() || !images.is_empty() {
161 let mut content: Vec<Value> = Vec::new();
162 if !text.is_empty() {
163 content.push(serde_json::json!({"type": "input_text", "text": text}));
164 }
165 for img in images {
166 content.push(serde_json::json!({
167 "type": "input_image",
168 "image_url": format!("data:{};base64,{}", img.media_type, img.data),
169 }));
170 }
171 out.push(serde_json::json!({
172 "role": "user",
173 "content": content,
174 }));
175 }
176 }
177 Role::Assistant => {
178 let mut text = String::new();
179 for b in &m.content {
180 match b {
181 ContentBlock::Text { text: t } => {
182 if !text.is_empty() {
183 text.push('\n');
184 }
185 text.push_str(t);
186 }
187 ContentBlock::ToolUse {
188 id,
189 name,
190 input: args,
191 } => {
192 out.push(serde_json::json!({
193 "type": "function_call",
194 "call_id": id,
195 "name": name,
196 "arguments": serde_json::to_string(args)
197 .map_err(|e| LlmError::Parse(e.to_string()))?,
198 }));
199 }
200 ContentBlock::Thinking { .. }
201 | ContentBlock::ToolResult { .. }
202 | ContentBlock::Image { .. }
203 | ContentBlock::Provider { .. } => {}
204 }
205 }
206 if !text.is_empty() {
207 out.push(serde_json::json!({
208 "role": "assistant",
209 "content": [{"type": "output_text", "text": text}],
210 }));
211 }
212 }
213 }
214 Ok(())
215}
216
217fn map_status_error(status: u16, body: &str, retry_after: Option<u64>) -> LlmError {
218 let message = serde_json::from_str::<Value>(body)
219 .ok()
220 .and_then(|v| {
221 v.pointer("/error/message")
222 .and_then(|m| m.as_str().map(String::from))
223 })
224 .unwrap_or_else(|| body.chars().take(500).collect());
225 match status {
226 401 | 403 => LlmError::Auth(message),
227 400 => LlmError::classify_400(message),
228 404 | 413 | 422 => LlmError::InvalidRequest(message),
229 429 => LlmError::RateLimit {
230 message,
231 retry_after_secs: retry_after,
232 },
233 503 | 529 => LlmError::Overloaded(message),
234 _ => LlmError::Api { status, message },
235 }
236}
237
238struct Accumulator {
239 message: AssistantMessage,
240 saw_completed: bool,
241 /// call_id → position of the ToolUse block in `message.content`.
242 tool_pos: std::collections::HashMap<String, usize>,
243 raw_json: std::collections::HashMap<String, String>,
244}
245
246impl Accumulator {
247 fn new(model: &str) -> Self {
248 Accumulator {
249 message: AssistantMessage::empty(model),
250 saw_completed: false,
251 tool_pos: std::collections::HashMap::new(),
252 raw_json: std::collections::HashMap::new(),
253 }
254 }
255
256 fn convert(&mut self, data: &str) -> Result<Option<StreamEvent>, LlmError> {
257 let v: Value =
258 serde_json::from_str(data).map_err(|e| LlmError::Parse(format!("bad chunk: {e}")))?;
259 let kind = v
260 .get("type")
261 .and_then(|t| t.as_str())
262 .ok_or_else(|| LlmError::Parse("responses event missing type".into()))?
263 .to_string();
264
265 match kind.as_str() {
266 "response.output_text.delta" => {
267 let Some(text) = v.get("delta").and_then(|d| d.as_str()) else {
268 return Ok(None);
269 };
270 append_text_block(&mut self.message.content, text);
271 Ok(Some(StreamEvent::TextDelta {
272 delta: text.to_string(),
273 partial: self.message.clone(),
274 }))
275 }
276 "response.output_item.added" => {
277 let item = v.get("item").cloned().unwrap_or(Value::Null);
278 if item.get("type").and_then(|t| t.as_str()) == Some("function_call") {
279 // Any emitted function_call means the model wants tools.
280 self.message.stop_reason = StopReason::ToolUse;
281 let call_id = item
282 .get("call_id")
283 .and_then(|c| c.as_str())
284 .unwrap_or_default()
285 .to_string();
286 let name = item
287 .get("name")
288 .and_then(|n| n.as_str())
289 .unwrap_or_default()
290 .to_string();
291 let pos = self.message.content.len();
292 self.message.content.push(ContentBlock::ToolUse {
293 id: call_id.clone(),
294 name: name.clone(),
295 input: Value::Object(Default::default()),
296 });
297 if let Some(item_id) = item.get("id").and_then(|i| i.as_str()) {
298 self.tool_pos.insert(item_id.to_string(), pos);
299 }
300 if !call_id.is_empty() {
301 self.tool_pos.insert(call_id.clone(), pos);
302 }
303 return Ok(Some(StreamEvent::ToolUseStart {
304 index: pos,
305 id: call_id,
306 name,
307 partial: self.message.clone(),
308 }));
309 }
310 Ok(None)
311 }
312 "response.function_call_arguments.delta" => {
313 let Some(delta) = v.get("delta").and_then(|d| d.as_str()) else {
314 return Ok(None);
315 };
316 let call_id = v
317 .get("item_id")
318 .or_else(|| v.get("call_id"))
319 .and_then(|c| c.as_str())
320 .unwrap_or_default();
321 // The added-item's `id` is used as item_id; map through both.
322 let pos = self.resolve_tool_pos(call_id);
323 let Some(pos) = pos else {
324 return Ok(None);
325 };
326 let raw = self.raw_json.entry(call_id.to_string()).or_default();
327 raw.push_str(delta);
328 let parsed: Value =
329 serde_json::from_str(raw).unwrap_or(Value::Object(Default::default()));
330 if let ContentBlock::ToolUse { input, .. } = &mut self.message.content[pos] {
331 *input = parsed;
332 }
333 Ok(Some(StreamEvent::ToolInputDelta {
334 index: pos,
335 delta: delta.to_string(),
336 partial: self.message.clone(),
337 }))
338 }
339 "response.completed" | "response.incomplete" => {
340 if let Some(id) = v.pointer("/response/id").and_then(|i| i.as_str()) {
341 self.message.response_id = Some(id.to_string());
342 }
343 if let Some(usage) = v.pointer("/response/usage") {
344 // Same normalization as the chat adapter: Responses'
345 // `input_tokens` already includes
346 // `input_tokens_details.cached_tokens`, so the cached
347 // share must be subtracted to get the non-cached
348 // remainder every adapter reports as `input_tokens`
349 // (docs/design/68-context-engine.md §1).
350 let raw_input = usage
351 .get("input_tokens")
352 .and_then(|x| x.as_u64())
353 .unwrap_or(0);
354 let cached_tokens = usage
355 .pointer("/input_tokens_details/cached_tokens")
356 .and_then(|x| x.as_u64())
357 .unwrap_or(0);
358 self.message.usage = Usage {
359 input_tokens: raw_input.saturating_sub(cached_tokens),
360 output_tokens: usage
361 .get("output_tokens")
362 .and_then(|x| x.as_u64())
363 .unwrap_or(0),
364 cache_read_input_tokens: (cached_tokens > 0).then_some(cached_tokens),
365 cache_creation_input_tokens: None,
366 ..Default::default()
367 };
368 }
369 if kind == "response.incomplete" {
370 self.message.stop_reason = StopReason::MaxTokens;
371 }
372 self.saw_completed = true;
373 Ok(Some(StreamEvent::End {
374 message: self.message.clone(),
375 }))
376 }
377 _ => Ok(None),
378 }
379 }
380
381 fn resolve_tool_pos(&self, call_or_item_id: &str) -> Option<usize> {
382 if let Some(pos) = self.tool_pos.get(call_or_item_id) {
383 return Some(*pos);
384 }
385 // item ids look like "fc_..."; the block id stores the call_id. Fall
386 // back to the most recent tool block when only one exists.
387 if self.tool_pos.len() == 1 {
388 return self.tool_pos.values().next().copied();
389 }
390 None
391 }
392}
393
394fn append_text_block(content: &mut Vec<ContentBlock>, text: &str) {
395 if let Some(ContentBlock::Text { text: last }) = content.last_mut() {
396 last.push_str(text);
397 return;
398 }
399 content.push(ContentBlock::text(text));
400}
401
402#[async_trait::async_trait]
403impl Provider for OpenAiResponsesProvider {
404 fn name(&self) -> &str {
405 "openai-responses"
406 }
407
408 fn circuit_key(&self) -> String {
409 crate::gate::route_identity(self.name(), &self.config.base_url, &self.config.api_key)
410 }
411
412 async fn stream(
413 &self,
414 request: ChatRequest,
415 cancel: CancellationToken,
416 ) -> Result<EventStream, LlmError> {
417 let provider_permit = self.gate.acquire(&cancel).await?;
418 let url = format!("{}/responses", self.config.base_url.trim_end_matches('/'));
419 let body = build_body(&self.config, &request)?;
420 let send_fut = self
421 .http
422 .post(&url)
423 .bearer_auth(&self.config.api_key)
424 .json(&body)
425 .send();
426 let response = tokio::select! {
427 _ = cancel.cancelled() => return Err(LlmError::Aborted { partial: None }),
428 r = send_fut => match r {
429 Ok(r) => r,
430 Err(e) => return Err(LlmError::Network(e.to_string())),
431 },
432 };
433
434 let status = response.status();
435 if !status.is_success() {
436 let retry_after = response
437 .headers()
438 .get("retry-after")
439 .and_then(|value| value.to_str().ok())
440 .and_then(|value| value.parse::<u64>().ok());
441 let text = response.text().await.unwrap_or_default();
442 return Err(map_status_error(status.as_u16(), &text, retry_after));
443 }
444
445 let model = request.model.clone();
446 let (mut sink, stream_rx) = channel(256);
447 let mut byte_stream = response.bytes_stream();
448 let mut decoder = SseDecoder::new();
449 let mut acc = Accumulator::new(&model);
450
451 tokio::spawn(async move {
452 loop {
453 tokio::select! {
454 _ = cancel.cancelled() => {
455 let partial = (!acc.message.content.is_empty()).then(|| Box::new(acc.message.clone()));
456 sink.close_error(LlmError::Aborted { partial }).await;
457 return;
458 }
459 chunk = byte_stream.next() => {
460 match chunk {
461 Some(Ok(bytes)) => {
462 decoder.push(&bytes);
463 while let Some(frame) = decoder.next_frame() {
464 match acc.convert(&frame.data) {
465 Ok(Some(event)) => sink.push(event),
466 Ok(None) => {}
467 Err(e) => {
468 sink.close_error(e).await;
469 return;
470 }
471 }
472 }
473 }
474 Some(Err(e)) => {
475 sink.close_error(LlmError::Network(e.to_string())).await;
476 return;
477 }
478 None => {
479 if acc.saw_completed {
480 sink.close_message(acc.message.clone()).await;
481 } else {
482 sink.close_error(LlmError::Parse(
483 "stream closed before response.completed".into(),
484 )).await;
485 }
486 return;
487 }
488 }
489 }
490 }
491 }
492 });
493
494 Ok(stream_rx.with_guard(provider_permit))
495 }
496}
497
498#[cfg(test)]
499mod build_body_tests {
500 #![allow(clippy::unwrap_used, clippy::expect_used)]
501 use super::*;
502 use crate::types::CacheHints;
503
504 fn config() -> OpenAiResponsesConfig {
505 OpenAiResponsesConfig::default()
506 }
507
508 #[test]
509 fn cache_key_on_sends_prompt_cache_key() {
510 let mut req = ChatRequest::new("gpt-5.6");
511 req.messages = vec![Message::user_text("hi")];
512 req.cache = Some(CacheHints {
513 session_key: "sess-1".into(),
514 breakpoints: Vec::new(),
515 });
516 let cfg = OpenAiResponsesConfig {
517 cache_key: true,
518 ..config()
519 };
520 let body = build_body(&cfg, &req).unwrap();
521 assert_eq!(body["prompt_cache_key"], "sess-1");
522 assert!(body.get("session_id").is_none());
523 // Cache hints imply this response might be chained from later.
524 assert_eq!(body["store"], true);
525 }
526
527 #[test]
528 fn openrouter_flag_adds_session_id() {
529 let mut req = ChatRequest::new("gpt-5.6");
530 req.messages = vec![Message::user_text("hi")];
531 req.cache = Some(CacheHints {
532 session_key: "sess-1".into(),
533 breakpoints: Vec::new(),
534 });
535 let cfg = OpenAiResponsesConfig {
536 cache_key: true,
537 openrouter: true,
538 ..config()
539 };
540 let body = build_body(&cfg, &req).unwrap();
541 assert_eq!(body["session_id"], "sess-1");
542 }
543
544 #[test]
545 fn no_cache_hints_keeps_store_false() {
546 let mut req = ChatRequest::new("gpt-5.6");
547 req.messages = vec![Message::user_text("hi")];
548 let body = build_body(&config(), &req).unwrap();
549 assert_eq!(body["store"], false);
550 assert!(body.get("prompt_cache_key").is_none());
551 }
552
553 #[test]
554 fn previous_response_id_chains_only_messages_after_the_last_assistant_turn() {
555 let mut req = ChatRequest::new("gpt-5.6");
556 req.messages = vec![
557 Message::user_text("first"),
558 Message::assistant(vec![ContentBlock::ToolUse {
559 id: "t1".into(),
560 name: "search".into(),
561 input: serde_json::json!({}),
562 }]),
563 Message {
564 role: Role::User,
565 content: vec![ContentBlock::tool_result("t1", "result")],
566 },
567 ];
568 req.previous_response_id = Some("resp_abc".into());
569 let body = build_body(&config(), &req).unwrap();
570 assert_eq!(body["previous_response_id"], "resp_abc");
571 let input = body["input"].as_array().unwrap();
572 // Only the tool-result message (after the last assistant turn) is
573 // sent; the earlier user/assistant exchange is already server-side.
574 assert_eq!(input.len(), 1);
575 assert_eq!(input[0]["type"], "function_call_output");
576 }
577
578 #[test]
579 fn no_previous_response_id_sends_full_history() {
580 let mut req = ChatRequest::new("gpt-5.6");
581 req.messages = vec![
582 Message::user_text("first"),
583 Message::assistant(vec![ContentBlock::text("answer")]),
584 ];
585 let body = build_body(&config(), &req).unwrap();
586 assert_eq!(body["input"].as_array().unwrap().len(), 2);
587 assert!(body.get("previous_response_id").is_none());
588 }
589}
590

Openai Responses

Indexing the workspace…

Openai Responses