Improve dotnet test (MTP) output handling: negotiate protocol 1.1.0 and forward IOutputDevice output

src/Cli/dotnet/Commands/Test/CliConstants.cs

@@ -83,7 +83,7 @@ internal static class ProtocolConstants
     /// <summary>
     /// The protocol versions that are supported by the current SDK. Multiple versions can be present and be semicolon separated.
     /// </summary>
-    internal const string SupportedVersions = "1.0.0";
+    internal const string SupportedVersions = "1.0.0;1.1.0";
 }
 
 internal static class ProjectProperties

src/Cli/dotnet/Commands/Test/MTP/Terminal/TerminalTestReporter.cs

@@ -963,6 +963,12 @@ private static void AppendLongDuration(ITerminal terminal, TimeSpan duration, bo
     public void ArtifactAdded(bool outOfProcess, string? assembly, string? targetFramework, string? architecture, string? executionId, string? testName, string path)
         => _artifacts.Add(new TestRunArtifact(outOfProcess, assembly, targetFramework, architecture, executionId, testName, path));
 
+    internal void WriteMessage(string text) =>
+        _terminalWithProgress.WriteToTerminal(terminal =>
+        {
+            terminal.Append(text);
+        });
+
     /// <summary>
     /// Let the user know that cancellation was triggered.
     /// </summary>

src/Cli/dotnet/Commands/Test/MTP/TestApplication.cs

@@ -1,7 +1,6 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
-using System.Collections.Concurrent;
 using System.Diagnostics;
 using System.Globalization;
 using System.IO;
@@ -23,6 +22,9 @@ internal sealed class TestApplication(
     TerminalTestReporter output,
     Action<CommandLineOptionMessages> onHelpRequested) : IDisposable
 {
+    private static readonly Version ProtocolVersion_1_1 = new(1, 1, 0);
+    private const int LiveOutputTailLineCount = 200;
+
     private readonly Lock _requestLock = new();
     private readonly BuildOptions _buildOptions = buildOptions;
     private readonly Action<CommandLineOptionMessages> _onHelpRequested = onHelpRequested;
@@ -34,12 +36,20 @@ internal sealed class TestApplication(
     private readonly Dictionary<NamedPipeServer, HandshakeMessage> _handshakes = new();
 
     private int _hasRun;
+    private int _protocolNegotiated;
+    private Version? _negotiatedProtocolVersion;
+    private ProcessOutputCollector? _standardOutputCollector;
+    private ProcessOutputCollector? _standardErrorCollector;
 
     public TestModule Module { get; } = module;
     public TestOptions TestOptions { get; } = testOptions;
 
     public bool HasFailureDuringDispose { get; private set; }
 
+    internal bool IsProtocol_1_1_OrHigher =>
+        _negotiatedProtocolVersion is { } negotiatedProtocolVersion &&
+        negotiatedProtocolVersion.CompareTo(ProtocolVersion_1_1) >= 0;
+
     public async Task<int> RunAsync(CtrlCCancellationManager ctrlC)
     {
         if (Interlocked.Exchange(ref _hasRun, 1) != 0)
@@ -68,18 +78,20 @@ public async Task<int> RunAsync(CtrlCCancellationManager ctrlC)
             // Note: even with 'process.StandardOutput.ReadToEndAsync()' or 'process.BeginOutputReadLine()', we ended up with
             // many TP threads just doing synchronous IO, slowing down the progress of the test run.
             // We want to read requests coming through the pipe and sending responses back to the test app as fast as possible.
-            // We are using ConcurrentQueue to avoid thread-safety issues for the timeout case.
+            // The collector is thread-safe for the timeout case.
             // In the timeout case, we leave stdOutTask and stdErrTask running, just we stop observing them.
-            var stdOutBuilder = new ConcurrentQueue<string>();
-            var stdErrBuilder = new ConcurrentQueue<string>();
+            var stdOutBuilder = new ProcessOutputCollector(LiveOutputTailLineCount, _handler.WriteMessage);
+            var stdErrBuilder = new ProcessOutputCollector(LiveOutputTailLineCount, _handler.WriteMessage);
+            Volatile.Write(ref _standardOutputCollector, stdOutBuilder);
+            Volatile.Write(ref _standardErrorCollector, stdErrBuilder);
 
             var stdOutTask = Task.Factory.StartNew(() =>
             {
                 var stdOut = process.StandardOutput;
                 string? currentLine;
                 while ((currentLine = stdOut.ReadLine()) is not null)
                 {
-                    stdOutBuilder.Enqueue(currentLine);
+                    stdOutBuilder.AddLine(currentLine, GetLiveOutputStreamingState());
                 }
             }, TaskCreationOptions.LongRunning);
 
@@ -89,7 +101,7 @@ public async Task<int> RunAsync(CtrlCCancellationManager ctrlC)
                 string? currentLine;
                 while ((currentLine = stdErr.ReadLine()) is not null)
                 {
-                    stdErrBuilder.Enqueue(currentLine);
+                    stdErrBuilder.AddLine(currentLine, GetLiveOutputStreamingState());
                 }
             }, TaskCreationOptions.LongRunning);
 
@@ -108,7 +120,7 @@ public async Task<int> RunAsync(CtrlCCancellationManager ctrlC)
             }
 
             var exitCode = process.ExitCode;
-            _handler.OnTestProcessExited(exitCode, string.Join(Environment.NewLine, stdOutBuilder), string.Join(Environment.NewLine, stdErrBuilder));
+            _handler.OnTestProcessExited(exitCode, stdOutBuilder.GetOutput(), stdErrBuilder.GetOutput());
 
             // This condition is to prevent considering the test app as successful when we didn't receive test session end.
             // We don't produce the exception if the exit code is already non-zero to avoid surfacing this exception when there is already a known failure.
@@ -130,6 +142,9 @@ public async Task<int> RunAsync(CtrlCCancellationManager ctrlC)
                 process.Dispose();
             }
 
+            Volatile.Write(ref _standardOutputCollector, null);
+            Volatile.Write(ref _standardErrorCollector, null);
+
             cancellationTokenSource.Cancel();
             await testAppPipeConnectionLoop;
         }
@@ -277,6 +292,7 @@ private Task<IResponse> OnRequest(NamedPipeServer server, IRequest request)
                         // properties, mismatching info, ...) respond with an empty negotiated version so
                         // Microsoft.Testing.Platform stops sending further messages on this connection.
                         bool handshakeAccepted = OnHandshakeMessage(handshakeMessage, negotiatedVersion.Length > 0);
+                        SetNegotiatedProtocolVersion(handshakeAccepted ? negotiatedVersion : string.Empty);
                         return Task.FromResult((IResponse)CreateHandshakeMessage(handshakeAccepted ? negotiatedVersion : string.Empty));
 
                     case CommandLineOptionMessages commandLineOptionMessages:
@@ -333,7 +349,7 @@ private Task<IResponse> OnRequest(NamedPipeServer server, IRequest request)
         }
     }
 
-    private static string GetSupportedProtocolVersion(HandshakeMessage handshakeMessage)
+    internal static string GetSupportedProtocolVersion(HandshakeMessage handshakeMessage)
     {
         if (!handshakeMessage.Properties.TryGetValue(HandshakeMessagePropertyNames.SupportedProtocolVersions, out string? protocolVersions) ||
             string.IsNullOrWhiteSpace(protocolVersions))
@@ -344,18 +360,37 @@ private static string GetSupportedProtocolVersion(HandshakeMessage handshakeMess
             return string.Empty;
         }
 
-        // NOTE: Today, ProtocolConstants.Version is only 1.0.0 (i.e, SDK supports only a single version).
-        // Whenever we support multiple versions in SDK, we should do intersection
-        // between protocolVersions given by MTP, and the versions supported by SDK.
-        // Then we return the "highest" version from the intersection.
-        // The current logic **assumes** that ProtocolConstants.SupportedVersions is a single version.
-        if (protocolVersions.Split(";").Contains(ProtocolConstants.SupportedVersions))
+        List<(Version Version, string Text)> sdkSupportedVersions = [];
+        foreach (string supportedVersion in ProtocolConstants.SupportedVersions.Split(';'))
         {
-            return ProtocolConstants.SupportedVersions;
+            string trimmedSupportedVersion = supportedVersion.Trim();
+            if (Version.TryParse(trimmedSupportedVersion, out Version? parsedSupportedVersion))
+            {
+                sdkSupportedVersions.Add((parsedSupportedVersion, trimmedSupportedVersion));
+            }
+        }
+
+        Version? highestCommonVersion = null;
+        string highestCommonVersionText = string.Empty;
+        foreach (string advertisedVersion in protocolVersions.Split(';'))
+        {
+            if (!Version.TryParse(advertisedVersion.Trim(), out Version? parsedAdvertisedVersion))
+            {
+                continue;
+            }
+
+            foreach ((Version sdkSupportedVersion, string sdkSupportedVersionText) in sdkSupportedVersions)
+            {
+                if (parsedAdvertisedVersion.Equals(sdkSupportedVersion) &&
+                    (highestCommonVersion is null || sdkSupportedVersion.CompareTo(highestCommonVersion) > 0))
+                {
+                    highestCommonVersion = sdkSupportedVersion;
+                    highestCommonVersionText = sdkSupportedVersionText;
+                }
+            }
         }
 
-        // The version given by MTP is not supported by SDK.
-        return string.Empty;
+        return highestCommonVersionText;
     }
 
     private static HandshakeMessage CreateHandshakeMessage(string version) =>
@@ -368,6 +403,28 @@ private static HandshakeMessage CreateHandshakeMessage(string version) =>
             { HandshakeMessagePropertyNames.SupportedProtocolVersions, version }
         });
 
+    private void SetNegotiatedProtocolVersion(string negotiatedVersion)
+    {
+        if (Version.TryParse(negotiatedVersion, out Version? parsedNegotiatedVersion) &&
+            (_negotiatedProtocolVersion is null || parsedNegotiatedVersion.CompareTo(_negotiatedProtocolVersion) > 0))
+        {
+            _negotiatedProtocolVersion = parsedNegotiatedVersion;
+        }
+
+        Volatile.Write(ref _protocolNegotiated, 1);
+        FlushBufferedOutputIfLiveStreamingEnabled();
+    }
+
+    private bool? GetLiveOutputStreamingState() =>
+        Volatile.Read(ref _protocolNegotiated) == 0 ? null : IsProtocol_1_1_OrHigher;
+
+    private void FlushBufferedOutputIfLiveStreamingEnabled()
+    {
+        bool? liveOutputStreamingState = GetLiveOutputStreamingState();
+        Volatile.Read(ref _standardOutputCollector)?.FlushBufferedOutputIfLiveStreamingEnabled(liveOutputStreamingState);
+        Volatile.Read(ref _standardErrorCollector)?.FlushBufferedOutputIfLiveStreamingEnabled(liveOutputStreamingState);
+    }
+
     public bool OnHandshakeMessage(HandshakeMessage handshakeMessage, bool gotSupportedVersion)
         => _handler.OnHandshakeReceived(handshakeMessage, gotSupportedVersion);
 
@@ -396,6 +453,87 @@ private void OnTestInProgressMessages(TestInProgressMessages testInProgressMessa
     private void OnSessionEvent(TestSessionEvent sessionEvent)
         => _handler.OnSessionEventReceived(sessionEvent);
 
+    private sealed class ProcessOutputCollector(int liveOutputTailLineCount, Action<string> writeOutput)
+    {
+        private readonly object _lock = new();
+        private readonly Queue<string> _lines = [];
+        private bool _liveStreamingEnabled;
+
+        public void AddLine(string line, bool? liveOutputStreamingState)
+        {
+            string? outputToWrite = null;
+            lock (_lock)
+            {
+                _lines.Enqueue(line);
+                if (liveOutputStreamingState == true)
+                {
+                    if (_liveStreamingEnabled)
+                    {
+                        outputToWrite = line + Environment.NewLine;
+                    }
+                    else
+                    {
+                        _liveStreamingEnabled = true;
+                        outputToWrite = JoinLinesWithTrailingNewLine(_lines);
+                    }
+
+                    TrimToBoundedTail();
+                }
+            }
+
+            if (outputToWrite is not null)
+            {
+                writeOutput(outputToWrite);
+            }
+        }
+
+        public void FlushBufferedOutputIfLiveStreamingEnabled(bool? liveOutputStreamingState)
+        {
+            string? outputToWrite = null;
+            lock (_lock)
+            {
+                if (liveOutputStreamingState == true && !_liveStreamingEnabled)
+                {
+                    _liveStreamingEnabled = true;
+                    outputToWrite = JoinLinesWithTrailingNewLine(_lines);
+                    TrimToBoundedTail();
+                }
+            }
+
+            if (!string.IsNullOrEmpty(outputToWrite))
+            {
+                writeOutput(outputToWrite);
+            }
+        }
+
+        public string GetOutput()
+        {
+            lock (_lock)
+            {
+                return string.Join(Environment.NewLine, _lines);
+            }
+        }
+
+        private void TrimToBoundedTail()
+        {
+            while (_lines.Count > liveOutputTailLineCount)
+            {
+                _lines.Dequeue();
+            }
+        }
+
+        private static string JoinLinesWithTrailingNewLine(IEnumerable<string> lines)
+        {
+            StringBuilder builder = new();
+            foreach (string line in lines)
+            {
+                builder.AppendLine(line);
+            }
+
+            return builder.ToString();
+        }
+    }
+
     public override string ToString()
     {
         StringBuilder builder = new();

src/Cli/dotnet/Commands/Test/MTP/TestApplicationHandler.cs

@@ -463,6 +463,14 @@ internal bool HasMismatchingTestSessionEventCount()
         return false;
     }
 
+    internal void WriteMessage(string? text)
+    {
+        if (!string.IsNullOrEmpty(text))
+        {
+            _output.WriteMessage(text);
+        }
+    }
+
     internal void OnTestProcessExited(int exitCode, string outputData, string errorData)
     {
         if (_receivedTestHostHandshake && _handshakeInfo.HasValue)

test/dotnet.Tests/CommandTests/Test/TestApplicationProtocolVersionTests.cs

@@ -0,0 +1,33 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using Microsoft.DotNet.Cli.Commands.Test;
+using Microsoft.DotNet.Cli.Commands.Test.IPC.Models;
+
+namespace dotnet.Tests.CommandTests.Test;
+
+public class TestApplicationProtocolVersionTests
+{
+    [Theory]
+    [InlineData("1.0.0;1.1.0", "1.1.0")]
+    [InlineData("1.0.0", "1.0.0")]
+    [InlineData("1.1.0", "1.1.0")]
+    [InlineData("2.0.0", "")]
+    [InlineData("", "")]
+    [InlineData(null, "")]
+    [InlineData("0.9.0;1.0.0;9.9.9", "1.0.0")]
+    public void GetSupportedProtocolVersion_ReturnsHighestCommonVersion(string? advertisedVersions, string expectedVersion)
+    {
+        var properties = new Dictionary<byte, string>();
+        if (advertisedVersions is not null)
+        {
+            properties[HandshakeMessagePropertyNames.SupportedProtocolVersions] = advertisedVersions;
+        }
+
+        var handshake = new HandshakeMessage(properties);
+
+        string supportedVersion = TestApplication.GetSupportedProtocolVersion(handshake);
+
+        supportedVersion.Should().Be(expectedVersion);
+    }
+}