Stream the tool calling rounds of the Responses API

app/MindWork AI Studio/Provider/BaseProvider.cs

@@ -39,20 +39,7 @@ public abstract class BaseProvider : IProvider, ISecretId
     /// </summary>
     private readonly ILogger logger;
 
-    protected static readonly JsonSerializerOptions JSON_SERIALIZER_OPTIONS = new()
+    protected static readonly JsonSerializerOptions JSON_SERIALIZER_OPTIONS = ProviderJsonOptions.OPTIONS;
-    {
-        PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower,
-        Converters =
-        {
-            new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower),
-            new AnnotationConverter(),
-            new MessageBaseConverter(),
-            new SubContentConverter(),
-            new SubContentImageSourceConverter(),
-            new SubContentImageUrlConverter(),
-        },
-        AllowTrailingCommas = false
-    };
 
     /// <summary>
     /// Constructor for the base provider.

app/MindWork AI Studio/Provider/OpenAI/ProviderOpenAI.cs

@@ -229,7 +229,7 @@ public sealed class ProviderOpenAI() : BaseProvider(LLMProviders.OPEN_AI, new Ur
                 additionalApiParameters,
                 providerTools,
                 runnableTools,
-                (requestDto, requestToken) => this.ExecuteResponsesRequest(requestDto, requestedSecret, requestToken));
+                (requestDto, requestToken) => this.StreamResponsesRequest(requestDto, requestedSecret, requestToken));
 
             var loop = Program.SERVICE_PROVIDER.GetRequiredService<IToolCallingLoop>();
             var loopContext = new ToolCallingLoopContext
@@ -316,22 +316,25 @@ public sealed class ProviderOpenAI() : BaseProvider(LLMProviders.OPEN_AI, new Ur
                 yield return content;
     }
 
-    private async Task<ResponsesResponse?> ExecuteResponsesRequest(ResponsesAPIRequest requestDto, RequestedSecret requestedSecret, CancellationToken token)
+    /// <summary>
+    /// Runs one round of a tool calling conversation against the Responses API.
+    /// </summary>
+    /// <remarks>
+    /// Nothing but the HTTP request is done here. The retries, the timeouts, and the error
+    /// classification come from the shared stream reader, which the tool calling rounds used to
+    /// go without; reading the events is the adapter's business.
+    /// </remarks>
+    private IAsyncEnumerable<ServerSentEvent> StreamResponsesRequest(ResponsesAPIRequest requestDto, RequestedSecret requestedSecret, CancellationToken token)
     {
-        using var request = new HttpRequestMessage(HttpMethod.Post, "responses");
+        return this.ReadServerSentEventsAsync("OpenAI", "responses call", RequestBuilder, token);
-        request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION));
-        request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json");
 
-        using var response = await this.HttpClient.SendAsync(request, token);
+        async Task<HttpRequestMessage> RequestBuilder()
-        if (!response.IsSuccessStatusCode)
         {
-            var responseBody = await response.Content.ReadAsStringAsync(token);
+            var request = new HttpRequestMessage(HttpMethod.Post, "responses");
-            LOGGER.LogError("Tool calling Responses API request failed with status code {ResponseStatusCode} and body: '{ResponseBody}'.", response.StatusCode, responseBody);
+            request.Headers.Authorization = new AuthenticationHeaderValue("Bearer", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION));
-            await ToolCallingMessages.SendToolCallingRequestFailedAsync((int)response.StatusCode);
+            request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json");
-            return null;
+            return request;
         }
-
-        return await response.Content.ReadFromJsonAsync<ResponsesResponse>(JSON_SERIALIZER_OPTIONS, token);
     }
 
     #pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously

app/MindWork AI Studio/Provider/OpenAI/ResponsesCompletedStreamLine.cs

@@ -0,0 +1,13 @@
+namespace AIStudio.Provider.OpenAI;
+
+/// <summary>
+/// The closing line of a streamed Responses API call, which repeats the whole response.
+/// </summary>
+/// <remarks>
+/// Everything the round produced comes back here, reasoning items included, in the same shape a
+/// non-streamed call would have returned. That is why a streamed tool calling round needs no
+/// reassembly: this line is the round.
+/// </remarks>
+/// <param name="Type">The type of the stream event.</param>
+/// <param name="Response">The response as a non-streamed call would have returned it.</param>
+public sealed record ResponsesCompletedStreamLine(string Type, ResponsesResponse? Response);

app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamAccumulator.cs

@@ -0,0 +1,122 @@
+using System.Text.Json;
+
+namespace AIStudio.Provider.OpenAI;
+
+/// <summary>
+/// Reads a streamed Responses API call back into the response the tool calling loop works with.
+/// </summary>
+/// <remarks>
+/// The API repeats the whole response when it is done, reasoning items included, so nothing has
+/// to be reassembled from fragments: that closing event is the round. What this type does beyond
+/// taking it is hand out text and sources while they arrive, and keep the finished output items
+/// as a fallback for gateways which never send that closing event.<br/><br/>
+/// No HTTP, no dependency injection, no provider: everything here is a decision about bytes, and
+/// those are the decisions worth having a test for.
+/// </remarks>
+public sealed class ResponsesStreamAccumulator
+{
+    private const string EVENT_COMPLETED = "response.completed";
+    private const string EVENT_TEXT_DELTA = "response.output_text.delta";
+    private const string EVENT_ANNOTATION_ADDED = "response.output_text.annotation.added";
+    private const string EVENT_OUTPUT_ITEM_DONE = "response.output_item.done";
+    
+    private readonly List<JsonElement> completedOutputItems = [];
+    private ResponsesResponse? completedResponse;
+    
+    /// <summary>
+    /// Takes the next event of the stream and returns what it has to show.
+    /// </summary>
+    /// <param name="serverSentEvent">The event to read.</param>
+    /// <returns>The text and sources of this event, both empty when it carried neither.</returns>
+    public ResponsesStreamPart Process(ServerSentEvent serverSentEvent)
+    {
+        if (serverSentEvent.Data.Length is 0)
+            return ResponsesStreamPart.Nothing;
+
+        string eventType;
+        try
+        {
+            using var document = JsonDocument.Parse(serverSentEvent.Data);
+            var root = document.RootElement;
+            if (root.ValueKind is not JsonValueKind.Object ||
+                !root.TryGetProperty("type", out var typeProperty) ||
+                typeProperty.ValueKind is not JsonValueKind.String)
+                return ResponsesStreamPart.Nothing;
+
+            eventType = typeProperty.GetString() ?? string.Empty;
+            
+            //
+            // The item is cloned because its document is disposed at the end of this block, and
+            // an element which outlives its document reads memory that is no longer there.
+            //
+            if (eventType is EVENT_OUTPUT_ITEM_DONE && root.TryGetProperty("item", out var outputItem))
+                this.completedOutputItems.Add(outputItem.Clone());
+        }
+        catch (JsonException)
+        {
+            // A line we cannot read is a line we skip, exactly as the plain text path does:
+            return ResponsesStreamPart.Nothing;
+        }
+
+        switch (eventType)
+        {
+            case EVENT_COMPLETED:
+                this.completedResponse = TryDeserialize<ResponsesCompletedStreamLine>(serverSentEvent.Data)?.Response ?? this.completedResponse;
+                return ResponsesStreamPart.Nothing;
+            
+            case EVENT_TEXT_DELTA:
+                var deltaLine = TryDeserialize<ResponsesDeltaStreamLine>(serverSentEvent.Data);
+                if (deltaLine is null || !deltaLine.ContainsContent())
+                    return ResponsesStreamPart.Nothing;
+
+                return new ResponsesStreamPart(deltaLine.GetContent().Content, []);
+            
+            case EVENT_ANNOTATION_ADDED:
+                var annotationLine = TryDeserialize<ResponsesAnnotationStreamLine>(serverSentEvent.Data);
+                if (annotationLine is null || !annotationLine.ContainsSources())
+                    return ResponsesStreamPart.Nothing;
+
+                return new ResponsesStreamPart(string.Empty, annotationLine.GetSources());
+            
+            default:
+                return ResponsesStreamPart.Nothing;
+        }
+    }
+    
+    /// <summary>
+    /// Builds the response of the round from everything the stream said.
+    /// </summary>
+    /// <returns>
+    /// The response, or null when the stream ended before it said anything usable. Null is how a
+    /// failed request and a truncated stream look from here, and both end the round.
+    /// </returns>
+    public ResponsesResponse? Build()
+    {
+        if (this.completedResponse is not null)
+            return this.completedResponse;
+
+        if (this.completedOutputItems.Count is 0)
+            return null;
+
+        //
+        // No closing event came, so the round is put back together from the items which did.
+        // Reasoning items are among them, which is what the next request needs to continue.
+        //
+        return new ResponsesResponse
+        {
+            Output = [..this.completedOutputItems],
+        };
+    }
+    
+    private static T? TryDeserialize<T>(string json) where T : class
+    {
+        try
+        {
+            return JsonSerializer.Deserialize<T>(json, ProviderJsonOptions.OPTIONS);
+        }
+        catch (JsonException)
+        {
+            return null;
+        }
+    }
+}

app/MindWork AI Studio/Provider/OpenAI/ResponsesStreamPart.cs

@@ -0,0 +1,19 @@
+namespace AIStudio.Provider.OpenAI;
+
+/// <summary>
+/// What one line of a streamed Responses API call has to show to the user.
+/// </summary>
+/// <param name="TextDelta">The text this line carried, empty when it carried none.</param>
+/// <param name="Sources">The sources this line announced, empty when it announced none.</param>
+public readonly record struct ResponsesStreamPart(string TextDelta, IList<ISource> Sources)
+{
+    /// <summary>
+    /// The part of a line which says nothing to the user, such as a bookkeeping event.
+    /// </summary>
+    public static ResponsesStreamPart Nothing => new(string.Empty, []);
+    
+    /// <summary>
+    /// Whether this part has anything to show at all.
+    /// </summary>
+    public bool HasContent => this.TextDelta.Length > 0 || this.Sources.Count > 0;
+}

app/MindWork AI Studio/Provider/OpenAI/ResponsesToolCallingAdapter.cs

@@ -15,7 +15,7 @@ namespace AIStudio.Provider.OpenAI;
 /// </remarks>
 public sealed class ResponsesToolCallingAdapter(Model chatModel, IList<object> baseInput, IDictionary<string, object> apiParameters, IList<object> providerTools,
     IReadOnlyList<(ToolDefinition Definition, IToolImplementation Implementation)> runnableTools,
-    Func<ResponsesAPIRequest, CancellationToken, Task<ResponsesResponse?>> executeRequestAsync) : IToolCallingProviderAdapter
+    Func<ResponsesAPIRequest, CancellationToken, IAsyncEnumerable<ServerSentEvent>> streamRequestAsync) : IToolCallingProviderAdapter
 {
     private readonly List<object> internalItems = [];
     private readonly List<string> recordedRequestTexts = [];
@@ -47,31 +47,35 @@ public sealed class ResponsesToolCallingAdapter(Model chatModel, IList<object> b
 
         requestInput.AddRange(this.internalItems);
 
-        var response = await executeRequestAsync(new ResponsesAPIRequest
+        var request = new ResponsesAPIRequest
         {
             Model = chatModel.Id,
             Input = requestInput,
-            Stream = false,
+            Stream = true,
             Store = false,
             Tools = includeTools ? this.effectiveProviderTools : [],
             AdditionalApiParameters = apiParameters,
-        }, token);
+        };
 
+        //
+        // The text goes out while it is being written, the round only once the stream closed it.
+        // Sources travel with the text because the API announces them as it cites them.
+        //
+        var accumulator = new ResponsesStreamAccumulator();
+        await foreach (var serverSentEvent in streamRequestAsync(request, token))
+        {
+            var part = accumulator.Process(serverSentEvent);
+            if (part.HasContent)
+                yield return ToolCallingStreamEvent.TextDelta(new ContentStreamChunk(part.TextDelta, part.Sources));
+        }
+
+        var response = accumulator.Build();
         if (response is null)
             yield break;
 
         this.lastResponse = response;
-        
-        //
-        // The whole round arrives at once for now, so its text goes out as one delta. What the
-        // loop and the UI see is already the streaming shape; only the pieces are still large.
-        //
-        var textOutput = response.GetTextOutput();
-        if (!string.IsNullOrEmpty(textOutput))
-            yield return ToolCallingStreamEvent.TextDelta(textOutput);
-
         yield return ToolCallingStreamEvent.RoundCompleted(new ToolCallingRound(
-            textOutput,
+            response.GetTextOutput(),
             response.GetFunctionCalls()
                 .Select(call => new ToolCallingRequestedCall(
                     call.CallId ?? string.Empty,

app/MindWork AI Studio/Provider/ProviderJsonOptions.cs

@@ -0,0 +1,38 @@
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+using AIStudio.Provider.Anthropic;
+using AIStudio.Provider.OpenAI;
+
+namespace AIStudio.Provider;
+
+/// <summary>
+/// The JSON options every provider request and response is read and written with.
+/// </summary>
+/// <remarks>
+/// They sit outside the provider base class so that the types which interpret a stream can share
+/// them without being a provider themselves. Those types are the ones worth testing, and a
+/// provider cannot be constructed in a test at all -- it reaches for the service provider in its
+/// constructor. Options rebuilt inside a test would be a second set of rules drifting away from
+/// the one that actually reads the wire.
+/// </remarks>
+public static class ProviderJsonOptions
+{
+    /// <summary>
+    /// The shared options.
+    /// </summary>
+    public static readonly JsonSerializerOptions OPTIONS = new()
+    {
+        PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower,
+        Converters =
+        {
+            new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower),
+            new AnnotationConverter(),
+            new MessageBaseConverter(),
+            new SubContentConverter(),
+            new SubContentImageSourceConverter(),
+            new SubContentImageUrlConverter(),
+        },
+        AllowTrailingCommas = false
+    };
+}