From 1243558bc8f73eedf37e71e5a6184de5cf1d5994 Mon Sep 17 00:00:00 2001 From: orangels Date: Sat, 9 May 2026 20:14:03 +0800 Subject: [PATCH] fix: include partial stream in interrupted debug logs --- src/protocols/anthropic.rs | 105 ++++++++++++++++++++++++++++++++++--- src/protocols/google.rs | 20 ++++--- src/protocols/openai.rs | 20 ++++--- 3 files changed, 121 insertions(+), 24 deletions(-) diff --git a/src/protocols/anthropic.rs b/src/protocols/anthropic.rs index 3175a35..a8164a2 100644 --- a/src/protocols/anthropic.rs +++ b/src/protocols/anthropic.rs @@ -191,18 +191,22 @@ pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result chunk, + Err(error) => { + write_response_request_error_debug( request, &debug_url, request_body.clone(), + status_code, + &response_headers, + &raw_stream, format!("Anthropic stream interrupted: {error}"), - "anthropic-request-error", - ); - error - }) - .context("Anthropic stream interrupted")?; + ) + .await?; + return Err(error).context("Anthropic stream interrupted"); + } + }; for line in buffer.feed(&chunk) { raw_stream.push_str(&line); raw_stream.push('\n'); @@ -891,6 +895,91 @@ mod tests { assert!(!raw.contains("anthropic-secret-token")); } + #[tokio::test] + async fn interrupted_stream_debug_records_status_headers_and_partial_stream() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("bind listener"); + let addr = listener.local_addr().expect("listener addr"); + let server = tokio::spawn(async move { + let (mut socket, _) = listener.accept().await.expect("accept request"); + let mut buffer = [0_u8; 4096]; + let _ = socket.read(&mut buffer).await.expect("read request"); + let chunk = "event: ping\ndata: {\"type\":\"ping\"}\n\n"; + let response = format!( + "HTTP/1.1 200 OK\r\n\ + content-type: text/event-stream\r\n\ + transfer-encoding: chunked\r\n\ + x-request-id: broken-stream\r\n\ + \r\n\ + {:x}\r\n{}\r\n\ + zz\r\nbroken\r\n", + chunk.len(), + chunk + ); + socket + .write_all(response.as_bytes()) + .await + .expect("write malformed response"); + }); + + let temp_dir = tempfile::tempdir().expect("create temp dir"); + let request = ModelRequest { + base_url: format!("http://{addr}"), + api_token: "anthropic-secret-token".to_string(), + model: "claude-test".to_string(), + prompt: "prompt before broken stream".to_string(), + temperature: 0.0, + max_tokens: 1024, + stream: true, + raw_debug: Some( + RawDebugConfig::new( + temp_dir.path().to_path_buf(), + "anthropic-claude-test".to_string(), + ) + .with_success_raw(false), + ), + thinking: None, + }; + + let error = super::send_stream(&Client::new(), &request) + .await + .expect_err("malformed chunked stream should fail"); + server.await.expect("server task"); + assert!(error.to_string().contains("stream interrupted")); + + let debug_files = std::fs::read_dir(temp_dir.path()) + .expect("read debug dir") + .collect::, _>>() + .expect("debug entries"); + assert_eq!(debug_files.len(), 1); + let raw = std::fs::read_to_string(debug_files[0].path()).expect("read raw debug file"); + let debug: serde_json::Value = serde_json::from_str(&raw).expect("debug json"); + + assert_eq!(debug["response"]["status"], 200); + assert_eq!( + debug["response"]["headers"]["x-request-id"], + "broken-stream" + ); + assert_eq!(debug["response"]["error_kind"], "request_error"); + assert!( + debug["response"]["body"] + .as_str() + .expect("partial stream") + .contains("event: ping") + ); + assert!( + debug["response"]["error"] + .as_str() + .expect("stream error") + .contains("Anthropic stream interrupted") + ); + assert!(!raw.contains("anthropic-secret-token")); + } + fn anthropic_request(base_url: String) -> ModelRequest { ModelRequest { base_url, diff --git a/src/protocols/google.rs b/src/protocols/google.rs index 0b892cc..228d501 100644 --- a/src/protocols/google.rs +++ b/src/protocols/google.rs @@ -184,18 +184,22 @@ pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result = None; while let Some(chunk) = stream.next().await { - let chunk = chunk - .map_err(|error| { - write_request_error_debug_blocking( + let chunk = match chunk { + Ok(chunk) => chunk, + Err(error) => { + write_response_request_error_debug( request, &debug_url, request_body.clone(), + status_code, + &response_headers, + &raw_stream, format!("Google stream interrupted: {error}"), - "google-request-error", - ); - error - }) - .context("Google stream interrupted")?; + ) + .await?; + return Err(error).context("Google stream interrupted"); + } + }; for line in buffer.feed(&chunk) { raw_stream.push_str(&line); raw_stream.push('\n'); diff --git a/src/protocols/openai.rs b/src/protocols/openai.rs index 850b6e4..40b2d19 100644 --- a/src/protocols/openai.rs +++ b/src/protocols/openai.rs @@ -181,18 +181,22 @@ pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result chunk, + Err(error) => { + write_response_request_error_debug( request, &debug_url, request_body.clone(), + status_code, + &response_headers, + &raw_stream, format!("OpenAI stream interrupted: {error}"), - "openai-request-error", - ); - error - }) - .context("OpenAI stream interrupted")?; + ) + .await?; + return Err(error).context("OpenAI stream interrupted"); + } + }; for line in buffer.feed(&chunk) { raw_stream.push_str(&line); raw_stream.push('\n');