Stream LLM token deltas over the conversation SSE endpoint

src/Infrastructure/BotSharp.Core/MessageHub/Observers/ConversationObserver.cs

@@ -38,10 +38,6 @@ public override void OnNext(HubObserveData<RoleDialogModel> 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<RoleDialogModel> 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();
+        }
     }
 }

src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs

@@ -18,6 +18,8 @@ public partial class ConversationController : ControllerBase
     private readonly IUserIdentity _user;
     private readonly JsonSerializerOptions _jsonOptions;
 
+    private const string StreamingFlag = "streaming";
+
     public ConversationController(
         IServiceProvider services,
         IUserIdentity user,
@@ -454,9 +456,17 @@ 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()
-        {
-            { ChatEvent.OnIndicationReceived, async data => await OnReceiveToolCallIndication(conversationId, data.Data) }
+        // 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) }
         });
 
         var conv = _services.GetRequiredService<IConversationService>();
@@ -557,15 +567,11 @@ 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);
 
-        var buffer = Encoding.UTF8.GetBytes($"data:{json}\n");
-        await response.Body.WriteAsync(buffer, 0, buffer.Length);
-        await Task.Delay(10);
-
-        buffer = Encoding.UTF8.GetBytes("\n");
+        var buffer = Encoding.UTF8.GetBytes($"data:{json}\n\n");
         await response.Body.WriteAsync(buffer, 0, buffer.Length);
     }
 
@@ -611,5 +617,39 @@ private async Task OnReceiveToolCallIndication(string conversationId, RoleDialog
         };
         await OnChunkReceived(Response, indicator);
     }
+
+    private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogModel msg)
+    {
+        var delta = new StreamingDelta
+        {
+            ConversationId = conversationId,
+            MessageId = msg.MessageId,
+            Function = StreamingFlag,
+            Text = !string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content,
+            Thought = msg.Thought
+        };
+        await OnChunkReceived(Response, delta);
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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<string, string?>? Thought { get; set; }
+    }
     #endregion
 }