1use serde_json::{Value, json};
2
3use super::{ContentGenerationKind, SseFrame};
4
5pub fn synthesize_sse(
7 kind: ContentGenerationKind,
8 body: &[u8],
9) -> Result<Vec<u8>, crate::transform::TransformError> {
10 let value: Value = serde_json::from_slice(body).map_err(|error| {
11 crate::transform::TransformError::InvalidInput {
12 reason: format!("synthetic stream response is not JSON: {error}"),
13 }
14 })?;
15 let mut out = String::new();
16 match kind {
17 ContentGenerationKind::OpenAiChatCompletions => synthesize_chat(&value, &mut out),
18 ContentGenerationKind::OpenAiResponses
19 | ContentGenerationKind::OpenAiResponsesWebSocket => synthesize_responses(&value, &mut out),
20 ContentGenerationKind::ClaudeMessages => synthesize_claude(&value, &mut out),
21 ContentGenerationKind::GeminiGenerateContent => {
22 out.push_str(&SseFrame::data(value.to_string()).encode());
23 }
24 }
25 Ok(out.into_bytes())
26}
27
28fn synthesize_chat(response: &Value, out: &mut String) {
29 let choices = response
30 .get("choices")
31 .and_then(Value::as_array)
32 .map(|choices| {
33 choices
34 .iter()
35 .map(|choice| {
36 let mut delta = choice.get("message").cloned().unwrap_or_else(|| json!({}));
37 if let Some(object) = delta.as_object_mut() {
38 object.remove("annotations");
39 }
40 json!({
41 "index":choice.get("index").cloned().unwrap_or_else(|| json!(0)),
42 "delta":delta,
43 "finish_reason":choice.get("finish_reason").cloned().unwrap_or(Value::Null),
44 "logprobs":choice.get("logprobs").cloned().unwrap_or(Value::Null),
45 })
46 })
47 .collect::<Vec<_>>()
48 })
49 .unwrap_or_default();
50 let mut chunk = response.clone();
51 if let Some(object) = chunk.as_object_mut() {
52 object.insert("object".into(), json!("chat.completion.chunk"));
53 object.insert("choices".into(), Value::Array(choices));
54 }
55 out.push_str(&SseFrame::data(chunk.to_string()).encode());
56 out.push_str(&SseFrame::data("[DONE]").encode());
57}
58
59fn synthesize_responses(response: &Value, out: &mut String) {
60 let mut started = response.clone();
61 if let Some(object) = started.as_object_mut() {
62 object.insert("status".into(), json!("in_progress"));
63 object.insert("output".into(), json!([]));
64 }
65 push_named(out, json!({"type":"response.created","response":started}));
66
67 for (output_index, item) in response
68 .get("output")
69 .and_then(Value::as_array)
70 .into_iter()
71 .flatten()
72 .enumerate()
73 {
74 let mut added_item = item.clone();
75 if let Some(object) = added_item.as_object_mut() {
76 object.insert("status".into(), json!("in_progress"));
77 }
78 push_named(
79 out,
80 json!({"type":"response.output_item.added","output_index":output_index,"item":added_item}),
81 );
82 let item_id = item
83 .get("id")
84 .cloned()
85 .unwrap_or_else(|| json!(format!("item_{output_index}")));
86 match item.get("type").and_then(Value::as_str) {
87 Some("message") => synthesize_response_message(out, item, output_index, &item_id),
88 Some("function_call") => {
89 let arguments = item.get("arguments").cloned().unwrap_or_else(|| json!(""));
90 push_named(
91 out,
92 json!({"type":"response.function_call_arguments.delta","item_id":item_id,"output_index":output_index,"delta":arguments}),
93 );
94 push_named(
95 out,
96 json!({"type":"response.function_call_arguments.done","item_id":item_id,"output_index":output_index,"name":item.get("name"),"arguments":arguments}),
97 );
98 }
99 Some("custom_tool_call") => {
100 let input = item.get("input").cloned().unwrap_or_else(|| json!(""));
101 push_named(
102 out,
103 json!({"type":"response.custom_tool_call_input.delta","item_id":item_id,"output_index":output_index,"delta":input}),
104 );
105 push_named(
106 out,
107 json!({"type":"response.custom_tool_call_input.done","item_id":item_id,"output_index":output_index,"input":input}),
108 );
109 }
110 Some("reasoning") => synthesize_response_reasoning(out, item, output_index, &item_id),
111 _ => {}
112 }
113 push_named(
114 out,
115 json!({"type":"response.output_item.done","output_index":output_index,"item":item}),
116 );
117 }
118 push_named(
119 out,
120 json!({"type":"response.completed","response":response}),
121 );
122}
123
124fn synthesize_response_message(
125 out: &mut String,
126 item: &Value,
127 output_index: usize,
128 item_id: &Value,
129) {
130 for (content_index, part) in item
131 .get("content")
132 .and_then(Value::as_array)
133 .into_iter()
134 .flatten()
135 .enumerate()
136 {
137 push_named(
138 out,
139 json!({"type":"response.content_part.added","item_id":item_id,"output_index":output_index,"content_index":content_index,"part":part}),
140 );
141 let (delta_type, done_type, field) = match part.get("type").and_then(Value::as_str) {
142 Some("refusal") => ("response.refusal.delta", "response.refusal.done", "refusal"),
143 _ => (
144 "response.output_text.delta",
145 "response.output_text.done",
146 "text",
147 ),
148 };
149 let text = part.get(field).cloned().unwrap_or_else(|| json!(""));
150 push_named(
151 out,
152 json!({"type":delta_type,"item_id":item_id,"output_index":output_index,"content_index":content_index,"delta":text}),
153 );
154 push_named(
155 out,
156 json!({"type":done_type,"item_id":item_id,"output_index":output_index,"content_index":content_index,(field):text}),
157 );
158 push_named(
159 out,
160 json!({"type":"response.content_part.done","item_id":item_id,"output_index":output_index,"content_index":content_index,"part":part}),
161 );
162 }
163}
164
165fn synthesize_response_reasoning(
166 out: &mut String,
167 item: &Value,
168 output_index: usize,
169 item_id: &Value,
170) {
171 for (content_index, part) in item
172 .get("content")
173 .and_then(Value::as_array)
174 .into_iter()
175 .flatten()
176 .enumerate()
177 {
178 if let Some(text) = part.get("text") {
179 push_named(
180 out,
181 json!({"type":"response.reasoning_text.delta","item_id":item_id,"output_index":output_index,"content_index":content_index,"delta":text}),
182 );
183 push_named(
184 out,
185 json!({"type":"response.reasoning_text.done","item_id":item_id,"output_index":output_index,"content_index":content_index,"text":text}),
186 );
187 }
188 }
189}
190
191fn synthesize_claude(response: &Value, out: &mut String) {
192 let mut message = response.clone();
193 if let Some(object) = message.as_object_mut() {
194 object.insert("content".into(), json!([]));
195 object.insert("stop_reason".into(), Value::Null);
196 object.insert("stop_sequence".into(), Value::Null);
197 }
198 push_named(out, json!({"type":"message_start","message":message}));
199 for (index, block) in response
200 .get("content")
201 .and_then(Value::as_array)
202 .into_iter()
203 .flatten()
204 .enumerate()
205 {
206 let mut start = block.clone();
207 if let Some(object) = start.as_object_mut() {
208 match block.get("type").and_then(Value::as_str) {
209 Some("text") => {
210 object.insert("text".into(), json!(""));
211 }
212 Some("thinking") => {
213 object.insert("thinking".into(), json!(""));
214 }
215 Some("tool_use") => {
216 object.insert("input".into(), json!({}));
217 }
218 _ => {}
219 }
220 }
221 push_named(
222 out,
223 json!({"type":"content_block_start","index":index,"content_block":start}),
224 );
225 match block.get("type").and_then(Value::as_str) {
226 Some("text") => push_named(
227 out,
228 json!({"type":"content_block_delta","index":index,"delta":{"type":"text_delta","text":block.get("text").cloned().unwrap_or_else(|| json!(""))}}),
229 ),
230 Some("thinking") => {
231 push_named(
232 out,
233 json!({"type":"content_block_delta","index":index,"delta":{"type":"thinking_delta","thinking":block.get("thinking").cloned().unwrap_or_else(|| json!(""))}}),
234 );
235 if let Some(signature) = block.get("signature") {
236 push_named(
237 out,
238 json!({"type":"content_block_delta","index":index,"delta":{"type":"signature_delta","signature":signature}}),
239 );
240 }
241 }
242 Some("tool_use") => push_named(
243 out,
244 json!({"type":"content_block_delta","index":index,"delta":{"type":"input_json_delta","partial_json":block.get("input").cloned().unwrap_or_else(|| json!({})).to_string()}}),
245 ),
246 _ => {}
247 }
248 push_named(out, json!({"type":"content_block_stop","index":index}));
249 }
250 push_named(
251 out,
252 json!({"type":"message_delta","delta":{"stop_reason":response.get("stop_reason").cloned().unwrap_or(Value::Null),"stop_sequence":response.get("stop_sequence").cloned().unwrap_or(Value::Null)},"usage":response.get("usage").cloned().unwrap_or_else(|| json!({}))}),
253 );
254 push_named(out, json!({"type":"message_stop"}));
255}
256
257fn push_named(out: &mut String, event: Value) {
258 let name = event
259 .get("type")
260 .and_then(Value::as_str)
261 .unwrap_or("message");
262 out.push_str(&SseFrame::event(name, event.to_string()).encode());
263}