Read every provider stream through one server-sent event reader

app/MindWork AI Studio/Provider/BaseProvider.cs

@@ -840,19 +840,20 @@ public abstract class BaseProvider : IProvider, ISecretId
     }
 
     /// <summary>
-    /// Streams the chat completion from the provider using the Chat Completion API.
+    /// Reads a server-sent event stream from the provider, line by line.
     /// </summary>
-    /// <param name="providerName">The name of the provider.</param>
+    /// <remarks>
+    /// Everything on the way to a line is here: the retries, the timeouts, the cancellation, and
+    /// the messages the user gets to see when any of it fails. What a line means is not here --
+    /// that differs per wire format, and reading it is the caller's business.
+    /// </remarks>
+    /// <param name="providerName">The name of the provider, for logging and error reporting.</param>
+    /// <param name="operationName">What is being streamed, for logging: a chat completion, say, or a responses call.</param>
     /// <param name="requestBuilder">A function that builds the request.</param>
     /// <param name="token">The cancellation token to use.</param>
-    /// <typeparam name="TDelta">The type of the delta lines inside the stream.</typeparam>
+    /// <returns>The events of the stream, in the order they arrived.</returns>
-    /// <typeparam name="TAnnotation">The type of the annotation lines inside the stream.</typeparam>
+    protected async IAsyncEnumerable<ServerSentEvent> ReadServerSentEventsAsync(string providerName, string operationName, Func<Task<HttpRequestMessage>> requestBuilder, [EnumeratorCancellation] CancellationToken token = default)
-    /// <returns>The stream of content chunks.</returns>
-    protected async IAsyncEnumerable<ContentStreamChunk> StreamChatCompletionInternal<TDelta, TAnnotation>(string providerName, Func<Task<HttpRequestMessage>> requestBuilder, [EnumeratorCancellation] CancellationToken token = default) where TDelta : IResponseStreamLine where TAnnotation : IAnnotationStreamLine
     {
-        // Check if annotations are supported:
-        var annotationSupported = typeof(TAnnotation) != typeof(NoResponsesAnnotationStreamLine) && typeof(TAnnotation) != typeof(NoChatCompletionAnnotationStreamLine);
-        
         StreamReader? streamReader = null;
         using var timeoutTokenSource = ExternalHttpClientTimeout.CreateTimeoutTokenSource(token);
         var timeoutToken = timeoutTokenSource.Token;
@@ -862,7 +863,7 @@ public abstract class BaseProvider : IProvider, ISecretId
             var responseData = await this.SendRequest(requestBuilder, token, timeoutToken);
             if(responseData.IsFailedAfterAllRetries)
             {
-                this.logger.LogError($"The {providerName} chat completion failed: {responseData.ErrorMessage}");
+                this.logger.LogError("The {ProviderName} {OperationName} failed: {ErrorMessage}", providerName, operationName, responseData.ErrorMessage);
                 yield break;
             }
             
@@ -880,108 +881,139 @@ public abstract class BaseProvider : IProvider, ISecretId
         {
             if (token.IsCancellationRequested)
             {
-                this.logger.LogWarning("The user canceled the chat completion request for {ProviderName} '{ProviderInstanceName}' before the response stream was opened.", providerName, this.InstanceName);
+                this.logger.LogWarning("The user canceled the {OperationName} request for {ProviderName} '{ProviderInstanceName}' before the response stream was opened.", operationName, providerName, this.InstanceName);
             }
             else if (this.IsTimeoutException(e, token))
             {
                 await this.SendTimeoutError("opening the chat response stream");
-                this.logger.LogError(e, "Timed out while opening the chat completion stream from {ProviderName} '{ProviderInstanceName}'.", providerName, this.InstanceName);
+                this.logger.LogError(e, "Timed out while opening the {OperationName} stream from {ProviderName} '{ProviderInstanceName}'.", operationName, providerName, this.InstanceName);
             }
             else
             {
                 await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to communicate with the LLM provider '{0}'. There were some problems with the request. The provider message is: '{1}'"), this.InstanceName, e.Message)));
-                this.logger.LogError($"Failed to stream chat completion from {providerName} '{this.InstanceName}': {e.Message}");
+                this.logger.LogError(e, "Failed to stream the {OperationName} from {ProviderName} '{ProviderInstanceName}': {ErrorMessage}", operationName, providerName, this.InstanceName, e.Message);
             }
         }
 
         if (streamReader is null)
             yield break;
-        
+
-        //
+        try
-        // Read the stream, line by line:
-        //
-        while (true)
         {
-            try
+            //
+            // Read the stream, line by line:
+            //
+            while (true)
             {
-                if(streamReader.EndOfStream)
+                try
+                {
+                    if(streamReader.EndOfStream)
+                        break;
+                }
+                catch (Exception e)
+                {
+                    await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to stream the LLM provider '{0}' answer. There were some problems with the stream. The message is: '{1}'"), this.InstanceName, e.Message)));
+                    this.logger.LogWarning(e, "Failed to read the end-of-stream state from {ProviderName} '{ProviderInstanceName}': {ErrorMessage}", providerName, this.InstanceName, e.Message);
                     break;
-            }
+                }
-            catch (Exception e)
-            {
-                await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to stream the LLM provider '{0}' answer. There were some problems with the stream. The message is: '{1}'"), this.InstanceName, e.Message)));
-                this.logger.LogWarning($"Failed to read the end-of-stream state from {providerName} '{this.InstanceName}': {e.Message}");
-                break;
-            }
 
-            // Check if the token is canceled:
+                // Check if the token is canceled:
-            if (token.IsCancellationRequested)
-            {
-                this.logger.LogWarning($"The user canceled the chat completion for {providerName} '{this.InstanceName}'.");
-                streamReader.Close();
-                yield break;
-            }
-
-            //
-            // Read the next line:
-            //
-            string? line;
-            try
-            {
-                line = await streamReader.ReadLineAsync(timeoutToken);
-            }
-            catch (Exception e)
-            {
                 if (token.IsCancellationRequested)
                 {
-                    this.logger.LogWarning("The user canceled the chat completion stream for {ProviderName} '{ProviderInstanceName}' while reading the next chunk.", providerName, this.InstanceName);
+                    this.logger.LogWarning("The user canceled the {OperationName} for {ProviderName} '{ProviderInstanceName}'.", operationName, providerName, this.InstanceName);
-                }
+                    yield break;
-                else if (this.IsTimeoutException(e, token))
-                {
-                    await this.SendTimeoutError("reading the chat response stream");
-                    this.logger.LogError(e, "Timed out while reading the chat stream from {ProviderName} '{ProviderInstanceName}'.", providerName, this.InstanceName);
-                }
-                else
-                {
-                    await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to stream the LLM provider '{0}' answer. Was not able to read the stream. The message is: '{1}'"), this.InstanceName, e.Message)));
-                    this.logger.LogError($"Failed to read the stream from {providerName} '{this.InstanceName}': {e.Message}");
                 }
 
-                break;
+                //
+                // Read the next line:
+                //
+                string? line;
+                try
+                {
+                    line = await streamReader.ReadLineAsync(timeoutToken);
+                }
+                catch (Exception e)
+                {
+                    if (token.IsCancellationRequested)
+                    {
+                        this.logger.LogWarning("The user canceled the {OperationName} stream for {ProviderName} '{ProviderInstanceName}' while reading the next chunk.", operationName, providerName, this.InstanceName);
+                    }
+                    else if (this.IsTimeoutException(e, token))
+                    {
+                        await this.SendTimeoutError("reading the chat response stream");
+                        this.logger.LogError(e, "Timed out while reading the {OperationName} stream from {ProviderName} '{ProviderInstanceName}'.", operationName, providerName, this.InstanceName);
+                    }
+                    else
+                    {
+                        await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to stream the LLM provider '{0}' answer. Was not able to read the stream. The message is: '{1}'"), this.InstanceName, e.Message)));
+                        this.logger.LogError(e, "Failed to read the stream from {ProviderName} '{ProviderInstanceName}': {ErrorMessage}", providerName, this.InstanceName, e.Message);
+                    }
+
+                    break;
+                }
+
+                if (line is null)
+                    break;
+
+                // Skip empty lines:
+                if (string.IsNullOrWhiteSpace(line))
+                    continue;
+                
+                if (this.TryCreateProviderRequestExceptionFromStreamLine(providerName, line, out var providerRequestException))
+                    throw providerRequestException;
+
+                //
+                // Only data lines carry a payload. Every other line goes out as it is, because
+                // some of them still say something the caller has to act on.
+                //
+                TryGetServerSentEventData(line, out var data);
+                yield return new ServerSentEvent(line, data);
             }
+        }
+        finally
+        {
+            streamReader.Dispose();
+        }
+    }
 
-            if (line is null)
+    /// <summary>
-                break;
+    /// Streams the chat completion from the provider using the Chat Completion API.
-
+    /// </summary>
-            // Skip empty lines:
+    /// <param name="providerName">The name of the provider.</param>
-            if (string.IsNullOrWhiteSpace(line))
+    /// <param name="requestBuilder">A function that builds the request.</param>
-                continue;
+    /// <param name="token">The cancellation token to use.</param>
-            
+    /// <typeparam name="TDelta">The type of the delta lines inside the stream.</typeparam>
-            if (this.TryCreateProviderRequestExceptionFromStreamLine(providerName, line, out var providerRequestException))
+    /// <typeparam name="TAnnotation">The type of the annotation lines inside the stream.</typeparam>
-                throw providerRequestException;
+    /// <returns>The stream of content chunks.</returns>
-
+    protected async IAsyncEnumerable<ContentStreamChunk> StreamChatCompletionInternal<TDelta, TAnnotation>(string providerName, Func<Task<HttpRequestMessage>> requestBuilder, [EnumeratorCancellation] CancellationToken token = default) where TDelta : IResponseStreamLine where TAnnotation : IAnnotationStreamLine
-            // Skip lines that do not start with "data:". According
+    {
-            // to the specification, we only want to read the data lines:
+        // Check if annotations are supported:
-            if (!TryGetServerSentEventData(line, out var jsonData))
+        var annotationSupported = typeof(TAnnotation) != typeof(NoResponsesAnnotationStreamLine) && typeof(TAnnotation) != typeof(NoChatCompletionAnnotationStreamLine);
+        
+        await foreach (var serverSentEvent in this.ReadServerSentEventsAsync(providerName, "chat completion", requestBuilder, token))
+        {
+            // Skip lines without a payload. According to the specification,
+            // we only want to read the data lines:
+            if (serverSentEvent.Data.Length is 0)
                 continue;
 
             // Check if the line is the end of the stream:
-            if (jsonData is "[DONE]")
+            if (serverSentEvent.Data is "[DONE]")
                 yield break;
 
             //
             // Process annotation lines:
             //
-            if (annotationSupported && line.Contains("""
+            if (annotationSupported && serverSentEvent.Line.Contains("""
-                                                     "annotations":[
+                                                                     "annotations":[
-                                                     """, StringComparison.InvariantCulture))
+                                                                     """, StringComparison.InvariantCulture))
             {
                 TAnnotation? providerResponse;
                 
                 try
                 {
                     // Deserialize the JSON data:
-                    providerResponse = JsonSerializer.Deserialize<TAnnotation>(jsonData, JSON_SERIALIZER_OPTIONS);
+                    providerResponse = JsonSerializer.Deserialize<TAnnotation>(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
 
                     if (providerResponse is null)
                         continue;
@@ -1009,7 +1041,7 @@ public abstract class BaseProvider : IProvider, ISecretId
                 try
                 {
                     // Deserialize the JSON data:
-                    providerResponse = JsonSerializer.Deserialize<TDelta>(jsonData, JSON_SERIALIZER_OPTIONS);
+                    providerResponse = JsonSerializer.Deserialize<TDelta>(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
 
                     if (providerResponse is null)
                         continue;
@@ -1028,8 +1060,6 @@ public abstract class BaseProvider : IProvider, ISecretId
                 yield return providerResponse.GetContent();
             }
         }
-        
-        streamReader.Dispose();
     }
 
     /// <summary>
@@ -1046,132 +1076,29 @@ public abstract class BaseProvider : IProvider, ISecretId
         // Check if annotations are supported:
         var annotationSupported = typeof(TAnnotation) != typeof(NoResponsesAnnotationStreamLine) && typeof(TAnnotation) != typeof(NoChatCompletionAnnotationStreamLine);
         
-        StreamReader? streamReader = null;
+        await foreach (var serverSentEvent in this.ReadServerSentEventsAsync(providerName, "responses call", requestBuilder, token))
-        using var timeoutTokenSource = ExternalHttpClientTimeout.CreateTimeoutTokenSource(token);
-        var timeoutToken = timeoutTokenSource.Token;
-        try
         {
-            // Send the request using exponential backoff:
+            // Check if the line is the end of the stream. This one is read off the raw line
-            var responseData = await this.SendRequest(requestBuilder, token, timeoutToken);
+            // rather than off a payload, because it has none:
-            if(responseData.IsFailedAfterAllRetries)
+            if (serverSentEvent.Line.StartsWith("event: response.completed", StringComparison.InvariantCulture))
-            {
-                this.logger.LogError($"The {providerName} responses call failed: {responseData.ErrorMessage}");
-                yield break;
-            }
-            
-            // Open the response stream:
-            var providerStream = await responseData.Response!.Content.ReadAsStreamAsync(timeoutToken);
-
-            // Add a stream reader to read the stream, line by line:
-            streamReader = new StreamReader(providerStream);
-        }
-        catch(ProviderRequestException)
-        {
-            throw;
-        }
-        catch(Exception e)
-        {
-            if (token.IsCancellationRequested)
-            {
-                this.logger.LogWarning("The user canceled the responses request for {ProviderName} '{ProviderInstanceName}' before the response stream was opened.", providerName, this.InstanceName);
-            }
-            else if (this.IsTimeoutException(e, token))
-            {
-                await this.SendTimeoutError("opening the chat response stream");
-                this.logger.LogError(e, "Timed out while opening the responses stream from {ProviderName} '{ProviderInstanceName}'.", providerName, this.InstanceName);
-            }
-            else
-            {
-                await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to communicate with the LLM provider '{0}'. There were some problems with the request. The provider message is: '{1}'"), this.InstanceName, e.Message)));
-                this.logger.LogError($"Failed to stream responses from {providerName} '{this.InstanceName}': {e.Message}");
-            }
-        }
-
-        if (streamReader is null)
-            yield break;
-        
-        //
-        // Read the stream, line by line:
-        //
-        while (true)
-        {
-            try
-            {
-                if(streamReader.EndOfStream)
-                    break;
-            }
-            catch (Exception e)
-            {
-                await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to stream the LLM provider '{0}' answer. There were some problems with the stream. The message is: '{1}'"), this.InstanceName, e.Message)));
-                this.logger.LogWarning($"Failed to read the end-of-stream state from {providerName} '{this.InstanceName}': {e.Message}");
-                break;
-            }
-
-            // Check if the token is canceled:
-            if (token.IsCancellationRequested)
-            {
-                this.logger.LogWarning($"The user canceled the responses for {providerName} '{this.InstanceName}'.");
-                streamReader.Close();
-                yield break;
-            }
-
-            //
-            // Read the next line:
-            //
-            string? line;
-            try
-            {
-                line = await streamReader.ReadLineAsync(timeoutToken);
-            }
-            catch (Exception e)
-            {
-                if (token.IsCancellationRequested)
-                {
-                    this.logger.LogWarning("The user canceled the responses stream for {ProviderName} '{ProviderInstanceName}' while reading the next chunk.", providerName, this.InstanceName);
-                }
-                else if (this.IsTimeoutException(e, token))
-                {
-                    await this.SendTimeoutError("reading the chat response stream");
-                    this.logger.LogError(e, "Timed out while reading the responses stream from {ProviderName} '{ProviderInstanceName}'.", providerName, this.InstanceName);
-                }
-                else
-                {
-                    await MessageBus.INSTANCE.SendError(new(Icons.Material.Filled.Stream, string.Format(TB("Tried to stream the LLM provider '{0}' answer. Was not able to read the stream. The message is: '{1}'"), this.InstanceName, e.Message)));
-                    this.logger.LogError($"Failed to read the stream from {providerName} '{this.InstanceName}': {e.Message}");
-                }
-
-                break;
-            }
-
-            if (line is null)
-                break;
-
-            // Skip empty lines:
-            if (string.IsNullOrWhiteSpace(line))
-                continue;
-            
-            if (this.TryCreateProviderRequestExceptionFromStreamLine(providerName, line, out var providerRequestException))
-                throw providerRequestException;
-
-            // Check if the line is the end of the stream:
-            if (line.StartsWith("event: response.completed", StringComparison.InvariantCulture))
                 yield break;
             
-            if (!TryGetServerSentEventData(line, out var jsonData))
+            // Skip lines without a payload:
+            if (serverSentEvent.Data.Length is 0)
                 continue;
 
             //
             // Find delta lines:
             //
-            if (jsonData.StartsWith("""
+            if (serverSentEvent.Data.StartsWith("""
-                                    {"type":"response.output_text.delta"
+                                                {"type":"response.output_text.delta"
-                                    """, StringComparison.InvariantCulture))
+                                                """, StringComparison.InvariantCulture))
             {
                 TDelta? providerResponse;
                 try
                 {
                     // Deserialize the JSON data:
-                    providerResponse = JsonSerializer.Deserialize<TDelta>(jsonData, JSON_SERIALIZER_OPTIONS);
+                    providerResponse = JsonSerializer.Deserialize<TDelta>(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
 
                     if (providerResponse is null)
                         continue;
@@ -1193,7 +1120,7 @@ public abstract class BaseProvider : IProvider, ISecretId
             //
             // Find annotation added lines:
             //
-            else if (annotationSupported && jsonData.StartsWith(
+            else if (annotationSupported && serverSentEvent.Data.StartsWith(
                          """
                          {"type":"response.output_text.annotation.added"
                          """, StringComparison.InvariantCulture))
@@ -1202,7 +1129,7 @@ public abstract class BaseProvider : IProvider, ISecretId
                 try
                 {
                     // Deserialize the JSON data:
-                    providerResponse = JsonSerializer.Deserialize<TAnnotation>(jsonData, JSON_SERIALIZER_OPTIONS);
+                    providerResponse = JsonSerializer.Deserialize<TAnnotation>(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
 
                     if (providerResponse is null)
                         continue;
@@ -1221,8 +1148,6 @@ public abstract class BaseProvider : IProvider, ISecretId
                 yield return new(string.Empty, providerResponse.GetSources());
             }
         }
-        
-        streamReader.Dispose();
     }
 
     /// <summary>

app/MindWork AI Studio/Provider/ServerSentEvent.cs

@@ -0,0 +1,13 @@
+namespace AIStudio.Provider;
+
+/// <summary>
+/// One event of a server-sent event stream, as it came off the wire.
+/// </summary>
+/// <remarks>
+/// The raw line travels next to its payload because not every decision can be made from the
+/// payload alone: the Responses API, for one, ends its stream with an "event:" line which carries
+/// no payload at all.
+/// </remarks>
+/// <param name="Line">The line as it arrived, including its "data:" prefix when it had one.</param>
+/// <param name="Data">The payload of a data line, empty for every other kind of line.</param>
+public readonly record struct ServerSentEvent(string Line, string Data);