(
&self,
response: reqwest::Response,
agent_run_id: &str,
token_sink: &S,
)
| 1417 | } |
| 1418 | |
| 1419 | async fn stream_anthropic_tool_sse<S: TokenSink + Sync>( |
| 1420 | &self, |
| 1421 | response: reqwest::Response, |
| 1422 | agent_run_id: &str, |
| 1423 | token_sink: &S, |
| 1424 | ) -> AppResult<LLMTurn> { |
| 1425 | let mut stream = response.bytes_stream(); |
| 1426 | let mut line_buffer = String::new(); |
| 1427 | let mut output = String::new(); |
| 1428 | let mut stop_reason: Option<String> = None; |
| 1429 | let mut pending_calls: HashMap<usize, StreamingToolCall> = HashMap::new(); |
| 1430 | // Track the most recent `event:` line so we can fall back to it when |
| 1431 | // the JSON payload omits the top-level `"type"` field (some gateways). |
| 1432 | let mut current_event = String::new(); |
| 1433 | |
| 1434 | while let Some(chunk) = stream.next().await { |
| 1435 | line_buffer.push_str(&String::from_utf8_lossy(&chunk?)); |
| 1436 | |
| 1437 | while let Some(pos) = line_buffer.find('\n') { |
| 1438 | let mut line = line_buffer[..pos].to_string(); |
| 1439 | line_buffer.drain(..=pos); |
| 1440 | if line.ends_with('\r') { |
| 1441 | let _ = line.pop(); |
| 1442 | } |
| 1443 | |
| 1444 | let should_stop = self.parse_anthropic_sse_line( |
| 1445 | &line, |
| 1446 | &mut current_event, |
| 1447 | agent_run_id, |
| 1448 | token_sink, |
| 1449 | &mut output, |
| 1450 | &mut pending_calls, |
| 1451 | &mut stop_reason, |
| 1452 | )?; |
| 1453 | |
| 1454 | if should_stop { |
| 1455 | return finalize_llm_turn(output, pending_calls, stop_reason); |
| 1456 | } |
| 1457 | } |
| 1458 | } |
| 1459 | |
| 1460 | finalize_llm_turn(output, pending_calls, stop_reason) |
| 1461 | } |
| 1462 | |
| 1463 | #[allow(clippy::too_many_arguments)] |
| 1464 | fn parse_anthropic_sse_line<S: TokenSink + Sync>( |
no test coverage detected