Skip to content

fix(agent): retry LLM completion on malformed 2xx JSON body #5351

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tlongwell-block merged 1 commit into from
Aug 8, 2026
+283 −3
Merged

fix(agent): retry LLM completion on malformed 2xx JSON body #5351

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
286 changes: 283 additions & 3 deletions crates/buzz-agent/src/llm.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2203,8 +2203,33 @@ where
}
}
}
return serde_json::from_slice(&buf)
.map_err(|e| PostError::Agent(AgentError::Llm(format!("json: {e}"))));
// A 2xx body that fails to parse (e.g. a provider flushing a
// truncated JSON document and closing the stream cleanly) is a
// transient upstream fault, not a request problem: retry it like a
// 5xx. Safe to re-send — a malformed body produced no parsed
// response, so no tool call was ever extracted from it, and the
// retried request is the identical completion POST captured in
// `body_bytes` at function entry.
match serde_json::from_slice(&buf) {
Ok(value) => return Ok(value),
Err(e) => {
if attempt + 1 < MAX_RETRIES {
tracing::warn!(
attempt = attempt + 1,
max_attempts = MAX_RETRIES,
error = %e,
"llm: malformed response body, retrying"
);
backoff_with_jitter(attempt).await;
continue;
}
return Err(PostError::Agent(terminal_llm_error(
call_start.elapsed(),
attempt + 1,
&format!("json: {e}"),
)));
}
}
}
// Unreachable in practice: every iteration either returns or continues.
// A fallthrough here would mean MAX_RETRIES was 0, which is rejected at
Expand Down Expand Up @@ -2610,7 +2635,31 @@ async fn openrouter_post(
}
}
}
return serde_json::from_slice(&buf).map_err(|e| AgentError::Llm(format!("json: {e}")));
// Same malformed-body retry as the shared `post()` terminal: a
// truncated 2xx JSON body is transient upstream trouble, and
// re-sending is provably tool-safe — nothing was parsed, so no tool
// call could have been extracted, and the retry re-uses the
// identical `body_bytes` completion request.
match serde_json::from_slice(&buf) {
Ok(value) => return Ok(value),
Err(e) => {
if attempt + 1 < MAX_RETRIES {
tracing::warn!(
attempt = attempt + 1,
max_attempts = MAX_RETRIES,
error = %e,
"llm: openrouter malformed response body, retrying"
);
backoff_with_jitter(attempt).await;
continue;
}
return Err(terminal_llm_error(
call_start.elapsed(),
attempt + 1,
&format!("json: {e}"),
));
}
}
}
Err(terminal_llm_error(
call_start.elapsed(),
Expand Down Expand Up @@ -4791,6 +4840,237 @@ mod tests {
);
}

/// Regression (write-compressor, tb21-twins-1): a provider returning
/// HTTP 200 with a *truncated* JSON body (cleanly closed, correct
/// framing, unparseable content) previously surfaced as a terminal
/// `AgentError::Llm("json: EOF while parsing a value ...")` on the very
/// first attempt, killing the agent turn. A malformed body is transient
/// upstream trouble and must be retried like a 5xx. Re-sending is
/// tool-safe: nothing was parsed, so no tool call was extracted from the
/// bad body, and the retry replays the identical completion request.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_retries_malformed_json_body_and_succeeds() {
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1/x", listener.local_addr().unwrap());
let accepts = Arc::new(AtomicU32::new(0));
let accepts_srv = accepts.clone();

tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
let n = accepts_srv.fetch_add(1, Ordering::SeqCst);
let mut buf = Vec::new();
let mut tmp = [0u8; 4096];
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(k) => buf.extend_from_slice(&tmp[..k]),
}
}
if n == 0 {
// First attempt: 200 OK with a truncated JSON document.
// Content-Length matches the bytes actually sent, so the
// body read completes cleanly — the fault is purely that
// the JSON is cut off mid-value.
let body = "{\"choices\":[{\"mess";
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
continue;
}
// Subsequent attempts: complete valid JSON.
let body = "{\"ok\":true}";
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
}
});

let client = Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let out = post(
&client,
&url,
&serde_json::json!({}),
false,
Duration::from_secs(5),
|b| b,
)
.await
.expect("post should succeed after retrying the malformed body");
assert_eq!(out, serde_json::json!({ "ok": true }));
assert_eq!(
accepts.load(Ordering::SeqCst),
2,
"server must see exactly 2 attempts (malformed body retried once)"
);
}

/// A persistently malformed 200 body exhausts MAX_RETRIES and surfaces
/// the terminal error with the `json:` detail plus cumulative
/// duration/attempt count — never an early first-attempt death.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn post_exhausts_retries_on_persistent_malformed_json() {
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1/x", listener.local_addr().unwrap());
let accepts = Arc::new(AtomicU32::new(0));
let accepts_srv = accepts.clone();

tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
accepts_srv.fetch_add(1, Ordering::SeqCst);
let mut buf = Vec::new();
let mut tmp = [0u8; 4096];
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(k) => buf.extend_from_slice(&tmp[..k]),
}
}
let body = "{\"choices\":[{\"mess";
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
}
});

let client = Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let err = post(
&client,
&url,
&serde_json::json!({}),
false,
Duration::from_secs(5),
|b| b,
)
.await
.unwrap_err();
match &err {
PostError::Agent(AgentError::Llm(msg)) => {
assert!(
msg.contains("json:"),
"expected the json parse detail, got: {msg}"
);
assert!(
msg.contains("cumulative") && msg.contains("3 attempts"),
"expected cumulative duration + exact attempt count, got: {msg}"
);
}
other => panic!("expected PostError::Agent(AgentError::Llm), got: {other:?}"),
}
assert_eq!(
accepts.load(Ordering::SeqCst),
MAX_RETRIES,
"server must see exactly MAX_RETRIES attempts — malformed bodies must be retried"
);
}

/// Same regression coverage for `openrouter_post`, which carries its own
/// retry loop: a truncated 200 body on the first attempt is retried and
/// the call succeeds on the second.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn openrouter_post_retries_malformed_json_body_and_succeeds() {
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpListener;

let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let url = format!("http://{}/v1/x", listener.local_addr().unwrap());
let accepts = Arc::new(AtomicU32::new(0));
let accepts_srv = accepts.clone();

tokio::spawn(async move {
loop {
let (mut sock, _) = match listener.accept().await {
Ok(p) => p,
Err(_) => return,
};
let n = accepts_srv.fetch_add(1, Ordering::SeqCst);
let mut buf = Vec::new();
let mut tmp = [0u8; 4096];
while !buf.windows(4).any(|w| w == b"\r\n\r\n") {
match sock.read(&mut tmp).await {
Ok(0) | Err(_) => return,
Ok(k) => buf.extend_from_slice(&tmp[..k]),
}
}
if n == 0 {
let body = "{\"choices\":[{\"mess";
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
continue;
}
let body = r#"{"choices":[{"message":{"role":"assistant","content":"ok"},"finish_reason":"stop"}]}"#;
let resp = format!(
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\
Content-Length: {}\r\nConnection: close\r\n\r\n{}",
body.len(),
body,
);
let _ = sock.write_all(resp.as_bytes()).await;
let _ = sock.shutdown().await;
}
});

let client = Client::builder()
.timeout(Duration::from_secs(5))
.build()
.unwrap();
let out = openrouter_post(&client, &url, &json!({}), "key", Duration::from_secs(5))
.await
.expect("openrouter_post should succeed after retrying the malformed body");
assert!(out.is_object(), "expected a JSON object: {out:?}");
assert_eq!(
accepts.load(Ordering::SeqCst),
2,
"server must see exactly 2 attempts (malformed body retried once)"
);
}

/// A body-read timeout on the first attempt triggers a retry under an
/// escalated budget, and the call succeeds on the second attempt.
///
Expand Down
Loading