From 28a6916121ac01d50ac19dba05eb7b9364ed2173 Mon Sep 17 00:00:00 2001 From: Manjunath Janardhan Date: Fri, 18 Sep 2026 19:48:35 +0530 Subject: [PATCH 1/3] .NET: terminate chat completions SSE stream with data: [DONE] The agent chat completions endpoint ended its SSE stream by simply closing the connection. OpenAI-compatible clients treat a final `data: [DONE]` frame as the completion signal, so they waited for a timeout or hung instead. GetStreamingChunksAsync now yields a null-payload sentinel after the agent stream completes, and the item formatter writes it as `[DONE]`. Routing the terminator through SseFormatter keeps that component the single owner of the `data: ` prefix and the blank-line frame terminator. The sentinel is yielded after the loop, so it only follows a stream that ran to completion: if the agent throws or the caller aborts, the exception propagates out of the enumerator and no terminator is written. A client must never read a truncated stream as a complete one. The Responses endpoint is deliberately unchanged -- that API signals completion with typed `response.*` events, and the recorded trace for it contains no [DONE] frame. The existing streaming conformance test could not catch this: it asserts through ParseChatCompletionChunksFromSse, which skips the `[DONE]` line because it is not JSON. The new test asserts on the raw SSE body instead. Fixes #8526 --- .../AIAgentChatCompletionsProcessor.cs | 27 +++++++++++++- .../OpenAIChatCompletionsConformanceTests.cs | 37 +++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs index f7bfef9ce72..8109a5fa917 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs @@ -1,6 +1,7 @@ // Copyright (c) Microsoft. All rights reserved. using System; +using System.Buffers; using System.Collections.Generic; using System.Linq; using System.Net.ServerSentEvents; @@ -79,14 +80,27 @@ public Task ExecuteAsync(HttpContext httpContext) destination: response.Body, itemFormatter: (sseItem, bufferWriter) => { + var chunk = sseItem.Data; + + // A null payload is the end-of-stream sentinel from GetStreamingChunksAsync. + // It is written through the formatter rather than appended to the response body + // so that SseFormatter stays the single owner of the "data: " prefix and the + // blank-line frame terminator; duplicating that framing here would let the two + // representations drift apart. + if (chunk is null) + { + bufferWriter.Write("[DONE]"u8); + return; + } + using var writer = new Utf8JsonWriter(bufferWriter); - JsonSerializer.Serialize(writer, sseItem.Data, ChatCompletionsJsonContext.Default.ChatCompletionChunk); + JsonSerializer.Serialize(writer, chunk, ChatCompletionsJsonContext.Default.ChatCompletionChunk); writer.Flush(); }, cancellationToken); } - private async IAsyncEnumerable> GetStreamingChunksAsync([EnumeratorCancellation] CancellationToken cancellationToken = default) + private async IAsyncEnumerable> GetStreamingChunksAsync([EnumeratorCancellation] CancellationToken cancellationToken = default) { // The Unix timestamp (in seconds) of when the chat completion was created. Each chunk has the same timestamp. DateTimeOffset? createdAt = null; @@ -167,6 +181,15 @@ private async IAsyncEnumerable> GetStreamingChunksA yield return new(chunk); } + + // OpenAI-compatible clients detect completion from a final "data: [DONE]" frame and + // otherwise wait for the connection to drop, which reads as a hang or a timeout. + // + // Emitted after the loop so it only follows a stream that ran to completion: if the + // agent throws, or the caller aborts the request, the exception propagates out of the + // enumerator above and no terminator is written. A client must never be able to read a + // truncated stream as a complete one. + yield return new(null); } } } diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs index ad7e6410f83..72c9818a70f 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs @@ -230,6 +230,43 @@ public async Task StreamingRequestResponseAsync() Assert.NotEmpty(accumulatedText); } + [Fact] + public async Task StreamingResponseEndsWithDoneSentinelAsync() + { + // The recorded OpenAI trace for this endpoint ends with "data: [DONE]", and + // OpenAI-compatible clients use that frame to detect completion rather than waiting for + // the connection to drop. StreamingRequestResponseAsync above cannot catch a missing + // terminator: it asserts through ParseChatCompletionChunksFromSse, which skips the + // "[DONE]" line because it is not JSON. + + // Arrange + string requestJson = LoadChatCompletionsTraceFile("streaming/request.json"); + HttpClient client = await this.CreateTestServerAsync("done-sentinel-agent", "You are a helpful assistant.", "Hello there."); + + // Act + HttpResponseMessage httpResponse = await this.SendChatCompletionRequestAsync(client, "done-sentinel-agent", requestJson); + string responseSse = await httpResponse.Content.ReadAsStringAsync(); + + // Assert - the terminator is present, is the final frame, and is not duplicated. + // The body must end with the blank line that closes the frame: a client reading frame by + // frame never sees a "data: [DONE]" that is not followed by "\n\n", so asserting on a + // trimmed body would pass even if the final frame were left unterminated. + Assert.Equal("text/event-stream", httpResponse.Content.Headers.ContentType?.MediaType); + Assert.Contains("data: [DONE]", responseSse); + Assert.EndsWith("data: [DONE]\n\n", responseSse, System.StringComparison.Ordinal); + + var dataLines = responseSse.Split('\n') + .Select(line => line.TrimEnd('\r')) + .Where(line => line.StartsWith("data: ", System.StringComparison.Ordinal)) + .ToList(); + Assert.Single(dataLines, line => line == "data: [DONE]"); + + // Assert - the terminator follows the payload chunks rather than replacing them, so a + // client that stops reading at [DONE] still receives the whole completion. + Assert.True(dataLines.Count > 1, "the stream should carry chat completion chunks before the terminator"); + Assert.All(dataLines[..^1], line => Assert.StartsWith("data: {", line, System.StringComparison.Ordinal)); + } + [Fact] public async Task FunctionCallingRequestResponseAsync() { From 22402523eba3a86f57e8b2dd5a030f22894ff555 Mon Sep 17 00:00:00 2001 From: Manjunath Janardhan Date: Fri, 18 Sep 2026 21:09:04 +0530 Subject: [PATCH 2/3] =?UTF-8?q?.NET:=20address=20review=20=E2=80=94=20pin?= =?UTF-8?q?=20the=20failure-path=20invariant=20and=20correct=20the=20ratio?= =?UTF-8?q?nale=20comment?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review comments from Copilot on #8532. The sentinel comment asserted that clients "wait for the connection to drop, which reads as a hang or a timeout". Live testing disproved that: the response is chunked, so the zero-length terminating chunk ends the body definitively and the official OpenAI client completes without the sentinel. The comment now states the actual interoperability problem -- consumers that treat the sentinel rather than end-of-body as the completion signal cannot recognize the stream as finished -- and names the one this repo ships, SseResponseIdCapture. The failure-path invariant is now covered. An earlier attempt at this test was discarded because a client-side assertion has no teeth: when the agent throws, the server aborts and the client receives no body, so the test passed even against a try/finally refactor that writes the terminator unconditionally. This version tees the response body into a buffer through test middleware, so it observes what the server actually wrote. Verified to fail against that refactor. The failing chat client and the tee stream are private to the test file rather than added to the shared TestHelpers, keeping the change to one test file. --- .../AIAgentChatCompletionsProcessor.cs | 7 +- .../OpenAIChatCompletionsConformanceTests.cs | 149 +++++++++++++++++- 2 files changed, 149 insertions(+), 7 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs index 8109a5fa917..b70932b13d5 100644 --- a/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs +++ b/dotnet/src/Microsoft.Agents.AI.Hosting.OpenAI/ChatCompletions/AIAgentChatCompletionsProcessor.cs @@ -182,8 +182,11 @@ public Task ExecuteAsync(HttpContext httpContext) yield return new(chunk); } - // OpenAI-compatible clients detect completion from a final "data: [DONE]" frame and - // otherwise wait for the connection to drop, which reads as a hang or a timeout. + // The OpenAI wire format ends a streaming chat completion with a literal + // "data: [DONE]" frame, and consumers that treat that sentinel -- rather than + // end-of-body -- as the completion signal cannot recognize the stream as finished + // without it. This repo ships one such consumer: SseResponseIdCapture in the DevUI + // Aspire integration matches "[DONE]"u8 explicitly. // // Emitted after the loop so it only follows a stream that ran to completion: if the // agent throws, or the caller aborts the request, the exception propagates out of the diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs index 72c9818a70f..694149ba356 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs @@ -1,12 +1,23 @@ // Copyright (c) Microsoft. All rights reserved. +using System; using System.Collections.Generic; +using System.IO; using System.Linq; using System.Net.Http; +using System.Runtime.CompilerServices; +using System.Text; using System.Text.Json; +using System.Threading; using System.Threading.Tasks; +using Microsoft.Agents.AI; using Microsoft.Agents.AI.Hosting.OpenAI.Tests; +using Microsoft.AspNetCore.Builder; +using Microsoft.AspNetCore.Hosting.Server; +using Microsoft.AspNetCore.TestHost; using Microsoft.Extensions.AI; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; namespace Microsoft.Agents.AI.Hosting.OpenAI.UnitTests; @@ -253,18 +264,18 @@ public async Task StreamingResponseEndsWithDoneSentinelAsync() // trimmed body would pass even if the final frame were left unterminated. Assert.Equal("text/event-stream", httpResponse.Content.Headers.ContentType?.MediaType); Assert.Contains("data: [DONE]", responseSse); - Assert.EndsWith("data: [DONE]\n\n", responseSse, System.StringComparison.Ordinal); + Assert.EndsWith("data: [DONE]\n\n", responseSse, StringComparison.Ordinal); var dataLines = responseSse.Split('\n') .Select(line => line.TrimEnd('\r')) - .Where(line => line.StartsWith("data: ", System.StringComparison.Ordinal)) + .Where(line => line.StartsWith("data: ", StringComparison.Ordinal)) .ToList(); Assert.Single(dataLines, line => line == "data: [DONE]"); // Assert - the terminator follows the payload chunks rather than replacing them, so a // client that stops reading at [DONE] still receives the whole completion. Assert.True(dataLines.Count > 1, "the stream should carry chat completion chunks before the terminator"); - Assert.All(dataLines[..^1], line => Assert.StartsWith("data: {", line, System.StringComparison.Ordinal)); + Assert.All(dataLines[..^1], line => Assert.StartsWith("data: {", line, StringComparison.Ordinal)); } [Fact] @@ -380,7 +391,7 @@ public async Task SystemMessageRequestResponseAsync() AssertJsonPropertyEquals(systemMessage, "role", "system"); AssertJsonPropertyExists(systemMessage, "content"); string systemContent = systemMessage.GetProperty("content").GetString()!; - Assert.Contains("pirate", systemContent, System.StringComparison.OrdinalIgnoreCase); + Assert.Contains("pirate", systemContent, StringComparison.OrdinalIgnoreCase); var userMessage = messages[1]; AssertJsonPropertyEquals(userMessage, "role", "user"); @@ -635,7 +646,7 @@ private static List ParseChatCompletionChunksFromSse(string sseCont { var line = lines[i].TrimEnd('\r'); - if (line.StartsWith("data: ", System.StringComparison.Ordinal)) + if (line.StartsWith("data: ", StringComparison.Ordinal)) { var jsonData = line.Substring("data: ".Length); @@ -659,4 +670,132 @@ private static List ParseChatCompletionChunksFromSse(string sseCont return chunks; } + + [Fact] + public async Task StreamingResponseOmitsDoneSentinelWhenAgentStreamFailsAsync() + { + // The sentinel must only ever follow a stream that ran to completion; a truncated response + // must never be marked complete. Asserting that from the client does not work: when the + // agent throws, the server aborts and the client receives no body at all, so a client-side + // check passes even against a try/finally refactor that writes the terminator + // unconditionally. This captures what the server actually wrote to the response body, which + // does distinguish the two. + + // Arrange - a host whose response body is tee'd into a buffer we can inspect afterwards + var recorded = new MemoryStream(); + using var failingChatClient = new FailingStreamChatClient(updatesBeforeFailure: 2); + + WebApplicationBuilder builder = WebApplication.CreateBuilder(); + builder.WebHost.UseTestServer(); + builder.Services.AddKeyedSingleton("chat-client", failingChatClient); + builder.AddAIAgent("failing-agent", "You are a helpful assistant.", chatClientServiceKey: "chat-client"); + builder.AddOpenAIChatCompletions(); + + await using WebApplication app = builder.Build(); + app.Use(async (context, next) => + { + Stream original = context.Response.Body; + context.Response.Body = new TeeStream(original, recorded); + try + { + await next(context); + } + finally + { + context.Response.Body = original; + } + }); + AIAgent agent = app.Services.GetRequiredKeyedService("failing-agent"); + app.MapOpenAIChatCompletions(agent, path: null, PermissiveMapOptions.ChatCompletions()); + await app.StartAsync(); + + TestServer testServer = (TestServer)app.Services.GetRequiredService(); + using HttpClient client = testServer.CreateClient(); + + // Act - the agent yields two updates and then throws partway through the body + try + { + using HttpResponseMessage response = await this.SendChatCompletionRequestAsync(client, "failing-agent", LoadChatCompletionsTraceFile("streaming/request.json")); + _ = await response.Content.ReadAsStringAsync(); + } + catch (Exception ex) when (ex is HttpRequestException or IOException or InvalidOperationException) + { + // Expected: the server aborts the response rather than completing it. + } + + await app.StopAsync(); + + // Assert - the server wrote payload frames but never the terminator + string written = Encoding.UTF8.GetString(recorded.ToArray()); + Assert.Contains("chat.completion.chunk", written); + Assert.DoesNotContain("[DONE]", written); + } + + /// A chat client whose streaming response yields a few updates and then throws. + private sealed class FailingStreamChatClient(int updatesBeforeFailure) : IChatClient + { + public Task GetResponseAsync(IEnumerable messages, ChatOptions? options = null, CancellationToken cancellationToken = default) + => throw new InvalidOperationException("mock streaming failure"); + + public async IAsyncEnumerable GetStreamingResponseAsync( + IEnumerable messages, + ChatOptions? options = null, + [EnumeratorCancellation] CancellationToken cancellationToken = default) + { + for (int i = 0; i < updatesBeforeFailure; i++) + { + await Task.Delay(1, cancellationToken); + yield return new ChatResponseUpdate { Contents = [new TextContent($"chunk{i} ")], Role = ChatRole.Assistant }; + } + + throw new InvalidOperationException("mock streaming failure"); + } + + public object? GetService(Type serviceType, object? serviceKey = null) => serviceType.IsInstanceOfType(this) ? this : null; + + public void Dispose() + { + } + } + + /// + /// Forwards writes to the real response body while copying them into a buffer. The copy happens + /// first so that bytes the server attempted to write are recorded even if the real write fails + /// because the response is already being torn down. + /// + private sealed class TeeStream(Stream inner, Stream capture) : Stream + { + public override bool CanRead => false; + public override bool CanSeek => false; + public override bool CanWrite => true; + public override long Length => throw new NotSupportedException(); + public override long Position { get => throw new NotSupportedException(); set => throw new NotSupportedException(); } + + public override void Write(byte[] buffer, int offset, int count) + { + capture.Write(buffer, offset, count); + inner.Write(buffer, offset, count); + } + + public override void Write(ReadOnlySpan buffer) + { + capture.Write(buffer); + inner.Write(buffer); + } + + public override async ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + await capture.WriteAsync(buffer, cancellationToken).ConfigureAwait(false); + await inner.WriteAsync(buffer, cancellationToken).ConfigureAwait(false); + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => this.WriteAsync(buffer.AsMemory(offset, count), cancellationToken).AsTask(); + + public override void Flush() => inner.Flush(); + public override Task FlushAsync(CancellationToken cancellationToken) => inner.FlushAsync(cancellationToken); + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + public override void SetLength(long value) => throw new NotSupportedException(); + } } From 7b749b498f2f97906358c192d894aedaa178657e Mon Sep 17 00:00:00 2001 From: Manjunath Janardhan Date: Fri, 18 Sep 2026 21:33:31 +0530 Subject: [PATCH 3/3] .NET: tidy the failure-path test after a review pass Review findings on the test added in 22402523e. No behaviour change; the test still fails against the try/finally append refactor it exists to block. - Dispose the capture buffer (`using var recorded`), which was leaked. - Hoist the trace load into Arrange, matching the Arrange/Act/Assert split the rest of the file uses; it was sitting in the Act block. - Record why the host is built inline instead of through ConformanceTestBase.CreateTestServerAsync: this test needs response-body middleware and a throwing chat client, neither of which belongs in the shared harness for one caller. Without the note the duplication reads as an oversight and invites a refactor back into the base class. --- .../OpenAIChatCompletionsConformanceTests.cs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs index 694149ba356..3f6c9fdc30f 100644 --- a/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs +++ b/dotnet/tests/Microsoft.Agents.AI.Hosting.OpenAI.UnitTests/OpenAIChatCompletionsConformanceTests.cs @@ -681,8 +681,13 @@ public async Task StreamingResponseOmitsDoneSentinelWhenAgentStreamFailsAsync() // unconditionally. This captures what the server actually wrote to the response body, which // does distinguish the two. - // Arrange - a host whose response body is tee'd into a buffer we can inspect afterwards - var recorded = new MemoryStream(); + // Arrange - a host whose response body is tee'd into a buffer we can inspect afterwards. + // The host is built here rather than through ConformanceTestBase.CreateTestServerAsync + // because this test needs response-body middleware and a throwing chat client, and neither + // belongs in the shared harness for a single caller. If the base class's wiring changes, + // this setup does not follow it automatically -- keep them in step deliberately. + string requestJson = LoadChatCompletionsTraceFile("streaming/request.json"); + using var recorded = new MemoryStream(); using var failingChatClient = new FailingStreamChatClient(updatesBeforeFailure: 2); WebApplicationBuilder builder = WebApplication.CreateBuilder(); @@ -715,7 +720,7 @@ public async Task StreamingResponseOmitsDoneSentinelWhenAgentStreamFailsAsync() // Act - the agent yields two updates and then throws partway through the body try { - using HttpResponseMessage response = await this.SendChatCompletionRequestAsync(client, "failing-agent", LoadChatCompletionsTraceFile("streaming/request.json")); + using HttpResponseMessage response = await this.SendChatCompletionRequestAsync(client, "failing-agent", requestJson); _ = await response.Content.ReadAsStringAsync(); } catch (Exception ex) when (ex is HttpRequestException or IOException or InvalidOperationException)