| 1162 | } |
| 1163 | |
| 1164 | async fn consume_remote_sse( |
| 1165 | response: reqwest::Response, |
| 1166 | session_id: &str, |
| 1167 | format: RunOutputFormat, |
| 1168 | ) -> anyhow::Result<()> { |
| 1169 | let mut stream = response.bytes_stream(); |
| 1170 | let mut buffer = String::new(); |
| 1171 | let mut current_event: Option<String> = None; |
| 1172 | let mut current_data: Vec<String> = Vec::new(); |
| 1173 | |
| 1174 | let dispatch_event = |event_name: Option<String>, data: String| -> anyhow::Result<()> { |
| 1175 | if data.trim().is_empty() { |
| 1176 | return Ok(()); |
| 1177 | } |
| 1178 | |
| 1179 | let parsed: serde_json::Value = |
| 1180 | serde_json::from_str(&data).unwrap_or_else(|_| serde_json::json!({ "raw": data })); |
| 1181 | let event_type = event_name |
| 1182 | .or_else(|| { |
| 1183 | parsed |
| 1184 | .get("type") |
| 1185 | .and_then(|v| v.as_str()) |
| 1186 | .map(|s| s.to_string()) |
| 1187 | }) |
| 1188 | .unwrap_or_else(|| "message".to_string()); |
| 1189 | |
| 1190 | if matches!(format, RunOutputFormat::Json) { |
| 1191 | let mut output = serde_json::Map::new(); |
| 1192 | output.insert( |
| 1193 | "type".to_string(), |
| 1194 | serde_json::Value::String(event_type.clone()), |
| 1195 | ); |
| 1196 | output.insert( |
| 1197 | "timestamp".to_string(), |
| 1198 | serde_json::json!(chrono::Utc::now().timestamp_millis()), |
| 1199 | ); |
| 1200 | output.insert( |
| 1201 | "sessionID".to_string(), |
| 1202 | serde_json::Value::String(session_id.to_string()), |
| 1203 | ); |
| 1204 | match parsed { |
| 1205 | serde_json::Value::Object(map) => { |
| 1206 | for (k, v) in map { |
| 1207 | output.insert(k, v); |
| 1208 | } |
| 1209 | } |
| 1210 | other => { |
| 1211 | output.insert("data".to_string(), other); |
| 1212 | } |
| 1213 | } |
| 1214 | println!("{}", serde_json::Value::Object(output)); |
| 1215 | return Ok(()); |
| 1216 | } |
| 1217 | |
| 1218 | let payload: RemoteStreamEvent = |
| 1219 | serde_json::from_value(parsed).unwrap_or(RemoteStreamEvent { |
| 1220 | event_type: Some(event_type.clone()), |
| 1221 | content: None, |