1use serde_json::{Value, json};
2
3use super::{OpenAiError, OpenAiResponseMetadata, OpenAiSurface};
4
5fn upstream_invalid(message: impl Into<String>, _param: Option<impl Into<String>>) -> OpenAiError {
6 OpenAiError::upstream_protocol(message)
7}
8
9#[derive(Debug, Clone)]
10pub struct SseEvent {
11 pub event: Option<String>,
12 pub data: Value,
13}
14
15#[derive(Debug, Clone)]
16pub enum BlockKind {
17 Text {
18 text: String,
19 },
20 Thinking {
21 text: String,
22 },
23 Tool {
24 id: String,
25 name: String,
26 arguments: String,
27 },
28 HostedSearch {
29 id: String,
30 name: String,
31 arguments: String,
32 },
33 HostedResult,
34}
35
36#[derive(Debug, Clone)]
37pub struct Block {
38 pub index: usize,
39 pub kind: BlockKind,
40}
41
42#[derive(Debug, Clone, Default)]
43pub struct Usage {
44 pub input_tokens: u64,
45 pub output_tokens: u64,
46 pub cache_read_tokens: u64,
47 pub cache_creation_tokens: u64,
48}
49
50impl Usage {
51 pub fn chat_value(&self) -> Value {
52 json!({
53 "prompt_tokens": self.input_tokens + self.cache_read_tokens,
54 "completion_tokens": self.output_tokens,
55 "total_tokens": self.input_tokens + self.cache_read_tokens + self.output_tokens,
56 "prompt_tokens_details": {"cached_tokens": self.cache_read_tokens},
57 })
58 }
59
60 pub fn responses_value(&self) -> Value {
61 json!({
62 "input_tokens": self.input_tokens + self.cache_read_tokens,
63 "input_tokens_details": {"cached_tokens": self.cache_read_tokens},
64 "output_tokens": self.output_tokens,
65 "output_tokens_details": {"reasoning_tokens": 0},
66 "total_tokens": self.input_tokens + self.cache_read_tokens + self.output_tokens,
67 })
68 }
69}
70
71#[derive(Debug, Clone, Default)]
72pub struct AnthropicAccumulator {
73 pub upstream_id: Option<String>,
74 pub model: Option<String>,
75 pub blocks: Vec<Block>,
76 pub stop_reason: Option<String>,
77 pub usage: Usage,
78 pub citations: Vec<Value>,
79 pub stopped: bool,
80}
81
82impl AnthropicAccumulator {
83 pub fn apply(&mut self, event: &SseEvent) -> Result<(), OpenAiError> {
84 let kind = event
85 .data
86 .get("type")
87 .and_then(Value::as_str)
88 .or(event.event.as_deref())
89 .unwrap_or_default();
90 match kind {
91 "message_start" => {
92 let message = event.data.get("message").ok_or_else(|| {
93 upstream_invalid(
94 "Provider message_start is missing 'message'",
95 None::<String>,
96 )
97 })?;
98 self.upstream_id = message
99 .get("id")
100 .and_then(Value::as_str)
101 .map(str::to_string);
102 self.model = message
103 .get("model")
104 .and_then(Value::as_str)
105 .map(str::to_string);
106 self.update_usage(message.get("usage"));
107 }
108 "content_block_start" => {
109 let index = required_index(&event.data)?;
110 let block = event.data.get("content_block").ok_or_else(|| {
111 upstream_invalid("Provider content block is missing", None::<String>)
112 })?;
113 let kind = match block.get("type").and_then(Value::as_str) {
114 Some("text") => BlockKind::Text {
115 text: block
116 .get("text")
117 .and_then(Value::as_str)
118 .unwrap_or_default()
119 .to_string(),
120 },
121 Some("thinking") => BlockKind::Thinking {
122 text: block
123 .get("thinking")
124 .and_then(Value::as_str)
125 .unwrap_or_default()
126 .to_string(),
127 },
128 Some("tool_use") => BlockKind::Tool {
129 id: required_block_string(block, "id")?,
130 name: required_block_string(block, "name")?,
131 arguments: block
132 .get("input")
133 .filter(|value| !value.is_null())
134 .map(Value::to_string)
135 .filter(|value| value != "{}")
136 .unwrap_or_default(),
137 },
138 Some("server_tool_use") => BlockKind::HostedSearch {
139 id: required_block_string(block, "id")?,
140 name: required_block_string(block, "name")?,
141 arguments: String::new(),
142 },
143 Some(kind) if kind.ends_with("_tool_result") => BlockKind::HostedResult,
144 Some(other) => {
145 return Err(upstream_invalid(
146 format!("Unsupported provider output block '{other}'"),
147 None::<String>,
148 ));
149 }
150 None => {
151 return Err(upstream_invalid(
152 "Provider output block has no type",
153 None::<String>,
154 ));
155 }
156 };
157 self.blocks.push(Block { index, kind });
158 }
159 "content_block_delta" => {
160 let index = required_index(&event.data)?;
161 let delta = event.data.get("delta").ok_or_else(|| {
162 upstream_invalid("Provider content delta is missing", None::<String>)
163 })?;
164 let block = self
165 .blocks
166 .iter_mut()
167 .rev()
168 .find(|block| block.index == index)
169 .ok_or_else(|| {
170 upstream_invalid(
171 "Provider delta references an unknown block",
172 None::<String>,
173 )
174 })?;
175 match (delta.get("type").and_then(Value::as_str), &mut block.kind) {
176 (Some("text_delta"), BlockKind::Text { text }) => {
177 text.push_str(
178 delta
179 .get("text")
180 .and_then(Value::as_str)
181 .unwrap_or_default(),
182 );
183 }
184 (Some("thinking_delta"), BlockKind::Thinking { text }) => {
185 text.push_str(
186 delta
187 .get("thinking")
188 .and_then(Value::as_str)
189 .unwrap_or_default(),
190 );
191 }
192 (Some("signature_delta"), BlockKind::Thinking { .. }) => {}
193 (
194 Some("input_json_delta"),
195 BlockKind::Tool { arguments, .. }
196 | BlockKind::HostedSearch { arguments, .. },
197 ) => {
198 arguments.push_str(
199 delta
200 .get("partial_json")
201 .and_then(Value::as_str)
202 .unwrap_or_default(),
203 );
204 }
205 (Some("citations_delta"), BlockKind::Text { .. }) => {
206 if let Some(citation) = delta.get("citation") {
207 self.citations.push(citation.clone());
208 }
209 }
210 _ => {
211 return Err(upstream_invalid(
212 "Provider delta does not match its content block",
213 None::<String>,
214 ));
215 }
216 }
217 }
218 "content_block_stop" => {
219 let index = required_index(&event.data)?;
220 if let Some(arguments) = self.blocks.iter().find_map(|block| {
221 if block.index != index {
222 return None;
223 }
224 match &block.kind {
225 BlockKind::Tool { arguments, .. }
226 | BlockKind::HostedSearch { arguments, .. } => Some(arguments),
227 _ => None,
228 }
229 }) && !arguments.is_empty()
230 && serde_json::from_str::<Value>(arguments).is_err()
231 {
232 return Err(OpenAiError::upstream_protocol(
233 "Provider emitted malformed tool arguments",
234 ));
235 }
236 }
237 "message_delta" => {
238 self.stop_reason = event
239 .data
240 .pointer("/delta/stop_reason")
241 .and_then(Value::as_str)
242 .map(str::to_string);
243 self.update_usage(event.data.get("usage"));
244 }
245 "message_stop" => self.stopped = true,
246 "ping" => {}
247 "error" => {
248 return Err(OpenAiError {
249 status: http::StatusCode::BAD_GATEWAY,
250 kind: event
251 .data
252 .pointer("/error/type")
253 .and_then(Value::as_str)
254 .unwrap_or("api_error")
255 .into(),
256 message: event
257 .data
258 .pointer("/error/message")
259 .and_then(Value::as_str)
260 .unwrap_or("Provider stream failed")
261 .into(),
262 param: None,
263 code: None,
264 retry_after: None,
265 });
266 }
267 other if !other.is_empty() => {
268 return Err(upstream_invalid(
269 format!("Unsupported provider stream event '{other}'"),
270 None::<String>,
271 ));
272 }
273 _ => {}
274 }
275 Ok(())
276 }
277
278 fn update_usage(&mut self, usage: Option<&Value>) {
279 let Some(usage) = usage else { return };
280 if let Some(value) = usage.get("input_tokens").and_then(Value::as_u64) {
281 self.usage.input_tokens = value;
282 }
283 if let Some(value) = usage.get("output_tokens").and_then(Value::as_u64) {
284 self.usage.output_tokens = value;
285 }
286 if let Some(value) = usage.get("cache_read_input_tokens").and_then(Value::as_u64) {
287 self.usage.cache_read_tokens = value;
288 }
289 if let Some(value) = usage
290 .get("cache_creation_input_tokens")
291 .and_then(Value::as_u64)
292 {
293 self.usage.cache_creation_tokens = value;
294 }
295 }
296}
297
298pub fn buffered_response(
299 surface: OpenAiSurface,
300 events: &[SseEvent],
301 response_id: &str,
302 model: &str,
303 created: u64,
304 response_metadata: &OpenAiResponseMetadata,
305) -> Result<Value, OpenAiError> {
306 let mut state = AnthropicAccumulator::default();
307 for event in events {
308 state.apply(event)?;
309 }
310 if !state.stopped {
311 return Err(upstream_invalid(
312 "Provider stream ended before message_stop",
313 None::<String>,
314 ));
315 }
316 match surface {
317 OpenAiSurface::ChatCompletions => Ok(chat_response(&state, response_id, model, created)),
318 OpenAiSurface::Responses => Ok(responses_response(
319 &state,
320 response_id,
321 model,
322 created,
323 response_metadata,
324 )),
325 }
326}
327
328pub fn chat_response(
329 state: &AnthropicAccumulator,
330 response_id: &str,
331 model: &str,
332 created: u64,
333) -> Value {
334 let content = state
335 .blocks
336 .iter()
337 .filter_map(|block| match &block.kind {
338 BlockKind::Text { text } => Some(text.as_str()),
339 _ => None,
340 })
341 .collect::<String>();
342 let reasoning = state
343 .blocks
344 .iter()
345 .filter_map(|block| match &block.kind {
346 BlockKind::Thinking { text } => Some(text.as_str()),
347 _ => None,
348 })
349 .collect::<String>();
350 let tools: Vec<Value> = state
351 .blocks
352 .iter()
353 .filter_map(|block| match &block.kind {
354 BlockKind::Tool {
355 id,
356 name,
357 arguments,
358 } => Some(json!({
359 "id":id,
360 "type":"function",
361 "function":{"name":name, "arguments":normalized_arguments(arguments)},
362 })),
363 _ => None,
364 })
365 .collect();
366 let mut message = serde_json::Map::from_iter([
367 ("role".to_string(), json!("assistant")),
368 (
369 "content".to_string(),
370 if content.is_empty() {
371 Value::Null
372 } else {
373 Value::String(content)
374 },
375 ),
376 ("refusal".to_string(), Value::Null),
377 ]);
378 if !reasoning.is_empty() {
379 message.insert("reasoning_content".to_string(), Value::String(reasoning));
380 }
381 if !tools.is_empty() {
382 message.insert("tool_calls".to_string(), Value::Array(tools));
383 }
384 if !state.citations.is_empty() {
385 message.insert(
386 "annotations".to_string(),
387 Value::Array(state.citations.iter().map(chat_citation).collect()),
388 );
389 }
390 json!({
391 "id":response_id,
392 "object":"chat.completion",
393 "created":created,
394 "model":model,
395 "choices":[{
396 "index":0,
397 "message":message,
398 "finish_reason":chat_finish_reason(state.stop_reason.as_deref()),
399 "logprobs":null,
400 }],
401 "usage":state.usage.chat_value(),
402 })
403}
404
405pub fn responses_response(
406 state: &AnthropicAccumulator,
407 response_id: &str,
408 model: &str,
409 created: u64,
410 response_metadata: &OpenAiResponseMetadata,
411) -> Value {
412 let mut output = Vec::new();
413 for block in state
414 .blocks
415 .iter()
416 .filter(|block| !matches!(block.kind, BlockKind::HostedResult))
417 {
418 match &block.kind {
419 BlockKind::Text { text } => output.push(json!({
420 "id":format!("msg_{}", stable_suffix(response_id, block.index)),
421 "type":"message",
422 "role":"assistant",
423 "status":"completed",
424 "content":[{"type":"output_text", "text":text, "annotations":state.citations.iter().map(responses_citation).collect::<Vec<_>>() }],
425 })),
426 BlockKind::Thinking { text } => output.push(json!({
427 "id":format!("rs_{}", stable_suffix(response_id, block.index)),
428 "type":"reasoning",
429 "summary":[{"type":"summary_text", "text":text}],
430 "status":"completed",
431 })),
432 BlockKind::Tool {
433 id,
434 name,
435 arguments,
436 } => output.push(json!({
437 "id":format!("fc_{}", stable_suffix(response_id, block.index)),
438 "type":"function_call",
439 "call_id":id,
440 "name":name,
441 "arguments":normalized_arguments(arguments),
442 "status":"completed",
443 })),
444 BlockKind::HostedSearch {
445 id,
446 name,
447 arguments,
448 } => output.push(json!({
449 "id":id,
450 "type":"web_search_call",
451 "status":"completed",
452 "action":hosted_search_action(name, arguments),
453 })),
454 BlockKind::HostedResult => unreachable!(),
455 }
456 }
457 let incomplete = state.stop_reason.as_deref() == Some("max_tokens");
458 json!({
459 "id":response_id,
460 "object":"response",
461 "created_at":created,
462 "status":if incomplete { "incomplete" } else { "completed" },
463 "model":model,
464 "output":output,
465 "parallel_tool_calls":false,
466 "tool_choice":response_metadata.tool_choice,
467 "tools":response_metadata.tools,
468 "error":null,
469 "incomplete_details":if incomplete { json!({"reason":"max_output_tokens"}) } else { Value::Null },
470 "usage":state.usage.responses_value(),
471 })
472}
473
474pub fn chat_finish_reason(reason: Option<&str>) -> &'static str {
475 match reason {
476 Some("tool_use") => "tool_calls",
477 Some("max_tokens") => "length",
478 _ => "stop",
479 }
480}
481
482pub fn hosted_search_action(_name: &str, arguments: &str) -> Value {
483 let query = serde_json::from_str::<Value>(arguments)
484 .ok()
485 .and_then(|value| {
486 value
487 .get("query")
488 .and_then(Value::as_str)
489 .map(str::to_string)
490 })
491 .unwrap_or_default();
492 json!({"type":"search", "query":query})
493}
494
495pub fn chat_citation(citation: &Value) -> Value {
496 json!({
497 "type":"url_citation",
498 "url_citation":{
499 "url":citation.get("url").and_then(Value::as_str).unwrap_or_default(),
500 "title":citation.get("title").and_then(Value::as_str).unwrap_or_default(),
501 "start_index":citation.get("start_index").and_then(Value::as_u64).unwrap_or_default(),
502 "end_index":citation.get("end_index").and_then(Value::as_u64).unwrap_or_default(),
503 }
504 })
505}
506
507pub fn responses_citation(citation: &Value) -> Value {
508 json!({
509 "type":"url_citation",
510 "url":citation.get("url").and_then(Value::as_str).unwrap_or_default(),
511 "title":citation.get("title").and_then(Value::as_str).unwrap_or_default(),
512 "start_index":citation.get("start_index").and_then(Value::as_u64).unwrap_or_default(),
513 "end_index":citation.get("end_index").and_then(Value::as_u64).unwrap_or_default(),
514 })
515}
516
517pub fn normalized_arguments(arguments: &str) -> String {
518 if arguments.is_empty() {
519 "{}".to_string()
520 } else {
521 arguments.to_string()
522 }
523}
524
525fn stable_suffix(response_id: &str, index: usize) -> String {
526 format!("{}_{index}", response_id.trim_start_matches("resp_"))
527}
528
529fn required_index(value: &Value) -> Result<usize, OpenAiError> {
530 value
531 .get("index")
532 .and_then(Value::as_u64)
533 .and_then(|value| usize::try_from(value).ok())
534 .ok_or_else(|| upstream_invalid("Provider event has an invalid index", None::<String>))
535}
536
537fn required_block_string(block: &Value, key: &str) -> Result<String, OpenAiError> {
538 block
539 .get(key)
540 .and_then(Value::as_str)
541 .filter(|value| !value.is_empty())
542 .map(str::to_string)
543 .ok_or_else(|| {
544 upstream_invalid(
545 format!("Provider tool block has an invalid '{key}'"),
546 None::<String>,
547 )
548 })
549}
550
551#[cfg(test)]
552mod tests {
553 use super::*;
554
555 fn events() -> Vec<SseEvent> {
556 vec![
557 SseEvent {
558 event: Some("message_start".into()),
559 data: json!({"type":"message_start","message":{"id":"msg_1","model":"kimi-k2.6","usage":{"input_tokens":7}}}),
560 },
561 SseEvent {
562 event: Some("content_block_start".into()),
563 data: json!({"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}),
564 },
565 SseEvent {
566 event: Some("content_block_delta".into()),
567 data: json!({"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"hello"}}),
568 },
569 SseEvent {
570 event: Some("content_block_stop".into()),
571 data: json!({"type":"content_block_stop","index":0}),
572 },
573 SseEvent {
574 event: Some("content_block_start".into()),
575 data: json!({"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"call_1","name":"lookup","input":{}}}),
576 },
577 SseEvent {
578 event: Some("content_block_delta".into()),
579 data: json!({"type":"content_block_delta","index":1,"delta":{"type":"input_json_delta","partial_json":"{\"q\":\"x\"}"}}),
580 },
581 SseEvent {
582 event: Some("content_block_stop".into()),
583 data: json!({"type":"content_block_stop","index":1}),
584 },
585 SseEvent {
586 event: Some("message_delta".into()),
587 data: json!({"type":"message_delta","delta":{"stop_reason":"tool_use"},"usage":{"output_tokens":3}}),
588 },
589 SseEvent {
590 event: Some("message_stop".into()),
591 data: json!({"type":"message_stop"}),
592 },
593 ]
594 }
595
596 #[test]
597 fn maps_hosted_search_and_citations_to_responses() {
598 let events = vec![
599 SseEvent {
600 event: Some("message_start".into()),
601 data: json!({"type":"message_start","message":{"id":"msg_1","usage":{}}}),
602 },
603 SseEvent {
604 event: Some("content_block_start".into()),
605 data: json!({"type":"content_block_start","index":0,"content_block":{"type":"server_tool_use","id":"search_1","name":"x_search","input":{}}}),
606 },
607 SseEvent {
608 event: Some("content_block_delta".into()),
609 data: json!({"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"query\":\"rust\"}"}}),
610 },
611 SseEvent {
612 event: Some("content_block_stop".into()),
613 data: json!({"type":"content_block_stop","index":0}),
614 },
615 SseEvent {
616 event: Some("content_block_start".into()),
617 data: json!({"type":"content_block_start","index":1,"content_block":{"type":"x_search_tool_result","tool_use_id":"search_1","content":[]}}),
618 },
619 SseEvent {
620 event: Some("content_block_stop".into()),
621 data: json!({"type":"content_block_stop","index":1}),
622 },
623 SseEvent {
624 event: Some("content_block_start".into()),
625 data: json!({"type":"content_block_start","index":2,"content_block":{"type":"text","text":""}}),
626 },
627 SseEvent {
628 event: Some("content_block_delta".into()),
629 data: json!({"type":"content_block_delta","index":2,"delta":{"type":"text_delta","text":"result"}}),
630 },
631 SseEvent {
632 event: Some("content_block_delta".into()),
633 data: json!({"type":"content_block_delta","index":2,"delta":{"type":"citations_delta","citation":{"url":"https://example.com","title":"Example"}}}),
634 },
635 SseEvent {
636 event: Some("content_block_stop".into()),
637 data: json!({"type":"content_block_stop","index":2}),
638 },
639 SseEvent {
640 event: Some("message_delta".into()),
641 data: json!({"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":1}}),
642 },
643 SseEvent {
644 event: Some("message_stop".into()),
645 data: json!({"type":"message_stop"}),
646 },
647 ];
648 let response = buffered_response(
649 OpenAiSurface::Responses,
650 &events,
651 "resp_test",
652 "grok-4.5",
653 1,
654 &OpenAiResponseMetadata::default(),
655 )
656 .unwrap();
657 assert!(
658 response["output"]
659 .as_array()
660 .unwrap()
661 .iter()
662 .any(|item| item["type"] == "web_search_call")
663 );
664 let message = response["output"]
665 .as_array()
666 .unwrap()
667 .iter()
668 .find(|item| item["type"] == "message")
669 .unwrap();
670 assert_eq!(
671 message["content"][0]["annotations"][0]["type"],
672 "url_citation"
673 );
674 }
675
676 #[test]
677 fn renders_chat_tool_calls_and_usage() {
678 let response = buffered_response(
679 OpenAiSurface::ChatCompletions,
680 &events(),
681 "chatcmpl_test",
682 "kimi-k2.6",
683 1,
684 &OpenAiResponseMetadata::default(),
685 )
686 .unwrap();
687 assert_eq!(response["choices"][0]["message"]["content"], "hello");
688 assert_eq!(
689 response["choices"][0]["message"]["tool_calls"][0]["function"]["name"],
690 "lookup"
691 );
692 assert_eq!(response["choices"][0]["finish_reason"], "tool_calls");
693 assert_eq!(response["usage"]["total_tokens"], 10);
694 }
695
696 #[test]
697 fn renders_responses_function_items() {
698 let response = buffered_response(
699 OpenAiSurface::Responses,
700 &events(),
701 "resp_test",
702 "kimi-k2.6",
703 1,
704 &OpenAiResponseMetadata::default(),
705 )
706 .unwrap();
707 assert_eq!(response["object"], "response");
708 assert_eq!(response["output"][0]["type"], "message");
709 assert_eq!(response["output"][1]["type"], "function_call");
710 assert_eq!(response["usage"]["total_tokens"], 10);
711 }
712}