Procházet zdrojové kódy

fix: include partial stream in interrupted debug logs

codex/lq-token-test-init
orangels před 1 měsícem
rodič
revize
1243558bc8
3 změnil soubory, kde provedl 121 přidání a 24 odebrání
  1. +97
    -8
      src/protocols/anthropic.rs
  2. +12
    -8
      src/protocols/google.rs
  3. +12
    -8
      src/protocols/openai.rs

+ 97
- 8
src/protocols/anthropic.rs Zobrazit soubor

@@ -191,18 +191,22 @@ pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result<Mode
let mut done = false;

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!("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::<Result<Vec<_>, _>>()
.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,


+ 12
- 8
src/protocols/google.rs Zobrazit soubor

@@ -184,18 +184,22 @@ pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result<Mode
let mut first_token_ms: Option<u128> = 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');


+ 12
- 8
src/protocols/openai.rs Zobrazit soubor

@@ -181,18 +181,22 @@ pub async fn send_stream(client: &Client, request: &ModelRequest) -> Result<Mode
let mut done = false;

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!("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');


Načítá se…
Zrušit
Uložit