From af35040df5325076e8fa27404d03a8207a43c928 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 11:33:54 +0800 Subject: [PATCH 1/5] Stream LLM token deltas over the conversation SSE endpoint SendMessageSse only subscribed to OnIndicationReceived, so the token deltas the completion providers already raise as OnReceiveLlmStreamMessage never reached the response. They went only to the SignalR hub via ChatHubObserver, which is why BotSharp-UI shows a typewriter effect while an SSE client sees a single complete frame. Subscribing alone was not enough: ConversationObserver invoked listeners only inside its OnIndicationReceived branch, so a listener registered for any other event was silently dropped. Hoisting that dispatch out of the branches lets every registered listener receive its own event. The only other registration in the tree is Twilio's, also on OnIndicationReceived, so nothing dormant is activated. Frames are now written in one atomic write under a lock, because deltas are raised from the provider's execution context and can reach OnChunkReceived while the message callback is midway through a frame, and HttpResponse.Body is not thread-safe. The 10ms delay that sat between the two former writes is replaced by an explicit flush: at one delay per frame it would have added seconds of dead time across a token stream, and nothing was flushing before. Not verified end to end yet -- reproducing a token stream needs an LLM API key, which is not in the repo. Co-Authored-By: Claude Opus 5 --- .../Observers/ConversationObserver.cs | 10 +++-- .../Conversation/ConversationController.cs | 38 ++++++++++++++++--- 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs b/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs index 294031fa3..788cc6b96 100644 --- a/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs +++ b/src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs @@ -38,10 +38,6 @@ public override void OnNext(HubObserveData value) _logger.LogDebug($"[{nameof(ConversationObserver)}]: Receive {value.EventName} => {value.Data.Indication} ({conv.ConversationId})"); - if (_listeners.TryGetValue(value.EventName, out var func) && func != null) - { - func(value).ConfigureAwait(false).GetAwaiter().GetResult(); - } } else if (value.EventName == ChatEvent.OnIntermediateMessageReceivedFromAssistant) { @@ -53,5 +49,11 @@ public override void OnNext(HubObserveData value) storage.Append(conv.ConversationId, value.Data).ConfigureAwait(false).GetAwaiter().GetResult(); } } + + // Hoisted out of the branches above, which only ever reached a listener for OnIndicationReceived. + if (_listeners.TryGetValue(value.EventName, out var func) && func != null) + { + func(value).ConfigureAwait(false).GetAwaiter().GetResult(); + } } } diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index d6d5cda32..498edd64f 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -18,6 +18,11 @@ public partial class ConversationController : ControllerBase private readonly IUserIdentity _user; private readonly JsonSerializerOptions _jsonOptions; + private const string StreamingFlag = "streaming"; + + // Deltas arrive on the provider's context and would interleave with the message callback's frame write. + private readonly SemaphoreSlim _sseWriteLock = new(1, 1); + public ConversationController( IServiceProvider services, IUserIdentity user, @@ -456,7 +461,8 @@ public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string var observer = _services.GetRequiredService(); using var container = observer.SubscribeObservers>(conversationId, listeners: new() { - { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(conversationId, data.Data) } + { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(conversationId, data.Data) }, + { ChatEvent.OnReceiveLlmStreamMessage, async data => await OnReceiveStreamingDelta(conversationId, data.Data) } }); var conv = _services.GetRequiredService(); @@ -561,12 +567,18 @@ private async Task OnChunkReceived(HttpResponse response, ChatResponseModel mess { var json = JsonSerializer.Serialize(message, _jsonOptions); - var buffer = Encoding.UTF8.GetBytes($"data:{json}\n"); - await response.Body.WriteAsync(buffer, 0, buffer.Length); - await Task.Delay(10); + var buffer = Encoding.UTF8.GetBytes($"data:{json}\n\n"); - buffer = Encoding.UTF8.GetBytes("\n"); - await response.Body.WriteAsync(buffer, 0, buffer.Length); + await _sseWriteLock.WaitAsync(); + try + { + await response.Body.WriteAsync(buffer, 0, buffer.Length); + await response.Body.FlushAsync(); + } + finally + { + _sseWriteLock.Release(); + } } private async Task OnEventCompleted(HttpResponse response) @@ -611,5 +623,19 @@ private async Task OnReceiveToolCallIndication(string conversationId, RoleDialog }; await OnChunkReceived(Response, indicator); } + + private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogModel msg) + { + var delta = new ChatResponseModel + { + ConversationId = conversationId, + MessageId = msg.MessageId, + Text = !string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content, + Function = StreamingFlag, + Thought = msg.Thought, + States = [] + }; + await OnChunkReceived(Response, delta); + } #endregion } \ No newline at end of file From 7755d245e7e44bff424ca7575fb42973ecabd4c8 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 13:37:12 +0800 Subject: [PATCH 2/5] Drop the SSE write lock; there is no concurrent writer The previous commit justified a lock by claiming the delta listener and the message callback can write a frame at the same time. That is wrong. MessageHub wraps its subject in Subject.Synchronize, so no two OnNext calls propagate concurrently, and Push runs synchronously on the caller's thread with no scheduler. ConversationObserver then invokes the listener with GetAwaiter().GetResult(), so the provider's streaming loop blocks until the frame is written. The message callback is reached later in the same await chain, via HandleAssistantMessage inside conv.SendMessage, once that loop has already finished. Routing and conversation carry no Task.WhenAll, Task.Run or Parallel work, so the whole path is one logical thread. What the previous commit changed for real reasons stays: one write per frame instead of two, no 10ms delay between them, and an explicit flush rather than relying on Kestrel's buffering to release a frame. Co-Authored-By: Claude Opus 5 --- .../Conversation/ConversationController.cs | 16 ++-------------- 1 file changed, 2 insertions(+), 14 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 498edd64f..36e43c480 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -20,9 +20,6 @@ public partial class ConversationController : ControllerBase private const string StreamingFlag = "streaming"; - // Deltas arrive on the provider's context and would interleave with the message callback's frame write. - private readonly SemaphoreSlim _sseWriteLock = new(1, 1); - public ConversationController( IServiceProvider services, IUserIdentity user, @@ -568,17 +565,8 @@ private async Task OnChunkReceived(HttpResponse response, ChatResponseModel mess var json = JsonSerializer.Serialize(message, _jsonOptions); var buffer = Encoding.UTF8.GetBytes($"data:{json}\n\n"); - - await _sseWriteLock.WaitAsync(); - try - { - await response.Body.WriteAsync(buffer, 0, buffer.Length); - await response.Body.FlushAsync(); - } - finally - { - _sseWriteLock.Release(); - } + await response.Body.WriteAsync(buffer, 0, buffer.Length); + await response.Body.FlushAsync(); } private async Task OnEventCompleted(HttpResponse response) From a6abd1eb2d8fd7cd8c751c9ad66f39b023512a7e Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 15:58:28 +0800 Subject: [PATCH 3/5] Drop the redundant flush; WriteAsync already releases the frame Kestrel's response body is a PipeWriter, and writing to it makes the bytes available to the transport, so the explicit flush after it did nothing. Verified by removing it and streaming a reply against a local OneBrain: 67 frames still arrived over 1.75s across 35 distinct timestamps, the same shape as with the flush in place. Worth recording what this method still costs, because the flush was never the expensive part. A delta frame serializes the whole ChatResponseModel for a few characters of text: measured over a 1008-frame reply, 5,419 bytes of text went out as 607,122 bytes, and 323 of the 602 bytes per frame were an empty Sender object repeated every token. Trimming what a delta carries is where the gain is, not the flushing. Co-Authored-By: Claude Opus 5 --- .../Controllers/Conversation/ConversationController.cs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 36e43c480..abe6ab1ea 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -566,7 +566,6 @@ private async Task OnChunkReceived(HttpResponse response, ChatResponseModel mess var buffer = Encoding.UTF8.GetBytes($"data:{json}\n\n"); await response.Body.WriteAsync(buffer, 0, buffer.Length); - await response.Body.FlushAsync(); } private async Task OnEventCompleted(HttpResponse response) From de7e101bbf92f5ac353074b21ff51474ab9db7e2 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 16:22:08 +0800 Subject: [PATCH 4/5] Stop sending a whole reply's fields with every token A delta reused ChatResponseModel, so five characters of text went out as 602 bytes; 323 of them were an empty Sender object, repeated once per token. Over a 1008-frame reply that turned 5,419 bytes of text into 607,122 bytes. Deltas now carry only what a consumer needs to place them: conversation id, message id, the streaming marker and the text. Measured on the same prompt afterwards: 155 bytes per frame, amplification down from 112x to 29.3x. The message id has to stay -- consumers reject a non-indicating frame without one and would drop every delta. Half of what is left is the two ids repeated per frame. Dropping the conversation id would save another 30%, at the cost of a delta no longer looking like the other frame kinds. Co-Authored-By: Claude Opus 5 --- .../Conversation/ConversationController.cs | 30 +++++++++++++++---- 1 file changed, 25 insertions(+), 5 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index abe6ab1ea..c3db99e10 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -560,7 +560,7 @@ private FileContentResult BuildFileResult(string file) return File(bytes, "application/octet-stream", Path.GetFileName(file), enableRangeProcessing: enableRangeProcessing); } - private async Task OnChunkReceived(HttpResponse response, ChatResponseModel message) + private async Task OnChunkReceived(HttpResponse response, object message) { var json = JsonSerializer.Serialize(message, _jsonOptions); @@ -613,16 +613,36 @@ private async Task OnReceiveToolCallIndication(string conversationId, RoleDialog private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogModel msg) { - var delta = new ChatResponseModel + var delta = new StreamingDelta { ConversationId = conversationId, MessageId = msg.MessageId, - Text = !string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content, Function = StreamingFlag, - Thought = msg.Thought, - States = [] + Text = !string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content, + Thought = msg.Thought }; await OnChunkReceived(Response, delta); } + + /// + /// Serializing a full ChatResponseModel per token sent 602 bytes for 5 characters of text, 323 of them + /// an empty Sender repeated every token. message_id has to stay: consumers reject a non-indicating + /// frame without one. + /// + private sealed class StreamingDelta + { + [System.Text.Json.Serialization.JsonPropertyName("conversation_id")] + public string ConversationId { get; set; } = string.Empty; + + [System.Text.Json.Serialization.JsonPropertyName("message_id")] + public string MessageId { get; set; } = string.Empty; + + public string? Function { get; set; } + + public string Text { get; set; } = string.Empty; + + [System.Text.Json.Serialization.JsonIgnore(Condition = System.Text.Json.Serialization.JsonIgnoreCondition.WhenWritingNull)] + public Dictionary? Thought { get; set; } + } #endregion } \ No newline at end of file From 125d5f96d7d0472a0c57585bdfdfdd7cd153b328 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 17:42:37 +0800 Subject: [PATCH 5/5] Keep the SSE endpoint from pushing every token into SignalR too Both registered observers ran for every token. ConversationObserver reached the listener that writes the frame; ChatHubObserver answered the same event by building a DTO, serializing it, and awaiting a push to the SignalR group -- for a caller that is reading the response body and has no SignalR connection at all. With a Redis backplane configured, Clients.Group publishes whether or not the group has local members, so that was one Redis round trip per token, awaited synchronously on the completion loop's thread. The endpoint now names the observer it needs. SendMessage is untouched, and that is the endpoint BotSharp-UI posts to, so clients that do want events over SignalR are unaffected. The trade is that a conversation driven through /sse no longer feeds the SignalR hub, so watching the same conversation live in the UI while a caller drives it over SSE will show nothing. Picking one transport per conversation was already the assumption. Verified after the change: 82 delta frames still stream over 2.05s, 155 bytes each, and their concatenation still matches the closing frame's text. Co-Authored-By: Claude Opus 5 --- .../Controllers/Conversation/ConversationController.cs | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index c3db99e10..f17f27caf 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -456,7 +456,14 @@ await conv.SendMessage(agentId, inputMsg, public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input) { var observer = _services.GetRequiredService(); - using var container = observer.SubscribeObservers>(conversationId, listeners: new() + // ChatHubObserver is left out on purpose. It answers every event by serializing a DTO and pushing it + // to the SignalR group, which for a caller reading this response is work for nobody -- and with a + // Redis backplane configured that push leaves the process once per token. Callers who want events + // over SignalR post to SendMessage instead, where every observer still runs. + using var container = observer.SubscribeObservers>( + conversationId, + names: [nameof(BotSharp.Core.MessageHub.Observers.ConversationObserver)], + listeners: new() { { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(conversationId, data.Data) }, { ChatEvent.OnReceiveLlmStreamMessage, async data => await OnReceiveStreamingDelta(conversationId, data.Data) }