optimize sse

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

@@ -1,4 +1,4 @@
-using BotSharp.Abstraction.Files.Constants;
+using BotSharp.Abstraction.Files.Constants;
 using BotSharp.Abstraction.Files.Enums;
 using BotSharp.Abstraction.Infrastructures.Enums;
 using BotSharp.Abstraction.MessageHub.Models;
@@ -18,19 +18,23 @@ public partial class ConversationController : ControllerBase
     private readonly IServiceProvider _services;
     private readonly IUserIdentity _user;
     private readonly JsonSerializerOptions _jsonOptions;
+    private readonly ILogger<ConversationController> _logger;
 
     private const string StreamingFlag = "streaming";
     private const string DoneFlag = "done";
     private const string IndicatingFlag = "indicating";
+    private const string ErrorFlag = "error";
 
     public ConversationController(
         IServiceProvider services,
         IUserIdentity user,
-        BotSharpOptions options)
+        BotSharpOptions options,
+        ILogger<ConversationController> logger)
     {
         _services = services;
         _user = user;
         _jsonOptions = InitJsonOptions(options);
+        _logger = logger;
     }
 
     [HttpPost("/conversation/{agentId}")]
@@ -500,12 +504,10 @@ public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string
             convCancellation.RegisterConversation(conversationId);
         }
 
-        Response.StatusCode = 200;
-        Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.ContentType, "text/event-stream");
-        Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.CacheControl, "no-cache");
-        Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.Connection, "keep-alive");
+        PrepareSseResponse();
 
         var cancelled = false;
+        var failed = false;
         try
         {
             await conv.SendMessage(agentId, inputMsg,
@@ -523,23 +525,42 @@ await conv.SendMessage(agentId, inputMsg,
                         response.MetaData = msg.MetaData;
                         response.States = state.GetStates();
 
-                        await OnChunkReceived(Response, response);
+                        await WriteFrame(Response, response);
                     });
         }
         catch (OperationCanceledException) when (input.IsStreamingMessage)
         {
-            // The 200 and part of the stream are already on the wire, so this cannot surface as an error
-            // response. The done frame below reports the cancellation instead.
+            // Already committed to a 200, so the closing frame reports the cancellation instead.
             cancelled = true;
         }
+        catch (Exception ex)
+        {
+            // Not rethrown: that would reset the connection and lose the frames already written. The
+            // closing frame reports the failure; the reason stays here, where the exception is logged in full.
+            _logger.LogError(ex, $"Streaming conversation {conversationId} failed. {ex.Message}");
+            failed = true;
+        }
         finally
         {
             convCancellation?.UnregisterConversation(conversationId);
         }
 
-        // Nothing else in the stream marks the end of a response: without this frame a client cannot tell
-        // a finished reply from a dropped connection or a proxy timeout.
-        await OnEventCompleted(Response, conversationId, cancelled);
+        // Every response ends with this frame: without it a client cannot tell a finished reply from a
+        // dropped connection or a proxy timeout. Skipped when the client is already gone -- the write it
+        // would attempt sits outside the catch above, so a failure there has nothing to handle it.
+        if (!HttpContext.RequestAborted.IsCancellationRequested)
+        {
+            await OnEventCompleted(Response, conversationId, cancelled, failed);
+        }
+    }
+
+    private void PrepareSseResponse()
+    {
+        Response.ContentType = "text/event-stream";
+        Response.Headers[Microsoft.Net.Http.Headers.HeaderNames.CacheControl] = "no-cache";
+        // nginx buffers proxied responses by default, holding every frame back until the agent has
+        // finished. Connection is left unset: illegal on HTTP/2, and Kestrel manages it on HTTP/1.1.
+        Response.Headers["X-Accel-Buffering"] = "no";
     }
 
     [HttpPost("/conversation/{conversationId}/stop-streaming")]
@@ -589,26 +610,24 @@ private FileContentResult BuildFileResult(string file)
         return File(bytes, "application/octet-stream", Path.GetFileName(file), enableRangeProcessing: enableRangeProcessing);
     }
 
-    private async Task OnChunkReceived(HttpResponse response, object message)
+    private async Task WriteFrame(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);
     }
 
-    private async Task OnEventCompleted(HttpResponse response, string conversationId, bool cancelled)
+    private async Task OnEventCompleted(HttpResponse response, string conversationId, bool cancelled, bool failed)
     {
         var completion = new StreamingCompletion
         {
-            Function = DoneFlag,
+            Function = failed ? ErrorFlag : DoneFlag,
             ConversationId = conversationId,
             Cancelled = cancelled
         };
 
-        var json = JsonSerializer.Serialize(completion, _jsonOptions);
-        var buffer = Encoding.UTF8.GetBytes($"data:{json}\n\n");
-        await response.Body.WriteAsync(buffer, 0, buffer.Length);
+        await WriteFrame(response, completion);
     }
 
     private JsonSerializerOptions InitJsonOptions(BotSharpOptions options)
@@ -642,7 +661,7 @@ private async Task OnReceiveToolCallIndication(string conversationId, RoleDialog
             Instruction = msg.Instruction,
             States = []
         };
-        await OnChunkReceived(Response, indicator);
+        await WriteFrame(Response, indicator);
     }
 
     private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogModel msg)
@@ -655,7 +674,7 @@ private async Task OnReceiveStreamingDelta(string conversationId, RoleDialogMode
             Text = !string.IsNullOrEmpty(msg.SecondaryContent) ? msg.SecondaryContent : msg.Content,
             Thought = msg.Thought
         };
-        await OnChunkReceived(Response, delta);
+        await WriteFrame(Response, delta);
     }
     #endregion
 }

src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/Response/StreamingCompletion.cs

@@ -1,13 +1,11 @@
-using System.Text.Json.Serialization;
+using System.Text.Json.Serialization;
 
 namespace BotSharp.OpenAPI.ViewModels.Conversations;
 
 /// <summary>
 /// Payload of the terminating frame. Flagged in the body rather than with an SSE event name because
-/// consumers read this stream line by line and drop anything that is not a data: line.
-///
-/// Deliberately carries no message_id: consumers take any non-indicating frame that has one for a real
-/// agent reply, and would render this one as an empty message.
+/// consumers read this stream line by line and drop anything that is not a data: line. It carries no
+/// message_id on purpose: a consumer takes any non-indicating frame that has one for a real reply.
 /// </summary>
 public sealed class StreamingCompletion
 {