diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index c3db99e10..c7fe50603 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; @@ -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(); - 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) } @@ -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); + /// + /// 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 + }; } [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"); } + /// + /// 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"); + private async Task WriteFrame(HttpResponse response, string frame) + { + var buffer = Encoding.UTF8.GetBytes(frame); await response.Body.WriteAsync(buffer, 0, buffer.Length); } 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 } + }; + } +}