Feature/sse token streaming onebrain

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;
@@ -456,7 +458,14 @@ await conv.SendMessage(agentId, inputMsg,
     public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string conversationId, [FromBody] NewMessageModel input)
     {
         var observer = _services.GetRequiredService<IObserverService>();
-        using var container = observer.SubscribeObservers<HubObserveData<RoleDialogModel>>(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<HubObserveData<RoleDialogModel>>(
+            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) }
@@ -493,10 +502,18 @@ 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 = msg.IsStreaming
+                        ? string.Empty
+                        : (!string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content);
                     response.MessageLabel = msg.MessageLabel;
                     response.Function = msg.FunctionName;
-                    response.RichContent = msg.SecondaryRichContent ?? msg.RichContent;
+                    response.RichContent = msg.IsStreaming
+                        ? WithoutStreamedText(msg.SecondaryRichContent ?? msg.RichContent)
+                        : msg.SecondaryRichContent ?? msg.RichContent;
                     response.Instruction = msg.Instruction;
                     response.Data = msg.Data;
                     response.Thought = msg.Thought;
@@ -506,11 +523,30 @@ await conv.SendMessage(agentId, inputMsg,
                     await OnChunkReceived(Response, response);
                 });
 
-        response.States = state.GetStates();
-        response.MessageId = inputMsg.MessageId;
-        response.ConversationId = conversationId;
+        await OnEventCompleted(Response);
+    }
 
-        // await OnEventCompleted(Response);
+    /// <summary>
+    /// 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.
+    /// </summary>
+    private static RichContent<IRichMessage>? WithoutStreamedText(RichContent<IRichMessage>? content)
+    {
+        if (content?.Message is not TextMessage)
+        {
+            return content;
+        }
+
+        return new RichContent<IRichMessage>(new TextMessage(string.Empty))
+        {
+            Recipient = content.Recipient,
+            FillPostback = content.FillPostback,
+            Editor = content.Editor,
+            EditorAttributes = content.EditorAttributes
+        };
     }
 
     [HttpPost("/conversation/{conversationId}/stop-streaming")]
@@ -562,18 +598,21 @@ 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");
     }
 
+    /// <summary>
+    /// 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.
+    /// </summary>
     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");
+    private async Task WriteFrame(HttpResponse response, string frame)
+    {
+        var buffer = Encoding.UTF8.GetBytes(frame);
         await response.Body.WriteAsync(buffer, 0, buffer.Length);
     }
 

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<string>();
+        var observer = BuildObserver();
+        observer.SetEventListeners(new Dictionary<string, Func<HubObserveData<RoleDialogModel>, 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<string>();
+        var observer = BuildObserver();
+        observer.SetEventListeners(new Dictionary<string, Func<HubObserveData<RoleDialogModel>, 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<IConversationService>();
+        conversation.SetupGet(x => x.ConversationId).Returns("conversation-1");
+
+        var services = new ServiceCollection();
+        services.AddSingleton(conversation.Object);
+        services.AddSingleton(new Mock<IConversationStorage>().Object);
+        services.AddSingleton(new Mock<IRoutingContext>().Object);
+
+        return new ConversationObserver(services.BuildServiceProvider(), NullLogger<ConversationObserver>.Instance);
+    }
+
+    private static HubObserveData<RoleDialogModel> BuildEvent(string eventName, string content)
+    {
+        return new HubObserveData<RoleDialogModel>
+        {
+            EventName = eventName,
+            RefId = "conversation-1",
+            Data = new RoleDialogModel(AgentRole.Assistant, content) { Indication = content }
+        };
+    }
+}