diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs
new file mode 100644
index 00000000..857219a6
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicContentBlockBuilder.cs
@@ -0,0 +1,227 @@
+using System.Buffers;
+using System.Text;
+using System.Text.Json;
+
+namespace AIStudio.Provider.Anthropic;
+
+///
+/// Puts one streamed content block back together.
+///
+///
+/// A block opens with a seed, grows through fragments, and has to end up as the very block the
+/// provider would have sent had we not streamed: it goes back on the next request, and Anthropic
+/// checks what it gets. A thinking block is the sharp edge here -- its signature has to return
+/// byte for byte with the text it was made for, or the next round is refused with a 400.
+/// This is a pure function over bytes: no HTTP, no state beyond the block itself. That is what
+/// makes it the piece worth testing against recorded streams.
+///
+public sealed class AnthropicContentBlockBuilder
+{
+ private const string TYPE_TEXT = "text";
+ private const string TYPE_TOOL_USE = "tool_use";
+ private const string TYPE_THINKING = "thinking";
+
+ private const string DELTA_TEXT = "text_delta";
+ private const string DELTA_INPUT_JSON = "input_json_delta";
+ private const string DELTA_THINKING = "thinking_delta";
+ private const string DELTA_SIGNATURE = "signature_delta";
+
+ private const string EMPTY_OBJECT = "{}";
+
+ private readonly JsonElement seed;
+ private readonly StringBuilder text = new();
+ private readonly StringBuilder toolArguments = new();
+ private readonly StringBuilder thinking = new();
+ private string signature;
+
+ ///
+ /// Opens a block from the seed the provider sent for it.
+ ///
+ /// The block as it opened.
+ public AnthropicContentBlockBuilder(JsonElement contentBlock)
+ {
+ //
+ // The seed is cloned because the document it was read from is gone by the time this block
+ // is built, and an element which outlives its document reads memory that is no longer
+ // there.
+ //
+ this.seed = contentBlock.ValueKind is JsonValueKind.Object ? contentBlock.Clone() : default;
+ this.BlockType = ReadString(this.seed, "type");
+
+ //
+ // Anthropic seeds a block with what it already has, which is usually nothing. When it is
+ // not nothing, it belongs in front of everything that follows.
+ //
+ this.text.Append(ReadString(this.seed, TYPE_TEXT));
+ this.thinking.Append(ReadString(this.seed, TYPE_THINKING));
+ this.signature = ReadString(this.seed, "signature");
+ }
+
+ ///
+ /// What kind of block this is: text, a tool use, thinking, or something we do not know.
+ ///
+ public string BlockType { get; }
+
+ ///
+ /// The ID of the tool use, for a tool use block.
+ ///
+ public string ToolUseId => ReadString(this.seed, "id");
+
+ ///
+ /// The tool arguments as they came off the wire, set only when they never parsed into an object.
+ ///
+ ///
+ /// The block itself carries an empty object then, because that is what may go back to the
+ /// provider. The call still has to be rejected rather than run with no arguments at all,
+ /// which is what this text is for.
+ ///
+ public string? UnparsableToolArguments { get; private set; }
+
+ ///
+ /// Adds the next piece of this block.
+ ///
+ /// The piece as it arrived.
+ /// The text to show, empty for every piece which is not text.
+ public string Append(AnthropicStreamDelta delta)
+ {
+ switch (delta.Type)
+ {
+ case DELTA_TEXT when delta.Text is not null:
+ this.text.Append(delta.Text);
+ return delta.Text;
+
+ case DELTA_INPUT_JSON when delta.PartialJson is not null:
+ this.toolArguments.Append(delta.PartialJson);
+ return string.Empty;
+
+ case DELTA_THINKING when delta.Thinking is not null:
+ this.thinking.Append(delta.Thinking);
+ return string.Empty;
+
+ case DELTA_SIGNATURE when delta.Signature is not null:
+ this.signature = delta.Signature;
+ return string.Empty;
+
+ default:
+ return string.Empty;
+ }
+ }
+
+ ///
+ /// Builds the finished block, in the shape a non-streamed call would have returned it.
+ ///
+ public JsonElement Build()
+ {
+ switch (this.BlockType)
+ {
+ case TYPE_TEXT:
+ return this.BuildFromSeed(new()
+ {
+ ["type"] = JsonSerializer.Serialize(TYPE_TEXT),
+ ["text"] = JsonSerializer.Serialize(this.text.ToString()),
+ });
+
+ case TYPE_THINKING:
+ //
+ // The signature travels with the thinking it belongs to. Anthropic refuses the
+ // next round without it, so it is written even when it stayed empty: a missing
+ // field and an empty one fail the same way, and the empty one says where to look.
+ //
+ return this.BuildFromSeed(new()
+ {
+ ["type"] = JsonSerializer.Serialize(TYPE_THINKING),
+ ["thinking"] = JsonSerializer.Serialize(this.thinking.ToString()),
+ ["signature"] = JsonSerializer.Serialize(this.signature),
+ });
+
+ case TYPE_TOOL_USE:
+ return this.BuildFromSeed(new()
+ {
+ ["input"] = this.BuildToolInput(),
+ });
+
+ default:
+ //
+ // Redacted thinking and anything we have not seen before go back untouched. We
+ // cannot read them, which is precisely why we must not rewrite them either.
+ //
+ return this.seed;
+ }
+ }
+
+ ///
+ /// The tool arguments as the JSON object they have to be.
+ ///
+ ///
+ /// A tool without arguments gets no fragment at all, so an empty buffer is an empty object.
+ /// A buffer which is not an object is kept aside instead: the block needs something the
+ /// provider accepts, while the call needs the text that made it invalid.
+ ///
+ private string BuildToolInput()
+ {
+ var arguments = this.toolArguments.ToString();
+ if (string.IsNullOrWhiteSpace(arguments))
+ return EMPTY_OBJECT;
+
+ try
+ {
+ using var document = JsonDocument.Parse(arguments);
+ if (document.RootElement.ValueKind is JsonValueKind.Object)
+ return arguments;
+ }
+ catch (JsonException)
+ {
+ // Falls through to the same place a well-formed non-object does:
+ }
+
+ this.UnparsableToolArguments = arguments;
+ return EMPTY_OBJECT;
+ }
+
+ ///
+ /// Writes the given properties over a copy of the seed.
+ ///
+ ///
+ /// Copying rather than rebuilding keeps whatever the provider sent along that we do not know
+ /// about. The values are JSON text, so that a string is escaped exactly once.
+ ///
+ /// The properties to write, as property name to JSON text.
+ private JsonElement BuildFromSeed(Dictionary overrides)
+ {
+ var buffer = new ArrayBufferWriter();
+ using (var writer = new Utf8JsonWriter(buffer))
+ {
+ writer.WriteStartObject();
+ if (this.seed.ValueKind is JsonValueKind.Object)
+ foreach (var property in this.seed.EnumerateObject())
+ {
+ if (overrides.ContainsKey(property.Name))
+ continue;
+
+ property.WriteTo(writer);
+ }
+
+ foreach (var (propertyName, json) in overrides)
+ {
+ writer.WritePropertyName(propertyName);
+ using var value = JsonDocument.Parse(json);
+ value.RootElement.WriteTo(writer);
+ }
+
+ writer.WriteEndObject();
+ }
+
+ using var document = JsonDocument.Parse(buffer.WrittenMemory);
+ return document.RootElement.Clone();
+ }
+
+ private static string ReadString(JsonElement item, string propertyName)
+ {
+ if (item.ValueKind is not JsonValueKind.Object ||
+ !item.TryGetProperty(propertyName, out var property) ||
+ property.ValueKind is not JsonValueKind.String)
+ return string.Empty;
+
+ return property.GetString() ?? string.Empty;
+ }
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs
new file mode 100644
index 00000000..075a1978
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicMessageStreamAccumulator.cs
@@ -0,0 +1,151 @@
+using System.Text.Json;
+
+namespace AIStudio.Provider.Anthropic;
+
+///
+/// Reads a streamed Anthropic messages call back into the answer the tool calling loop works with.
+///
+///
+/// Anthropic streams a message as a set of content blocks which open, grow, and close, correlated
+/// by their index and interleaved with one another. This type keeps one builder per index and
+/// hands out text as it arrives; everything else is bookkeeping until the message ends.
+/// No HTTP, no dependency injection, no provider: what happens here are decisions about bytes,
+/// and those are the decisions worth having a test for.
+///
+public sealed class AnthropicMessageStreamAccumulator
+{
+ private const string EVENT_BLOCK_START = "content_block_start";
+ private const string EVENT_BLOCK_DELTA = "content_block_delta";
+ private const string EVENT_BLOCK_STOP = "content_block_stop";
+ private const string EVENT_MESSAGE_DELTA = "message_delta";
+ private const string EVENT_MESSAGE_STOP = "message_stop";
+
+ private const string DELTA_TEXT = "text_delta";
+
+ private readonly Dictionary openBlocks = [];
+ private readonly SortedDictionary finishedBlocks = [];
+ private readonly Dictionary unparsableToolArguments = [];
+ private string stopReason = string.Empty;
+ private bool messageEnded;
+
+ ///
+ /// Takes the next event of the stream and returns what it has to show.
+ ///
+ /// The event to read.
+ /// The text of this event, empty when it carried none.
+ public AnthropicStreamPart Process(ServerSentEvent serverSentEvent)
+ {
+ if (serverSentEvent.Data.Length is 0)
+ return AnthropicStreamPart.Nothing;
+
+ AnthropicStreamLine line;
+ try
+ {
+ line = JsonSerializer.Deserialize(serverSentEvent.Data, ProviderJsonOptions.OPTIONS);
+ }
+ catch (JsonException)
+ {
+ // A line we cannot read is a line we skip, exactly as the plain text path does:
+ return AnthropicStreamPart.Nothing;
+ }
+
+ switch (line.Type)
+ {
+ case EVENT_BLOCK_START:
+ this.openBlocks[line.Index] = new AnthropicContentBlockBuilder(line.ContentBlock);
+ return AnthropicStreamPart.Nothing;
+
+ case EVENT_BLOCK_DELTA:
+ if (!this.openBlocks.TryGetValue(line.Index, out var openBlock))
+ {
+ //
+ // A delta for a block which never opened. Only text can be salvaged from
+ // that: a tool use without its ID and name is unanswerable, and thinking
+ // without its signature would have the next round refused. Text is kept as a
+ // block of its own so that what the user reads is what the model is told it
+ // said.
+ //
+ if (line.Delta.Type is not DELTA_TEXT)
+ return AnthropicStreamPart.Nothing;
+
+ openBlock = new AnthropicContentBlockBuilder(EmptyTextBlock());
+ this.openBlocks[line.Index] = openBlock;
+ }
+
+ return new AnthropicStreamPart(openBlock.Append(line.Delta));
+
+ case EVENT_BLOCK_STOP:
+ if (this.openBlocks.Remove(line.Index, out var finishedBlock))
+ this.Finish(line.Index, finishedBlock);
+
+ return AnthropicStreamPart.Nothing;
+
+ case EVENT_MESSAGE_DELTA:
+ //
+ // The stop reason ends the message as surely as the closing event does. Taking
+ // both means a gateway which sends only one of them still gets a round out.
+ //
+ if (!string.IsNullOrWhiteSpace(line.Delta.StopReason))
+ {
+ this.stopReason = line.Delta.StopReason;
+ this.messageEnded = true;
+ }
+
+ return AnthropicStreamPart.Nothing;
+
+ case EVENT_MESSAGE_STOP:
+ this.messageEnded = true;
+ this.MaterializeOpenBlocks();
+ return AnthropicStreamPart.Nothing;
+
+ default:
+ return AnthropicStreamPart.Nothing;
+ }
+ }
+
+ ///
+ /// Builds the answer of the round from everything the stream said.
+ ///
+ ///
+ /// The answer, or null when the stream ended before the message did. Null is how a failed
+ /// request and a stream cut off mid-sentence look from here, and both end the round.
+ ///
+ public AnthropicResponse? Build()
+ {
+ if (!this.messageEnded)
+ return null;
+
+ // Blocks whose closing event never came are finished here rather than dropped:
+ this.MaterializeOpenBlocks();
+
+ return new AnthropicResponse
+ {
+ StopReason = this.stopReason,
+ Content = [..this.finishedBlocks.Values],
+ UnparsableToolInputs = this.unparsableToolArguments,
+ };
+ }
+
+ private void MaterializeOpenBlocks()
+ {
+ foreach (var (index, builder) in this.openBlocks)
+ this.Finish(index, builder);
+
+ this.openBlocks.Clear();
+ }
+
+ private void Finish(int index, AnthropicContentBlockBuilder builder)
+ {
+ this.finishedBlocks[index] = builder.Build();
+
+ // Read after the block was built, because that is when the arguments are parsed:
+ if (builder.UnparsableToolArguments is not null && !string.IsNullOrWhiteSpace(builder.ToolUseId))
+ this.unparsableToolArguments[builder.ToolUseId] = builder.UnparsableToolArguments;
+ }
+
+ private static JsonElement EmptyTextBlock() => JsonSerializer.SerializeToElement(new
+ {
+ type = "text",
+ text = string.Empty,
+ });
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs
index b94d23cb..99071fc9 100644
--- a/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicResponse.cs
@@ -11,6 +11,15 @@ public sealed record AnthropicResponse
public IList Content { get; init; } = [];
+ ///
+ /// The argument text of those tool uses whose arguments never parsed, by tool use ID.
+ ///
+ ///
+ /// Empty for a non-streamed answer, where the arguments either arrived as an object or did
+ /// not arrive at all.
+ ///
+ public IReadOnlyDictionary UnparsableToolInputs { get; init; } = new Dictionary();
+
///
/// The tool calls the model asked for.
///
@@ -25,6 +34,7 @@ public sealed record AnthropicResponse
Id = ReadString(x, "id"),
Name = ReadString(x, "name"),
Input = x.TryGetProperty("input", out var input) ? input : default,
+ UnparsableArguments = this.UnparsableToolInputs.GetValueOrDefault(ReadString(x, "id")),
})
.Where(x => !string.IsNullOrWhiteSpace(x.Id) && !string.IsNullOrWhiteSpace(x.Name))
.ToList();
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs
new file mode 100644
index 00000000..573538d3
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamDelta.cs
@@ -0,0 +1,18 @@
+namespace AIStudio.Provider.Anthropic;
+
+///
+/// One piece of a streamed content block.
+///
+///
+/// Which of the fields is set depends on what the block is made of: text arrives as text, tool
+/// arguments as fragments of JSON, and a thinking block brings its signature in one piece at the
+/// end. The stop reason belongs to the message rather than to a block, and shares this shape
+/// because the API sends it in a delta of its own.
+///
+/// What kind of piece this is.
+/// The piece of text, for a text delta.
+/// The fragment of the tool arguments, for an input JSON delta.
+/// The piece of thinking, for a thinking delta.
+/// The signature of a thinking block, for a signature delta.
+/// Why the model stopped, for the message delta.
+public readonly record struct AnthropicStreamDelta(string? Type, string? Text, string? PartialJson, string? Thinking, string? Signature, string? StopReason);
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs
new file mode 100644
index 00000000..e58ffff1
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamLine.cs
@@ -0,0 +1,12 @@
+using System.Text.Json;
+
+namespace AIStudio.Provider.Anthropic;
+
+///
+/// One line of a streamed Anthropic messages call.
+///
+/// The kind of event this line reports.
+/// Which content block the event belongs to; blocks are correlated by it.
+/// The block as it opens, for a content block start.
+/// The piece this event adds, for a content block delta or a message delta.
+public readonly record struct AnthropicStreamLine(string? Type, int Index, JsonElement ContentBlock, AnthropicStreamDelta Delta);
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs
new file mode 100644
index 00000000..bc8f1bd1
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicStreamPart.cs
@@ -0,0 +1,22 @@
+namespace AIStudio.Provider.Anthropic;
+
+///
+/// What one line of a streamed Anthropic messages call has to show to the user.
+///
+///
+/// Only text ever shows. Thinking does not: neither of the two paths has ever put it on screen,
+/// and doing so would be a feature of its own rather than a side effect of streaming.
+///
+/// The text this line carried, empty when it carried none.
+public readonly record struct AnthropicStreamPart(string TextDelta)
+{
+ ///
+ /// The part of a line that says nothing to the user, such as an opening or closing block.
+ ///
+ public static AnthropicStreamPart Nothing => new(string.Empty);
+
+ ///
+ /// Whether this part has anything to show at all.
+ ///
+ public bool HasContent => this.TextDelta.Length > 0;
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs
index 32be87e3..cee4d160 100644
--- a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolCallingAdapter.cs
@@ -1,3 +1,5 @@
+using System.Runtime.CompilerServices;
+
using AIStudio.Tools.ToolCallingSystem;
using AIStudio.Tools.ToolCallingSystem.Harness;
@@ -15,7 +17,7 @@ namespace AIStudio.Provider.Anthropic;
///
public sealed class AnthropicToolCallingAdapter(Model chatModel, IList baseMessages, string systemPrompt, int maxTokens,
IDictionary apiParameters, IReadOnlyList<(ToolDefinition Definition, IToolImplementation Implementation)> runnableTools,
- Func> executeRequestAsync) : IToolCallingProviderAdapter
+ Func> streamRequestAsync) : IToolCallingProviderAdapter
{
private readonly List internalMessages = [];
private readonly List pendingToolResults = [];
@@ -27,7 +29,7 @@ public sealed class AnthropicToolCallingAdapter(Model chatModel, IList RecordedRequestTexts => this.recordedRequestTexts;
///
- public async Task ExecuteRoundAsync(string? finalResponseInstruction, bool includeTools, CancellationToken token = default)
+ public async IAsyncEnumerable ExecuteRoundAsync(string? finalResponseInstruction, bool includeTools, [EnumeratorCancellation] CancellationToken token = default)
{
//
// The results of the previous round are flushed here rather than when they were recorded:
@@ -39,7 +41,7 @@ public sealed class AnthropicToolCallingAdapter(Model chatModel, IList 0 ? this.tools : null,
AdditionalApiParameters = apiParameters,
- }, token);
+ };
+ //
+ // The text goes out while it is being written; the blocks are put back together behind
+ // it, because they have to return to the provider exactly as they arrived.
+ //
+ var accumulator = new AnthropicMessageStreamAccumulator();
+ await foreach (var serverSentEvent in streamRequestAsync(request, token))
+ {
+ var part = accumulator.Process(serverSentEvent);
+ if (part.HasContent)
+ yield return ToolCallingStreamEvent.TextDelta(part.TextDelta);
+ }
+
+ var response = accumulator.Build();
if (response is null)
- return null;
+ yield break;
this.lastResponse = response;
- return new ToolCallingRound(
+ yield return ToolCallingStreamEvent.RoundCompleted(new ToolCallingRound(
response.GetTextOutput(),
response.GetToolUses()
.Select(toolUse => new ToolCallingRequestedCall(
@@ -66,7 +81,7 @@ public sealed class AnthropicToolCallingAdapter(Model chatModel, IList
diff --git a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs
index bd42c326..52f3a62a 100644
--- a/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs
+++ b/app/MindWork AI Studio/Provider/Anthropic/AnthropicToolUse.cs
@@ -10,8 +10,19 @@ public sealed record AnthropicToolUse
public JsonElement Input { get; init; }
+ ///
+ /// The arguments as they came off the wire, set only when they never parsed into an object.
+ ///
+ ///
+ /// Only a streamed round can have these: the arguments arrive in fragments there, and a
+ /// stream which ends mid-fragment leaves text which is not an object. The block carries an
+ /// empty object in that case, because that is what may go back to the provider -- while the
+ /// call itself has to be rejected rather than run without the arguments it asked for.
+ ///
+ public string? UnparsableArguments { get; init; }
+
///
/// The arguments as JSON text, which is what the tool executor works with.
///
- public string Arguments => this.Input.ValueKind is JsonValueKind.Undefined ? "{}" : this.Input.GetRawText();
+ public string Arguments => this.UnparsableArguments ?? (this.Input.ValueKind is JsonValueKind.Undefined ? "{}" : this.Input.GetRawText());
}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs b/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs
index 17ad8309..21b32049 100644
--- a/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs
+++ b/app/MindWork AI Studio/Provider/Anthropic/ProviderAnthropic.cs
@@ -75,8 +75,8 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n
//
// Prepare the tools we want to use. When the model may call one, the conversation runs
- // through the harness instead of being streamed straight away: tool rounds are not
- // streamed, only the final answer is.
+ // through the harness instead of going straight to the streaming path below. It streams
+ // there as well, round by round -- what the harness adds is the tools in between.
//
var toolRegistry = Program.SERVICE_PROVIDER.GetService();
var toolExecutor = Program.SERVICE_PROVIDER.GetService();
@@ -93,7 +93,7 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n
if (toolExecutor is not null && runnableTools.Count > 0)
{
var adapter = new AnthropicToolCallingAdapter(chatModel, [..messages], systemPrompt, maxTokens, apiParameters, runnableTools,
- (requestDto, requestToken) => this.ExecuteMessagesRequest(requestDto, requestedSecret, requestToken));
+ (requestDto, requestToken) => this.StreamMessagesRequest(requestDto, requestedSecret, requestToken));
var loop = Program.SERVICE_PROVIDER.GetRequiredService();
var loopContext = new ToolCallingLoopContext
@@ -151,30 +151,25 @@ public sealed class ProviderAnthropic() : BaseProvider(LLMProviders.ANTHROPIC, n
}
///
- /// Runs one non-streamed messages request, as the tool rounds need it.
+ /// Runs one round of a tool calling conversation against the messages API.
///
///
- /// Tool rounds are not streamed: the whole answer has to be there before its tool calls can
- /// be executed. Only the final answer reaches the user through the streaming path.
+ /// 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 rounds used to go
+ /// without; reading the events is the adapter's business.
///
- /// The answer, or null when the request failed and the user was already told.
- private async Task ExecuteMessagesRequest(ChatRequest requestDto, RequestedSecret requestedSecret, CancellationToken token)
+ private IAsyncEnumerable StreamMessagesRequest(ChatRequest requestDto, RequestedSecret requestedSecret, CancellationToken token)
{
- using var request = new HttpRequestMessage(HttpMethod.Post, "messages");
- request.Headers.Add("x-api-key", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION));
- request.Headers.Add("anthropic-version", "2023-06-01");
- request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json");
-
- using var response = await this.HttpClient.SendAsync(request, token);
- if (!response.IsSuccessStatusCode)
+ async Task RequestBuilder()
{
- var responseBody = await response.Content.ReadAsStringAsync(token);
- LOGGER.LogError("Tool calling messages request failed with status code {ResponseStatusCode} and body: '{ResponseBody}'.", response.StatusCode, responseBody);
- await ToolCallingMessages.SendToolCallingRequestFailedAsync((int)response.StatusCode);
- return null;
+ var request = new HttpRequestMessage(HttpMethod.Post, "messages");
+ request.Headers.Add("x-api-key", await requestedSecret.Secret.Decrypt(Program.ENCRYPTION));
+ request.Headers.Add("anthropic-version", "2023-06-01");
+ request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json");
+ return request;
}
- return await response.Content.ReadFromJsonAsync(JSON_SERIALIZER_OPTIONS, token);
+ return this.ReadServerSentEventsAsync("Anthropic", "messages call", RequestBuilder, token);
}
#pragma warning disable CS1998 // Async method lacks 'await' operators and will run synchronously
diff --git a/app/MindWork AI Studio/Provider/BaseProvider.cs b/app/MindWork AI Studio/Provider/BaseProvider.cs
index 76541bbb..c35ff5ac 100644
--- a/app/MindWork AI Studio/Provider/BaseProvider.cs
+++ b/app/MindWork AI Studio/Provider/BaseProvider.cs
@@ -3,12 +3,10 @@ using System.Net.Http.Headers;
using System.Runtime.CompilerServices;
using System.Text;
using System.Text.Json;
-using System.Text.Json.Serialization;
using AIStudio.Chat;
using AIStudio.Models;
using AIStudio.Models.Live;
-using AIStudio.Provider.Anthropic;
using AIStudio.Provider.OpenAI;
using AIStudio.Provider.SelfHosted;
using AIStudio.Settings;
@@ -39,20 +37,7 @@ public abstract class BaseProvider : IProvider, ISecretId
///
private readonly ILogger logger;
- protected static readonly JsonSerializerOptions JSON_SERIALIZER_OPTIONS = new()
- {
- PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower,
- Converters =
- {
- new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower),
- new AnnotationConverter(),
- new MessageBaseConverter(),
- new SubContentConverter(),
- new SubContentImageSourceConverter(),
- new SubContentImageUrlConverter(),
- },
- AllowTrailingCommas = false
- };
+ protected static readonly JsonSerializerOptions JSON_SERIALIZER_OPTIONS = ProviderJsonOptions.OPTIONS;
///
/// Constructor for the base provider.
@@ -840,19 +825,20 @@ public abstract class BaseProvider : IProvider, ISecretId
}
///
- /// Streams the chat completion from the provider using the Chat Completion API.
+ /// Reads a server-sent event stream from the provider, line by line.
///
- /// The name of the provider.
+ ///
+ /// 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.
+ ///
+ /// The name of the provider, for logging and error reporting.
+ /// What is being streamed, for logging: a chat completion, say, or a responses call.
/// A function that builds the request.
/// The cancellation token to use.
- /// The type of the delta lines inside the stream.
- /// The type of the annotation lines inside the stream.
- /// The stream of content chunks.
- protected async IAsyncEnumerable StreamChatCompletionInternal(string providerName, Func> requestBuilder, [EnumeratorCancellation] CancellationToken token = default) where TDelta : IResponseStreamLine where TAnnotation : IAnnotationStreamLine
+ /// The events of the stream, in the order they arrived.
+ protected async IAsyncEnumerable ReadServerSentEventsAsync(string providerName, string operationName, Func> requestBuilder, [EnumeratorCancellation] CancellationToken token = default)
{
- // 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 +848,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 +866,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;
-
- //
- // Read the stream, line by line:
- //
- while (true)
+
+ try
{
- 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:
- 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)
- {
+ // Check if the token is canceled:
if (token.IsCancellationRequested)
{
- this.logger.LogWarning("The user canceled the chat completion 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 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}");
+ this.logger.LogWarning("The user canceled the {OperationName} for {ProviderName} '{ProviderInstanceName}'.", operationName, providerName, this.InstanceName);
+ yield break;
}
- 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)
- break;
-
- // Skip empty lines:
- if (string.IsNullOrWhiteSpace(line))
- continue;
-
- if (this.TryCreateProviderRequestExceptionFromStreamLine(providerName, line, out var providerRequestException))
- throw providerRequestException;
-
- // Skip lines that do not start with "data:". According
- // to the specification, we only want to read the data lines:
- if (!TryGetServerSentEventData(line, out var jsonData))
+ ///
+ /// Streams the chat completion from the provider using the Chat Completion API.
+ ///
+ /// The name of the provider.
+ /// A function that builds the request.
+ /// The cancellation token to use.
+ /// The type of the delta lines inside the stream.
+ /// The type of the annotation lines inside the stream.
+ /// The stream of content chunks.
+ protected async IAsyncEnumerable StreamChatCompletionInternal(string providerName, Func> 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);
+
+ 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("""
- "annotations":[
- """, StringComparison.InvariantCulture))
+ if (annotationSupported && serverSentEvent.Line.Contains("""
+ "annotations":[
+ """, StringComparison.InvariantCulture))
{
TAnnotation? providerResponse;
try
{
// Deserialize the JSON data:
- providerResponse = JsonSerializer.Deserialize(jsonData, JSON_SERIALIZER_OPTIONS);
+ providerResponse = JsonSerializer.Deserialize(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
if (providerResponse is null)
continue;
@@ -1009,7 +1026,7 @@ public abstract class BaseProvider : IProvider, ISecretId
try
{
// Deserialize the JSON data:
- providerResponse = JsonSerializer.Deserialize(jsonData, JSON_SERIALIZER_OPTIONS);
+ providerResponse = JsonSerializer.Deserialize(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
if (providerResponse is null)
continue;
@@ -1028,8 +1045,6 @@ public abstract class BaseProvider : IProvider, ISecretId
yield return providerResponse.GetContent();
}
}
-
- streamReader.Dispose();
}
///
@@ -1046,132 +1061,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;
- using var timeoutTokenSource = ExternalHttpClientTimeout.CreateTimeoutTokenSource(token);
- var timeoutToken = timeoutTokenSource.Token;
- try
+ await foreach (var serverSentEvent in this.ReadServerSentEventsAsync(providerName, "responses call", requestBuilder, token))
{
- // Send the request using exponential backoff:
- var responseData = await this.SendRequest(requestBuilder, token, timeoutToken);
- if(responseData.IsFailedAfterAllRetries)
- {
- 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))
+ // Check if the line is the end of the stream. This one is read off the raw line
+ // rather than off a payload, because it has none:
+ if (serverSentEvent.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("""
- {"type":"response.output_text.delta"
- """, StringComparison.InvariantCulture))
+ if (serverSentEvent.Data.StartsWith("""
+ {"type":"response.output_text.delta"
+ """, StringComparison.InvariantCulture))
{
TDelta? providerResponse;
try
{
// Deserialize the JSON data:
- providerResponse = JsonSerializer.Deserialize(jsonData, JSON_SERIALIZER_OPTIONS);
+ providerResponse = JsonSerializer.Deserialize(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
if (providerResponse is null)
continue;
@@ -1193,7 +1105,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 +1114,7 @@ public abstract class BaseProvider : IProvider, ISecretId
try
{
// Deserialize the JSON data:
- providerResponse = JsonSerializer.Deserialize(jsonData, JSON_SERIALIZER_OPTIONS);
+ providerResponse = JsonSerializer.Deserialize(serverSentEvent.Data, JSON_SERIALIZER_OPTIONS);
if (providerResponse is null)
continue;
@@ -1221,8 +1133,6 @@ public abstract class BaseProvider : IProvider, ISecretId
yield return new(string.Empty, providerResponse.GetSources());
}
}
-
- streamReader.Dispose();
}
///
@@ -1293,8 +1203,9 @@ public abstract class BaseProvider : IProvider, ISecretId
{
var adapter = new ChatCompletionToolCallingAdapter(requestFactory, systemPrompt, apiParameters,
runnableTools.Select(x => ProviderToolAdapters.ToChatCompletionTool(x.Definition)).ToList(), runnableTools,
- (requestDto, requestToken) => this.ExecuteChatCompletionRequest(requestDto, requestPath, requestedSecret, headersAction, requestToken),
- this.InstanceName, this.logger);
+ (requestDto, requestToken) => this.StreamChatCompletionRequest(requestDto, providerName, requestPath, requestedSecret, headersAction, requestToken),
+ ChatCompletionSourceReader.Read,
+ this.logger);
var loop = Program.SERVICE_PROVIDER.GetRequiredService();
var loopContext = new ToolCallingLoopContext
@@ -1365,16 +1276,17 @@ public abstract class BaseProvider : IProvider, ISecretId
CapabilityOverrides = this.CapabilityOverrides,
};
- private async Task ExecuteChatCompletionRequest(ChatCompletionAPIRequest requestDto, string requestPath, RequestedSecret requestedSecret,
- Action? headersAction, CancellationToken token)
+ ///
+ /// Runs one round of a tool calling conversation against a Chat Completions endpoint.
+ ///
+ ///
+ /// Nothing but the HTTP request is done here. Reading the events is the adapter's business,
+ /// and everything on the way to them -- the retries, the timeouts, the error classification --
+ /// belongs to the shared stream reader, which the tool rounds used to go without.
+ ///
+ private IAsyncEnumerable StreamChatCompletionRequest(ChatCompletionAPIRequest requestDto, string providerName, string requestPath,
+ RequestedSecret requestedSecret, Action? headersAction, CancellationToken token)
{
- var responseData = await this.SendRequest(RequestBuilder, token);
- if (responseData.IsFailedAfterAllRetries)
- return null;
-
- using var response = responseData.Response!;
- return await response.Content.ReadFromJsonAsync(JSON_SERIALIZER_OPTIONS, token);
-
async Task RequestBuilder()
{
var request = new HttpRequestMessage(HttpMethod.Post, requestPath);
@@ -1385,6 +1297,8 @@ public abstract class BaseProvider : IProvider, ISecretId
request.Content = new StringContent(JsonSerializer.Serialize(requestDto, JSON_SERIALIZER_OPTIONS), Encoding.UTF8, "application/json");
return request;
}
+
+ return this.ReadServerSentEventsAsync(providerName, "chat completion", RequestBuilder, token);
}
///
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionResponse.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionResponse.cs
deleted file mode 100644
index 7c23d0ef..00000000
--- a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionResponse.cs
+++ /dev/null
@@ -1,10 +0,0 @@
-namespace AIStudio.Provider.OpenAI;
-
-public sealed record ChatCompletionResponse
-{
- public string Id { get; init; } = string.Empty;
-
- public string Model { get; init; } = string.Empty;
-
- public IList Choices { get; init; } = [];
-}
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionResponseChoice.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionResponseChoice.cs
deleted file mode 100644
index 71887dc9..00000000
--- a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionResponseChoice.cs
+++ /dev/null
@@ -1,10 +0,0 @@
-namespace AIStudio.Provider.OpenAI;
-
-public sealed record ChatCompletionResponseChoice
-{
- public int Index { get; init; }
-
- public string FinishReason { get; init; } = string.Empty;
-
- public ChatCompletionResponseMessage Message { get; init; } = new();
-}
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionSourceReader.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionSourceReader.cs
new file mode 100644
index 00000000..e58be449
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionSourceReader.cs
@@ -0,0 +1,60 @@
+using System.Text.Json;
+
+namespace AIStudio.Provider.OpenAI;
+
+///
+/// Reads the sources a provider puts into its Chat Completions stream.
+///
+///
+/// Where those sit differs per provider: OpenAI announces them on annotation lines of their own,
+/// Perplexity puts its search results into the very line that carries the text. The plain text
+/// path reads both through the provider's own stream line types, and so does this -- otherwise
+/// the tool calling rounds would be the one place where a citation link goes missing.
+///
+public static class ChatCompletionSourceReader
+{
+ private const string DONE = "[DONE]";
+
+ ///
+ /// Reads whatever sources one line of the stream announced.
+ ///
+ /// The event to read.
+ /// The provider's delta stream line type.
+ /// The provider's annotation stream line type.
+ /// The sources of this line, empty when it announced none.
+ public static IList Read(ServerSentEvent serverSentEvent)
+ where TDelta : IResponseStreamLine
+ where TAnnotation : IAnnotationStreamLine
+ {
+ if (serverSentEvent.Data.Length is 0 || serverSentEvent.Data is DONE)
+ return [];
+
+ //
+ // The same split the plain text path makes, and for the same reason: a line is either an
+ // annotation line or a delta line, and reading it as both would count its sources twice.
+ //
+ var annotationSupported = typeof(TAnnotation) != typeof(NoResponsesAnnotationStreamLine) && typeof(TAnnotation) != typeof(NoChatCompletionAnnotationStreamLine);
+ if (annotationSupported && serverSentEvent.Line.Contains("""
+ "annotations":[
+ """, StringComparison.InvariantCulture))
+ {
+ var annotationLine = TryDeserialize(serverSentEvent.Data);
+ return annotationLine is not null && annotationLine.ContainsSources() ? annotationLine.GetSources() : [];
+ }
+
+ var deltaLine = TryDeserialize(serverSentEvent.Data);
+ return deltaLine is not null && deltaLine.ContainsSources() ? deltaLine.GetSources() : [];
+ }
+
+ private static T? TryDeserialize(string json)
+ {
+ try
+ {
+ return JsonSerializer.Deserialize(json, ProviderJsonOptions.OPTIONS);
+ }
+ catch (JsonException)
+ {
+ return default;
+ }
+ }
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionStreamDelta.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionStreamDelta.cs
new file mode 100644
index 00000000..2949f9f7
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionStreamDelta.cs
@@ -0,0 +1,36 @@
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace AIStudio.Provider.OpenAI;
+
+///
+/// What one choice of a streamed Chat Completions answer adds in this line.
+///
+///
+/// This is the delta of the plain text path plus the two fields that path has no use for: the
+/// reasoning some providers send alongside, and the tool calls the model asks for.
+///
+public sealed record ChatCompletionStreamDelta
+{
+ ///
+ /// The content as it arrived: a string for most providers, a list of parts for some.
+ ///
+ [JsonPropertyName("content")]
+ public JsonElement? RawContent { get; init; }
+
+ ///
+ /// The text of this fragment, whichever shape it arrived in.
+ ///
+ [JsonIgnore]
+ public string Content => ChatCompletionContent.GetText(this.RawContent) ?? string.Empty;
+
+ ///
+ /// The reasoning text some providers stream next to the answer.
+ ///
+ public string? ReasoningContent { get; init; }
+
+ ///
+ /// The fragments of the tool calls the model is asking for.
+ ///
+ public IList? ToolCalls { get; init; }
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionStreamPart.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionStreamPart.cs
new file mode 100644
index 00000000..6b152b93
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionStreamPart.cs
@@ -0,0 +1,19 @@
+namespace AIStudio.Provider.OpenAI;
+
+///
+/// What one line of a streamed Chat Completions answer has to show to the user.
+///
+/// The text this line carried, empty when it carried none.
+/// The sources this line announced, empty when it announced none.
+public readonly record struct ChatCompletionStreamPart(string TextDelta, IList Sources)
+{
+ ///
+ /// The part of a line which says nothing to the user, such as a fragment of a tool call.
+ ///
+ public static ChatCompletionStreamPart Nothing => new(string.Empty, []);
+
+ ///
+ /// Whether this part has anything to show at all.
+ ///
+ public bool HasContent => this.TextDelta.Length > 0 || this.Sources.Count > 0;
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallAccumulator.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallAccumulator.cs
new file mode 100644
index 00000000..91efbb9a
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallAccumulator.cs
@@ -0,0 +1,227 @@
+using System.Text;
+using System.Text.Json;
+
+namespace AIStudio.Provider.OpenAI;
+
+///
+/// Reads a streamed Chat Completions answer back into the message the tool calling loop works with.
+///
+///
+/// This one path serves seventeen providers, which is why every correlation here is staggered
+/// rather than assumed: a call is found by its index, failing that by its ID, failing that it is
+/// the one most recently opened. Gateways differ in all of these, and in whether they close the
+/// stream with a "[DONE]" at all.
+/// No HTTP, no dependency injection, no provider: what happens here are decisions about bytes,
+/// and those are the decisions worth having a test for.
+///
+///
+/// Reads the sources out of one line, in whichever shape this provider sends them. Left out, the
+/// round runs without sources, which is what a provider that sends none needs.
+///
+public sealed class ChatCompletionToolCallAccumulator(Func>? readSources = null)
+{
+ private const string DONE = "[DONE]";
+ private const string EMPTY_ARGUMENTS = "{}";
+
+ private readonly StringBuilder text = new();
+ private readonly StringBuilder reasoning = new();
+ private readonly List toolCalls = [];
+ private readonly Dictionary toolCallsByIndex = [];
+ private bool hasReadAnything;
+
+ ///
+ /// Takes the next event of the stream and returns what it has to show.
+ ///
+ /// The event to read.
+ /// The text of this event, empty when it carried none.
+ public ChatCompletionStreamPart Process(ServerSentEvent serverSentEvent)
+ {
+ if (serverSentEvent.Data.Length is 0 || serverSentEvent.Data is DONE)
+ return ChatCompletionStreamPart.Nothing;
+
+ ChatCompletionToolStreamLine? line;
+ try
+ {
+ line = JsonSerializer.Deserialize(serverSentEvent.Data, ProviderJsonOptions.OPTIONS);
+ }
+ catch (JsonException)
+ {
+ // A line we cannot read is a line we skip, exactly as the plain text path does:
+ return ChatCompletionStreamPart.Nothing;
+ }
+
+ //
+ // Only the first choice is ever used, here as much as on the plain text path: we never
+ // ask for more than one, and a provider which sends more has no say in which one counts.
+ //
+ //
+ // Sources are read off the same line, through the provider's own types: they may sit on
+ // a line of their own or right next to the text, and a line without any gives an empty
+ // list either way.
+ //
+ var sources = readSources?.Invoke(serverSentEvent) ?? [];
+
+ var delta = line?.Choices?.FirstOrDefault()?.Delta;
+ if (delta is null)
+ return WithSources(string.Empty, sources);
+
+ this.hasReadAnything = true;
+
+ if (!string.IsNullOrEmpty(delta.ReasoningContent))
+ this.reasoning.Append(delta.ReasoningContent);
+
+ foreach (var toolCallDelta in delta.ToolCalls ?? [])
+ {
+ if (toolCallDelta is null)
+ continue;
+
+ this.Apply(toolCallDelta);
+ }
+
+ var textDelta = delta.Content;
+ if (textDelta.Length is 0)
+ return WithSources(string.Empty, sources);
+
+ this.text.Append(textDelta);
+ return new ChatCompletionStreamPart(textDelta, sources);
+ }
+
+ ///
+ /// Builds the message of the round from everything the stream said.
+ ///
+ ///
+ /// The message, or null when no line of the stream was readable at all. Null is how a failed
+ /// request looks from here, and it ends the round.
+ ///
+ ///
+ /// The end of the stream is the end of the message. There is nothing else to wait for: a
+ /// "[DONE]" is not sent by every gateway, and a finish reason not by every one either.
+ ///
+ public ChatCompletionResponseMessage? Build()
+ {
+ if (!this.hasReadAnything)
+ return null;
+
+ var answer = this.text.ToString();
+ return new ChatCompletionResponseMessage
+ {
+ Role = "assistant",
+
+ //
+ // No text means no content field, the way a round which only calls a tool arrives
+ // when it is not streamed. Some providers reject an empty string in its place.
+ //
+ RawContent = answer.Length is 0 ? null : JsonSerializer.SerializeToElement(answer),
+ ReasoningContent = this.reasoning.Length is 0 ? null : this.reasoning.ToString(),
+ ToolCalls = this.toolCalls.Count is 0
+ ? null
+ : this.toolCalls.Select(toolCall => (ChatCompletionToolCall?)toolCall.Build()).ToList(),
+ };
+ }
+
+ private void Apply(ChatCompletionToolCallDelta toolCallDelta)
+ {
+ var toolCall = this.Resolve(toolCallDelta);
+
+ //
+ // The first non-empty value wins for everything but the arguments: some providers repeat
+ // the ID and the name with every fragment, and a later empty one must not erase them.
+ //
+ toolCall.Id ??= Coalesce(toolCallDelta.Id);
+ toolCall.Type ??= Coalesce(toolCallDelta.Type);
+ toolCall.Name ??= Coalesce(toolCallDelta.Function?.Name);
+
+ // The arguments are the one thing that is always appended, because that is how they come:
+ if (!string.IsNullOrEmpty(toolCallDelta.Function?.Arguments))
+ toolCall.Arguments.Append(toolCallDelta.Function.Arguments);
+ }
+
+ ///
+ /// Finds the call a fragment belongs to, or opens a new one for it.
+ ///
+ private ToolCallBuilder Resolve(ChatCompletionToolCallDelta toolCallDelta)
+ {
+ //
+ // The index is what the specification correlates by, so it comes first:
+ //
+ if (toolCallDelta.Index is { } index)
+ {
+ if (this.toolCallsByIndex.TryGetValue(index, out var knownByIndex))
+ return knownByIndex;
+
+ var openedByIndex = this.Open();
+ this.toolCallsByIndex[index] = openedByIndex;
+ return openedByIndex;
+ }
+
+ //
+ // Some gateways leave the index out and correlate by ID instead:
+ //
+ if (!string.IsNullOrWhiteSpace(toolCallDelta.Id))
+ {
+ var knownById = this.toolCalls.FirstOrDefault(x => string.Equals(x.Id, toolCallDelta.Id, StringComparison.Ordinal));
+ if (knownById is not null)
+ return knownById;
+
+ return this.Open();
+ }
+
+ //
+ // And some send neither once the call is open, which leaves the one we opened last. A
+ // fragment before any call was opened opens one, rather than being dropped.
+ //
+ return this.toolCalls.Count > 0 ? this.toolCalls[^1] : this.Open();
+ }
+
+ private ToolCallBuilder Open()
+ {
+ var toolCall = new ToolCallBuilder();
+ this.toolCalls.Add(toolCall);
+ return toolCall;
+ }
+
+ private static string? Coalesce(string? value) => string.IsNullOrWhiteSpace(value) ? null : value;
+
+ ///
+ /// A part for a line which brought sources but no text, or nothing at all.
+ ///
+ private static ChatCompletionStreamPart WithSources(string text, IList sources)
+ => sources.Count is 0 ? ChatCompletionStreamPart.Nothing : new ChatCompletionStreamPart(text, sources);
+
+ ///
+ /// One tool call while its fragments are still arriving.
+ ///
+ private sealed class ToolCallBuilder
+ {
+ public string? Id { get; set; }
+
+ public string? Type { get; set; }
+
+ public string? Name { get; set; }
+
+ public StringBuilder Arguments { get; } = new();
+
+ ///
+ /// Builds the call in the shape a non-streamed answer would have carried it.
+ ///
+ ///
+ /// A call without an ID, without a name, or with arguments which are not an object stays
+ /// as it is: the adapter has to see what the model actually sent, so that it can reject
+ /// the call the way an invalid one has to be rejected.
+ /// Empty arguments are the one exception, and they are not a correction but a
+ /// translation: a tool which takes nothing gets no fragment at all here, while the same
+ /// call arrives as an empty object when it is not streamed. Handing on the empty string
+ /// would have every parameterless tool rejected as invalid.
+ ///
+ public ChatCompletionToolCall Build() => new()
+ {
+ Id = this.Id,
+ Type = this.Type ?? "function",
+ Function = new ChatCompletionToolFunction
+ {
+ Name = this.Name,
+ Arguments = this.Arguments.Length is 0 ? EMPTY_ARGUMENTS : this.Arguments.ToString(),
+ },
+ };
+ }
+}
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallDelta.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallDelta.cs
new file mode 100644
index 00000000..0a3c6692
--- /dev/null
+++ b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallDelta.cs
@@ -0,0 +1,15 @@
+namespace AIStudio.Provider.OpenAI;
+
+///
+/// One fragment of a tool call in a streamed Chat Completions answer.
+///
+///
+/// A tool call arrives in pieces: the ID and the name usually with the first fragment, the
+/// arguments spread over as many as the model needs. The index is what ties the pieces of one
+/// call together while another call is being written at the same time.
+///
+/// Which call this fragment belongs to; null when the provider omits it.
+/// The ID of the call, sent once by most providers and repeated by some.
+/// The kind of call, which is "function" for everything we offer.
+/// The name and the arguments fragment of the call.
+public sealed record ChatCompletionToolCallDelta(int? Index, string? Id, string? Type, ChatCompletionToolFunction? Function);
\ No newline at end of file
diff --git a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallingAdapter.cs b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallingAdapter.cs
index 5c34b4f0..43bdfef4 100644
--- a/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallingAdapter.cs
+++ b/app/MindWork AI Studio/Provider/OpenAI/ChatCompletionToolCallingAdapter.cs
@@ -1,3 +1,4 @@
+using System.Runtime.CompilerServices;
using System.Text.Json;
using AIStudio.Tools.ToolCallingSystem;
@@ -17,8 +18,9 @@ public sealed class ChatCompletionToolCallingAdapter(
TextMessage systemPrompt, IDictionary apiParameters,
IList