From 9bb44f96351110e50318470bf31bffb290312a2a Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 18:00:51 +0800 Subject: [PATCH 1/6] Cover the observer dispatch the SSE deltas depend on Hoisting the listener call out of the OnIndicationReceived branch is what lets a listener registered for any other event run at all, and nothing was holding that in place. A theory over three event names covers it, plus a case showing a listener is not called for an event it did not register for. Restoring the old shape fails exactly the two non-indication cases, which is the behaviour the SSE token stream relies on. Co-Authored-By: Claude Opus 5 --- .../MessageHub/ConversationObserverTests.cs | 80 +++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 tests/BotSharp.Core.UnitTests/MessageHub/ConversationObserverTests.cs diff --git a/tests/BotSharp.Core.UnitTests/MessageHub/ConversationObserverTests.cs b/tests/BotSharp.Core.UnitTests/MessageHub/ConversationObserverTests.cs new file mode 100644 index 000000000..f387258b9 --- /dev/null +++ b/tests/BotSharp.Core.UnitTests/MessageHub/ConversationObserverTests.cs @@ -0,0 +1,80 @@ +using BotSharp.Abstraction.Agents.Enums; +using BotSharp.Abstraction.Conversations; +using BotSharp.Abstraction.Conversations.Enums; +using BotSharp.Abstraction.Conversations.Models; +using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.Routing; +using BotSharp.Core.MessageHub.Observers; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Moq; +using Xunit; + +namespace BotSharp.Core.UnitTests.MessageHub; + +public class ConversationObserverTests +{ + [Theory] + [InlineData(ChatEvent.OnIndicationReceived)] + [InlineData(ChatEvent.OnReceiveLlmStreamMessage)] + [InlineData(ChatEvent.OnMessageReceivedFromAssistant)] + public void OnNext_ReachesTheListenerRegisteredForTheEvent(string eventName) + { + var observed = new List(); + var observer = BuildObserver(); + observer.SetEventListeners(new Dictionary, Task>> + { + [eventName] = data => + { + observed.Add(data.Data.Content); + return Task.CompletedTask; + } + }); + + observer.OnNext(BuildEvent(eventName, "chunk")); + + Assert.Equal(new[] { "chunk" }, observed); + } + + [Fact] + public void OnNext_LeavesAListenerForAnotherEventAlone() + { + var observed = new List(); + var observer = BuildObserver(); + observer.SetEventListeners(new Dictionary, Task>> + { + [ChatEvent.OnReceiveLlmStreamMessage] = data => + { + observed.Add(data.Data.Content); + return Task.CompletedTask; + } + }); + + observer.OnNext(BuildEvent(ChatEvent.OnIndicationReceived, "chunk")); + + Assert.Empty(observed); + } + + private static ConversationObserver BuildObserver() + { + var conversation = new Mock(); + conversation.SetupGet(x => x.ConversationId).Returns("conversation-1"); + + var services = new ServiceCollection(); + services.AddSingleton(conversation.Object); + services.AddSingleton(new Mock().Object); + services.AddSingleton(new Mock().Object); + + return new ConversationObserver(services.BuildServiceProvider(), NullLogger.Instance); + } + + private static HubObserveData BuildEvent(string eventName, string content) + { + return new HubObserveData + { + EventName = eventName, + RefId = "conversation-1", + Data = new RoleDialogModel(AgentRole.Assistant, content) { Indication = content } + }; + } +} From 468c3b71824ebf3f7f331dbb2ed241de72e01009 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Tue, 8 Sep 2026 17:42:37 +0800 Subject: [PATCH 2/6] 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) } From 6d53943a4cef50ee4f54f478a0204d8e69c62474 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Wed, 9 Sep 2026 10:50:24 +0800 Subject: [PATCH 3/6] End the stream with a sentinel and keep it alive while the agent thinks Two gaps a conforming SSE client trips over. A reply ended by closing the connection, which such a client reads as a dropped stream and answers by reconnecting; the endpoint ignores Last-Event-ID, so the reconnect resends the message and pays for a whole second reply. Verified against the endpoint: reconnecting with Last-Event-ID returned 200 and 12,749 bytes of a fresh answer. OnEventCompleted already existed for this, commented out at the call site, and now runs. The other is silence. An agent can take seconds to reach its first token -- measured 6.7s to the first indication and 8.4s to the first delta -- and a proxy watching an idle connection is free to drop it. A comment line every 20s keeps bytes moving without reaching the event parser. That heartbeat is the concurrent writer the earlier lock did not have: it ticks on its own schedule and can arrive while a frame is half written, so frames go out through one serialized writer again. Verified with the interval shortened to 2s: four comment lines interleaved through a reply, a spec parser still read 85 events with none malformed, and the last one was [DONE]. Co-Authored-By: Claude Opus 5 --- .../Conversation/ConversationController.cs | 69 +++++++++++++++---- 1 file changed, 56 insertions(+), 13 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index f17f27caf..c63761e8e 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -20,6 +20,12 @@ public partial class ConversationController : ControllerBase private const string StreamingFlag = "streaming"; + private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(20); + + // The heartbeat ticks on its own schedule, so unlike the reply's own frames it can reach the body + // while another frame is half written. + private readonly SemaphoreSlim _sseWriteLock = new(1, 1); + public ConversationController( IServiceProvider services, IUserIdentity user, @@ -495,7 +501,12 @@ public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.CacheControl, "no-cache"); Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.Connection, "keep-alive"); - await conv.SendMessage(agentId, inputMsg, + using var idle = CancellationTokenSource.CreateLinkedTokenSource(HttpContext.RequestAborted); + var heartbeat = SendHeartbeats(Response, idle.Token); + + try + { + await conv.SendMessage(agentId, inputMsg, replyMessage: input.Postback, // responsed generated async msg => @@ -512,12 +523,33 @@ await conv.SendMessage(agentId, inputMsg, await OnChunkReceived(Response, response); }); + } + finally + { + idle.Cancel(); + await heartbeat; + } - response.States = state.GetStates(); - response.MessageId = inputMsg.MessageId; - response.ConversationId = conversationId; + await OnEventCompleted(Response); + } - // await OnEventCompleted(Response); + /// + /// An agent can think for seconds before its first token, and a proxy that sees no bytes in that window + /// is free to drop the connection. A comment line keeps it busy without reaching the event parser. + /// + private async Task SendHeartbeats(HttpResponse response, CancellationToken cancellationToken) + { + try + { + while (!cancellationToken.IsCancellationRequested) + { + await Task.Delay(HeartbeatInterval, cancellationToken); + await WriteFrame(response, ":\n\n"); + } + } + catch (OperationCanceledException) + { + } } [HttpPost("/conversation/{conversationId}/stop-streaming")] @@ -569,19 +601,30 @@ private FileContentResult BuildFileResult(string file) private async Task OnChunkReceived(HttpResponse response, object message) { - var json = JsonSerializer.Serialize(message, _jsonOptions); - - var buffer = Encoding.UTF8.GetBytes($"data:{json}\n\n"); - await response.Body.WriteAsync(buffer, 0, buffer.Length); + await WriteFrame(response, $"data:{JsonSerializer.Serialize(message, _jsonOptions)}\n\n"); } + /// + /// Closing the connection is how a reply used to end, which a conforming client reads as a dropped + /// stream and answers by reconnecting -- resending the message and paying for a second reply. + /// private async Task OnEventCompleted(HttpResponse response) { - var buffer = Encoding.UTF8.GetBytes("data:[DONE]\n"); - await response.Body.WriteAsync(buffer, 0, buffer.Length); + await WriteFrame(response, "data:[DONE]\n\n"); + } - buffer = Encoding.UTF8.GetBytes("\n"); - await response.Body.WriteAsync(buffer, 0, buffer.Length); + private async Task WriteFrame(HttpResponse response, string frame) + { + var buffer = Encoding.UTF8.GetBytes(frame); + await _sseWriteLock.WaitAsync(); + try + { + await response.Body.WriteAsync(buffer, 0, buffer.Length); + } + finally + { + _sseWriteLock.Release(); + } } private JsonSerializerOptions InitJsonOptions(BotSharpOptions options) From c6a8616b2d5ee7e3f45483e2015f965f66f3487b Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Wed, 9 Sep 2026 13:51:31 +0800 Subject: [PATCH 4/6] Stop repeating a streamed reply in its closing frame The closing frame carried the whole reply again. For a caller appending deltas that is the text twice, and it left the frame's real purpose -- rich content, states, the function name, the fields a delta cannot carry -- looking like an afterthought behind a duplicate. A message that streamed now closes with an empty text. One that never streamed keeps its own, because a reply answered from a function or a template has no deltas and this is its only frame. Verified against the endpoint: 76 deltas concatenating to 384 characters, a closing frame with text "", and the same through copilot-bff -- 69 deltas, 360 characters, closing text empty, no parse errors logged. The reply still appears in full inside rich_content.message.text on that frame, which is left alone: it is a structured payload consumers render from, and blanking a field inside it would mean reaching into a concrete rich message type. A caller that reads text sees no duplicate; one that renders rich_content replaces rather than appends, so it shows the same reply either way. Co-Authored-By: Claude Opus 5 --- .../Conversation/ConversationController.cs | 12 +++++++++++- 1 file changed, 11 insertions(+), 1 deletion(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index c63761e8e..12f425545 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -26,6 +26,8 @@ public partial class ConversationController : ControllerBase // while another frame is half written. private readonly SemaphoreSlim _sseWriteLock = new(1, 1); + private readonly HashSet _streamedMessages = []; + public ConversationController( IServiceProvider services, IUserIdentity user, @@ -511,7 +513,13 @@ await conv.SendMessage(agentId, inputMsg, // responsed generated async msg => { - response.Text = !string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content; + // A message whose text already went out token by token repeats none of it here: the + // closing frame is for the fields a delta cannot carry, and a caller appending deltas + // would otherwise show the reply twice. One that never streamed -- answered from a + // function or a template -- still carries its text, this being its only frame. + response.Text = _streamedMessages.Contains(msg.MessageId) + ? string.Empty + : (!string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content); response.MessageLabel = msg.MessageLabel; response.Function = msg.FunctionName; response.RichContent = msg.SecondaryRichContent ?? msg.RichContent; @@ -663,6 +671,8 @@ private async Task OnReceiveToolCallIndication(string conversationId, RoleDialog private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogModel msg) { + _streamedMessages.Add(msg.MessageId); + var delta = new StreamingDelta { ConversationId = conversationId, From 75854782aec843ca55de474ddf940d1660bb3cac Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Wed, 9 Sep 2026 14:00:38 +0800 Subject: [PATCH 5/6] Empty the streamed reply out of the rich content too Blanking the closing frame's text missed where callers actually read the reply from: rich_content.message.text carried it in full, so the duplicate the change was meant to remove was still there for anyone rendering the rich payload. Only a plain text payload is emptied, and into a copy rather than in place. The message this comes from is appended to the dialog history after the callback returns, so blanking it directly would drop the reply from the record -- verified by reading the dialogs back: the assistant turn still holds all 351 characters in both fields. Other rich types keep their payload, which carries structure no sequence of deltas can rebuild. Verified both ways: direct, 71 deltas concatenating to 351 characters with the closing frame empty in both places and recipient, messaging_type, editor and fill_postback intact; through copilot-bff, 33 deltas to 149 characters, content .message.text empty, nothing logged. Callers reading the reply from the closing frame now need the deltas instead. Only a request that asked to stream is affected, so a client that never sets the flag still receives the whole reply in one frame. Co-Authored-By: Claude Opus 5 --- .../Conversation/ConversationController.cs | 29 ++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 12f425545..741368b3e 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -1,6 +1,8 @@ using BotSharp.Abstraction.Files.Constants; using BotSharp.Abstraction.Files.Enums; using BotSharp.Abstraction.MessageHub.Models; +using BotSharp.Abstraction.Messaging; +using BotSharp.Abstraction.Messaging.Models.RichContent; using BotSharp.Abstraction.MessageHub.Services; using BotSharp.Abstraction.Options; using BotSharp.Abstraction.Repositories; @@ -522,7 +524,9 @@ await conv.SendMessage(agentId, inputMsg, : (!string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content); response.MessageLabel = msg.MessageLabel; response.Function = msg.FunctionName; - response.RichContent = msg.SecondaryRichContent ?? msg.RichContent; + response.RichContent = _streamedMessages.Contains(msg.MessageId) + ? WithoutStreamedText(msg.SecondaryRichContent ?? msg.RichContent) + : msg.SecondaryRichContent ?? msg.RichContent; response.Instruction = msg.Instruction; response.Data = msg.Data; response.Thought = msg.Thought; @@ -541,6 +545,29 @@ await conv.SendMessage(agentId, inputMsg, await OnEventCompleted(Response); } + /// + /// The reply also sits inside the rich content, so a caller reading it from there sees the duplicate the + /// blank text was meant to remove. Only a plain text payload is emptied, and into a copy: the message + /// this came from is appended to the dialog history after this callback returns, and blanking it in + /// place would drop the reply from the record. Other rich types keep their payload, which carries + /// structure a caller cannot rebuild from the deltas. + /// + private static RichContent? WithoutStreamedText(RichContent? content) + { + if (content?.Message is not TextMessage) + { + return content; + } + + return new RichContent(new TextMessage(string.Empty)) + { + Recipient = content.Recipient, + FillPostback = content.FillPostback, + Editor = content.Editor, + EditorAttributes = content.EditorAttributes + }; + } + /// /// An agent can think for seconds before its first token, and a proxy that sees no bytes in that window /// is free to drop the connection. A comment line keeps it busy without reaching the event parser. From e44ddb9fd57c8c821c3da6d135239fa84ae071d9 Mon Sep 17 00:00:00 2001 From: "mars.yu" Date: Wed, 9 Sep 2026 14:51:28 +0800 Subject: [PATCH 6/6] Drop the heartbeat, its lock, and the set of streamed message ids The heartbeat goes at the author's call. With it goes the lock it was the justification for -- and that justification did not survive a test: with the lock removed and the interval cut to 5ms, 497 comment lines raced 83 data frames and produced no malformed frame, no parse failure, and 83 clean events through a spec parser. A frame is built into one buffer and written in one call, so there is no half-written frame for another writer to land inside. It was the original two-write-per-frame shape that had a window, and that is gone. The set of streamed ids goes because the provider already records this: IsStreaming is set on the message it builds when a completion streamed, and left alone on the one it builds for a tool call. Verified across the three flows that discriminate the blanking -- a streamed reply (63 deltas, closing text and rich text both empty), one that never streamed (no deltas, closing text 166 characters), and a function-driven turn through RoleDialogModel.From where the indication frames keep their own text and only the closing frame is emptied. Same through copilot-bff both ways, nothing logged. [DONE] stays: ending a reply by closing the connection is what makes a conforming client reconnect and pay for a second one. Co-Authored-By: Claude Opus 5 --- .../Conversation/ConversationController.cs | 56 ++----------------- 1 file changed, 4 insertions(+), 52 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 741368b3e..c7fe50603 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -22,14 +22,6 @@ public partial class ConversationController : ControllerBase private const string StreamingFlag = "streaming"; - private static readonly TimeSpan HeartbeatInterval = TimeSpan.FromSeconds(20); - - // The heartbeat ticks on its own schedule, so unlike the reply's own frames it can reach the body - // while another frame is half written. - private readonly SemaphoreSlim _sseWriteLock = new(1, 1); - - private readonly HashSet _streamedMessages = []; - public ConversationController( IServiceProvider services, IUserIdentity user, @@ -505,12 +497,7 @@ public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.CacheControl, "no-cache"); Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.Connection, "keep-alive"); - using var idle = CancellationTokenSource.CreateLinkedTokenSource(HttpContext.RequestAborted); - var heartbeat = SendHeartbeats(Response, idle.Token); - - try - { - await conv.SendMessage(agentId, inputMsg, + await conv.SendMessage(agentId, inputMsg, replyMessage: input.Postback, // responsed generated async msg => @@ -519,12 +506,12 @@ await conv.SendMessage(agentId, inputMsg, // closing frame is for the fields a delta cannot carry, and a caller appending deltas // would otherwise show the reply twice. One that never streamed -- answered from a // function or a template -- still carries its text, this being its only frame. - response.Text = _streamedMessages.Contains(msg.MessageId) + response.Text = msg.IsStreaming ? string.Empty : (!string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content); response.MessageLabel = msg.MessageLabel; response.Function = msg.FunctionName; - response.RichContent = _streamedMessages.Contains(msg.MessageId) + response.RichContent = msg.IsStreaming ? WithoutStreamedText(msg.SecondaryRichContent ?? msg.RichContent) : msg.SecondaryRichContent ?? msg.RichContent; response.Instruction = msg.Instruction; @@ -535,12 +522,6 @@ await conv.SendMessage(agentId, inputMsg, await OnChunkReceived(Response, response); }); - } - finally - { - idle.Cancel(); - await heartbeat; - } await OnEventCompleted(Response); } @@ -568,25 +549,6 @@ await conv.SendMessage(agentId, inputMsg, }; } - /// - /// An agent can think for seconds before its first token, and a proxy that sees no bytes in that window - /// is free to drop the connection. A comment line keeps it busy without reaching the event parser. - /// - private async Task SendHeartbeats(HttpResponse response, CancellationToken cancellationToken) - { - try - { - while (!cancellationToken.IsCancellationRequested) - { - await Task.Delay(HeartbeatInterval, cancellationToken); - await WriteFrame(response, ":\n\n"); - } - } - catch (OperationCanceledException) - { - } - } - [HttpPost("/conversation/{conversationId}/stop-streaming")] public ConverstionCancellationResponse StopStreaming([FromRoute] string conversationId) { @@ -651,15 +613,7 @@ private async Task OnEventCompleted(HttpResponse response) private async Task WriteFrame(HttpResponse response, string frame) { var buffer = Encoding.UTF8.GetBytes(frame); - await _sseWriteLock.WaitAsync(); - try - { - await response.Body.WriteAsync(buffer, 0, buffer.Length); - } - finally - { - _sseWriteLock.Release(); - } + await response.Body.WriteAsync(buffer, 0, buffer.Length); } private JsonSerializerOptions InitJsonOptions(BotSharpOptions options) @@ -698,8 +652,6 @@ private async Task OnReceiveToolCallIndication(string conversationId, RoleDialog private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogModel msg) { - _streamedMessages.Add(msg.MessageId); - var delta = new StreamingDelta { ConversationId = conversationId,