diff --git a/crates/core/src/client.rs b/crates/core/src/client.rs index ce9dc0f..72546b5 100644 --- a/crates/core/src/client.rs +++ b/crates/core/src/client.rs @@ -475,11 +475,12 @@ async fn read_sse( Err(e) => { saw_terminator = false; finish_reason = TRUNCATED.into(); - // A body that fails to decode will fail the same way again; - // only the connection is worth a fresh attempt. - if !e.is_decode() { - interrupted = Some(describe_transport(&e)); - } + // Every error this stream yields is the connection failing + // under the body: a chunked or sized body cut short surfaces + // here, not as a clean end, and this client applies no content + // decoding that could fail on its own. So each one is worth a + // fresh attempt. + interrupted = Some(describe_transport(&e)); break; } }; @@ -1026,6 +1027,20 @@ mod tests { if sse.is_empty() { continue; } + // A body prefixed with CHUNKED: is framed the way real servers + // frame a stream, `Transfer-Encoding: chunked`, and the socket + // closes after the first chunk with no terminating chunk: the + // cut arrives as a body error, not as a clean end. + if let Some(payload) = sse.strip_prefix("CHUNKED:") { + let _ = stream.write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nTransfer-Encoding: chunked\r\n\r\n{:x}\r\n{payload}\r\n", + payload.len() + ) + .as_bytes(), + ); + continue; + } let _ = stream.write_all( format!("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n{sse}").as_bytes(), ); @@ -1081,6 +1096,31 @@ mod tests { ); } + /// The same interruption as a real server delivers it. Streams are framed + /// chunked, and a connection cut mid-body arrives as a read error rather + /// than a clean end. The bug this guards: that arm never restarted the + /// reply, so the fault it was written for still ended the turn. + #[tokio::test] + async fn a_stream_cut_mid_chunk_is_started_over() { + let (result, deltas, served) = stream_sequence(vec![ + "CHUNKED:data: {\"choices\":[{\"delta\":{\"reasoning_content\":\"let me\"},\"finish_reason\":null}]}\n\n".into(), + concat!( + "data: {\"choices\":[{\"delta\":{\"content\":\"all of it\"},\"finish_reason\":null}]}\n\n", + "data: {\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}]}\n\n", + "data: [DONE]\n\n", + ) + .into(), + ]) + .await; + assert_eq!(result.finish_reason, "stop"); + assert_eq!(result.content, "all of it"); + assert_eq!(served, 2, "the cut stream was started over on a second connection"); + assert_eq!(deltas.len(), 3, "{deltas:?}"); + assert_eq!(deltas[0], "reasoning:let me"); + assert!(deltas[1].starts_with(&format!("retry:2/{MAX_ATTEMPTS}:")), "{}", deltas[1]); + assert_eq!(deltas[2], "content:all of it"); + } + /// A connection the far side drops before answering is a transport fault /// after the connection existed: each one is resent after a backoff, /// announced, and the turn goes on with the reply that arrives.