From 2d506ca4efdeda512bfd7ae9a69065a53c4c6145 Mon Sep 17 00:00:00 2001 From: "nick.yi" Date: Thu, 10 Sep 2026 14:21:38 +0800 Subject: [PATCH 1/4] GTR-13451 --- .../Conversation/ConversationController.cs | 27 +++++++++++++------ .../Response/StreamingCompletion.cs | 8 +++--- 2 files changed, 22 insertions(+), 13 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 51e348bb3..97d80a19c 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -18,19 +18,23 @@ public partial class ConversationController : ControllerBase private readonly IServiceProvider _services; private readonly IUserIdentity _user; private readonly JsonSerializerOptions _jsonOptions; + private readonly ILogger _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 logger) { _services = services; _user = user; _jsonOptions = InitJsonOptions(options); + _logger = logger; } [HttpPost("/conversation/{agentId}")] @@ -506,6 +510,7 @@ public async Task SendMessageSse([FromRoute] string agentId, [FromRoute] string Response.Headers.Append(Microsoft.Net.Http.Headers.HeaderNames.Connection, "keep-alive"); var cancelled = false; + var failed = false; try { await conv.SendMessage(agentId, inputMsg, @@ -528,18 +533,24 @@ await conv.SendMessage(agentId, inputMsg, } 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. + await OnEventCompleted(Response, conversationId, cancelled, failed); } [HttpPost("/conversation/{conversationId}/stop-streaming")] @@ -597,11 +608,11 @@ private async Task OnChunkReceived(HttpResponse response, object message) 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 }; diff --git a/src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/Response/StreamingCompletion.cs b/src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/Response/StreamingCompletion.cs index 9c48798f2..706c51136 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/ViewModels/Conversations/Response/StreamingCompletion.cs +++ b/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; /// /// 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. /// public sealed class StreamingCompletion { From 3d87a76bbf7d910e040e5c749390acd03aa361b5 Mon Sep 17 00:00:00 2001 From: "nick.yi" Date: Thu, 10 Sep 2026 14:44:22 +0800 Subject: [PATCH 2/4] GTR-13451 --- .../Controllers/Conversation/ConversationController.cs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 97d80a19c..bd64979e6 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -549,8 +549,12 @@ await conv.SendMessage(agentId, inputMsg, } // Every response ends with this frame: without it a client cannot tell a finished reply from a - // dropped connection or a proxy timeout. - await OnEventCompleted(Response, conversationId, cancelled, failed); + // 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); + } } [HttpPost("/conversation/{conversationId}/stop-streaming")] From c16e9c2ca100183558d31071ceb9e63221a28768 Mon Sep 17 00:00:00 2001 From: "nick.yi" Date: Thu, 10 Sep 2026 14:52:06 +0800 Subject: [PATCH 3/4] optimize sse --- .../Conversation/ConversationController.cs | 14 ++++++-------- 1 file changed, 6 insertions(+), 8 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index bd64979e6..85b228c24 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/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; @@ -528,7 +528,7 @@ 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) @@ -604,7 +604,7 @@ 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); @@ -621,9 +621,7 @@ private async Task OnEventCompleted(HttpResponse response, string 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) @@ -657,7 +655,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) @@ -670,7 +668,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 } \ No newline at end of file From 2ef9bc8cf39ad82fd7f9ecb34a34057f35f89156 Mon Sep 17 00:00:00 2001 From: "nick.yi" Date: Thu, 10 Sep 2026 15:21:20 +0800 Subject: [PATCH 4/4] optimize sse response --- .../Conversation/ConversationController.cs | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 85b228c24..425f60dc5 100644 --- a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs +++ b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs @@ -504,10 +504,7 @@ 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; @@ -557,6 +554,15 @@ await conv.SendMessage(agentId, inputMsg, } } + 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")] public ConverstionCancellationResponse StopStreaming([FromRoute] string conversationId) {