diff --git a/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs b/src/Infrastructure/BotSharp.OpenAPI/Controllers/Conversation/ConversationController.cs index 51e348bb3..425f60dc5 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; @@ -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}")] @@ -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,7 +610,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); @@ -597,18 +618,16 @@ 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 }; - 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 } \ No newline at end of file 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 {