From e7a5d7bb8fc8c6cc4deb5bababd2cc47a6027492 Mon Sep 17 00:00:00 2001 From: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Date: Sat, 8 Aug 2026 17:03:17 -0400 Subject: [PATCH] fix(agent): retry LLM completion on malformed 2xx JSON body MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A provider that returns HTTP 200 with a truncated JSON document (cleanly closed, framing intact, content cut off mid-value) previously died on the first attempt with a terminal AgentError::Llm("json: EOF while parsing a value"), killing the agent turn before it produced anything. Observed live: deepseek via OpenRouter returned a truncated body in a tb21 bench trial and the agent exited with 0 turns (code -32000 at the ACP surface). Treat an unparseable success body like a 5xx in both HTTP loops (shared post() and openrouter_post()): retry under the existing MAX_RETRIES bound with the same backoff_with_jitter, and surface terminal_llm_error with the json parse detail when retries are exhausted. Tool-safety: the retry lives entirely inside the HTTP POST helpers, before any parsed response value exists. Tool calls are only ever extracted from a successfully parsed response; a malformed body produced none, so nothing downstream of it could have dispatched. The retry re-sends the identical request bytes captured at function entry — an LLM completion request, which executes no tools. No tool call can be re-run by this change. Co-authored-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> Signed-off-by: Eva <011987e296fd5006292d2f930b574be47c7801048d1983c46c425d3c95f0cffd@buzz.block.builderlab.xyz> --- crates/buzz-agent/src/llm.rs | 286 ++++++++++++++++++++++++++++++++++- 1 file changed, 283 insertions(+), 3 deletions(-) diff --git a/crates/buzz-agent/src/llm.rs b/crates/buzz-agent/src/llm.rs index dc9501aeef..c8c0a8550f 100644 --- a/crates/buzz-agent/src/llm.rs +++ b/crates/buzz-agent/src/llm.rs @@ -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 @@ -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(), @@ -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. ///