fix: fuse unfold streams to prevent panics from poll after termination (#5872)

lib/llm/src/backend.rs

@@ -262,7 +262,8 @@ impl
 
                 None => None,
             }
-        });
+        })
+        .fuse();
 
         // convert stream of processed Annotated<LLMEngineOutput> to Annotated<BackendOutput>
         //let mdcsum = self.mdcsum.clone();

lib/llm/src/migration.rs

@@ -73,7 +73,8 @@ impl
                 .next()
                 .await
                 .map(|response| (response, retry_manager))
-        });
+        })
+        .fuse();
         Ok(ResponseStream::new(Box::pin(response_stream), engine_ctx_))
     }
 }

lib/llm/src/preprocessor.rs

@@ -739,6 +739,7 @@ impl OpenAIPreprocessor {
                 }
             }
         })
+        .fuse()
     }
 
     /// Transform engine embedding output stream to OpenAI embedding response stream
@@ -905,6 +906,7 @@ impl OpenAIPreprocessor {
                 None
             }
         })
+        .fuse()
     }
 }