@@ -204,6 +204,179 @@ async fn streaming_parser_closes_when_caller_cancels() {
204204 assert ! ( next. is_none( ) ) ;
205205}
206206
207+ #[ tokio:: test]
208+ async fn streaming_transport_error_after_partial_delta_does_not_emit_done ( ) {
209+ use crate :: llm:: { LlmClient , StreamEvent } ;
210+
211+ let client = OpenAiClient :: new ( "k" . to_string ( ) , "model" . to_string ( ) ) . with_http_client (
212+ std:: sync:: Arc :: new ( FailingSseHttp {
213+ chunks : vec ! [
214+ "data: {\" choices\" :[{\" delta\" :{\" content\" :\" partial\" }}]}\n \n " . to_string( ) ,
215+ ] ,
216+ } ) ,
217+ ) ;
218+ let mut rx = client
219+ . complete_streaming (
220+ & [ Message :: user ( "go" ) ] ,
221+ None ,
222+ & [ ] ,
223+ tokio_util:: sync:: CancellationToken :: new ( ) ,
224+ )
225+ . await
226+ . expect ( "stream opened" ) ;
227+
228+ let mut text = String :: new ( ) ;
229+ let mut saw_done = false ;
230+ while let Some ( event) = rx. recv ( ) . await {
231+ match event {
232+ StreamEvent :: TextDelta ( delta) => text. push_str ( & delta) ,
233+ StreamEvent :: Done ( _) => saw_done = true ,
234+ _ => { }
235+ }
236+ }
237+
238+ assert_eq ! ( text, "partial" ) ;
239+ assert ! (
240+ !saw_done,
241+ "a failed transport must close without Done so the agent retries the turn"
242+ ) ;
243+ }
244+
245+ #[ tokio:: test]
246+ async fn streaming_clean_eof_after_partial_delta_does_not_emit_done ( ) {
247+ use crate :: llm:: { LlmClient , StreamEvent } ;
248+
249+ let client = glm_client ( vec ! [
250+ "data: {\" choices\" :[{\" delta\" :{\" content\" :\" partial\" }}]}\n \n " . to_string( ) ,
251+ ] ) ;
252+ let mut rx = client
253+ . complete_streaming (
254+ & [ Message :: user ( "go" ) ] ,
255+ None ,
256+ & [ ] ,
257+ tokio_util:: sync:: CancellationToken :: new ( ) ,
258+ )
259+ . await
260+ . expect ( "stream opened" ) ;
261+
262+ let mut text = String :: new ( ) ;
263+ let mut saw_done = false ;
264+ while let Some ( event) = rx. recv ( ) . await {
265+ match event {
266+ StreamEvent :: TextDelta ( delta) => text. push_str ( & delta) ,
267+ StreamEvent :: Done ( _) => saw_done = true ,
268+ _ => { }
269+ }
270+ }
271+
272+ assert_eq ! ( text, "partial" ) ;
273+ assert ! (
274+ !saw_done,
275+ "EOF without protocol terminal evidence must close without Done"
276+ ) ;
277+ }
278+
279+ #[ tokio:: test]
280+ async fn streaming_transport_error_after_finish_reason_can_finalize ( ) {
281+ let client = OpenAiClient :: new ( "k" . to_string ( ) , "model" . to_string ( ) ) . with_http_client (
282+ std:: sync:: Arc :: new ( FailingSseHttp {
283+ chunks : vec ! [
284+ "data: {\" choices\" :[{\" delta\" :{\" content\" :\" complete\" }}]}\n \n " . to_string( ) ,
285+ "data: {\" choices\" :[{\" delta\" :{},\" finish_reason\" :\" stop\" }]}\n \n " . to_string( ) ,
286+ ] ,
287+ } ) ,
288+ ) ;
289+
290+ let response = drain_to_done ( & client) . await ;
291+ assert_eq ! ( response. text( ) , "complete" ) ;
292+ assert_eq ! ( response. stop_reason. as_deref( ) , Some ( "stop" ) ) ;
293+ }
294+
295+ #[ tokio:: test]
296+ async fn streaming_clean_eof_after_finish_reason_can_finalize ( ) {
297+ let client = glm_client ( vec ! [
298+ "data: {\" choices\" :[{\" delta\" :{\" content\" :\" complete\" }}]}\n \n " . to_string( ) ,
299+ "data: {\" choices\" :[{\" delta\" :{},\" finish_reason\" :\" stop\" }]}\n \n " . to_string( ) ,
300+ ] ) ;
301+
302+ let response = drain_to_done ( & client) . await ;
303+ assert_eq ! ( response. text( ) , "complete" ) ;
304+ assert_eq ! ( response. stop_reason. as_deref( ) , Some ( "stop" ) ) ;
305+ }
306+
307+ #[ tokio:: test]
308+ async fn streaming_partial_response_is_not_finalized_after_cancellation ( ) {
309+ use crate :: llm:: { LlmClient , StreamEvent } ;
310+
311+ let client = OpenAiClient :: new ( "k" . to_string ( ) , "model" . to_string ( ) ) . with_http_client (
312+ std:: sync:: Arc :: new ( ChunksThenPendingSseHttp {
313+ chunks : vec ! [
314+ "data: {\" choices\" :[{\" delta\" :{\" content\" :\" partial\" }}]}\n \n " . to_string( ) ,
315+ ] ,
316+ } ) ,
317+ ) ;
318+ let cancellation = tokio_util:: sync:: CancellationToken :: new ( ) ;
319+ let mut rx = client
320+ . complete_streaming ( & [ Message :: user ( "go" ) ] , None , & [ ] , cancellation. clone ( ) )
321+ . await
322+ . expect ( "stream opened" ) ;
323+
324+ assert ! ( matches!(
325+ rx. recv( ) . await ,
326+ Some ( StreamEvent :: TextDelta ( text) ) if text == "partial"
327+ ) ) ;
328+ cancellation. cancel ( ) ;
329+
330+ let next = tokio:: time:: timeout ( std:: time:: Duration :: from_millis ( 100 ) , rx. recv ( ) )
331+ . await
332+ . expect ( "provider parser must stop after cancellation" ) ;
333+ assert ! ( next. is_none( ) , "cancellation must not synthesize Done" ) ;
334+ }
335+
336+ #[ tokio:: test]
337+ async fn streaming_done_closes_before_pending_transport_and_emits_once ( ) {
338+ use crate :: llm:: { LlmClient , StreamEvent } ;
339+
340+ let client = OpenAiClient :: new ( "k" . to_string ( ) , "model" . to_string ( ) ) . with_http_client (
341+ std:: sync:: Arc :: new ( ChunksThenPendingSseHttp {
342+ chunks : vec ! [
343+ "data: {\" choices\" :[{\" delta\" :{\" content\" :\" complete\" }}]}\n \n " . to_string( ) ,
344+ "data: [DONE]\n \n " . to_string( ) ,
345+ "data: [DONE]\n \n " . to_string( ) ,
346+ ] ,
347+ } ) ,
348+ ) ;
349+ let mut rx = client
350+ . complete_streaming (
351+ & [ Message :: user ( "go" ) ] ,
352+ None ,
353+ & [ ] ,
354+ tokio_util:: sync:: CancellationToken :: new ( ) ,
355+ )
356+ . await
357+ . expect ( "stream opened" ) ;
358+
359+ let events = tokio:: time:: timeout ( std:: time:: Duration :: from_secs ( 1 ) , async move {
360+ let mut events = Vec :: new ( ) ;
361+ while let Some ( event) = rx. recv ( ) . await {
362+ events. push ( event) ;
363+ }
364+ events
365+ } )
366+ . await
367+ . expect ( "[DONE] must close the parser without waiting for transport EOF" ) ;
368+ let done = events
369+ . into_iter ( )
370+ . filter_map ( |event| match event {
371+ StreamEvent :: Done ( response) => Some ( response) ,
372+ _ => None ,
373+ } )
374+ . collect :: < Vec < _ > > ( ) ;
375+
376+ assert_eq ! ( done. len( ) , 1 , "[DONE] must emit exactly one final response" ) ;
377+ assert_eq ! ( done[ 0 ] . text( ) , "complete" ) ;
378+ }
379+
207380#[ tokio:: test]
208381async fn streaming_parser_preserves_unicode_split_across_transport_chunks ( ) {
209382 let wire = concat ! (
0 commit comments