From 8836aa63bedac588d58e4e45bc75ccc97e1fa6c7 Mon Sep 17 00:00:00 2001 From: waldekmastykarz Date: Sat, 31 Oct 2026 09:59:50 +0100 Subject: [PATCH 1/3] Frame stdio stdin by newline so plugins see one message at a time. Closes #1757 JSON-RPC messages written back-to-back were read as a single chunk, so MockStdioResponsePlugin couldn't parse the body as JSON and left @stdin.body.id unresolved. Stdin is now split on newlines (partial lines are buffered until the newline arrives or stdin closes) and each message runs through the plugin pipeline individually. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../MockStdioResponsePluginTests.cs | 25 ++++ DevProxy.Tests/StdioLineFramerTests.cs | 79 ++++++++++++ DevProxy/Stdio/ProxySession.cs | 112 ++++++++++-------- DevProxy/Stdio/StdioLineFramer.cs | 68 +++++++++++ 4 files changed, 235 insertions(+), 49 deletions(-) create mode 100644 DevProxy.Tests/StdioLineFramerTests.cs create mode 100644 DevProxy/Stdio/StdioLineFramer.cs diff --git a/DevProxy.Integration.Tests/MockStdioResponsePluginTests.cs b/DevProxy.Integration.Tests/MockStdioResponsePluginTests.cs index 0dc20607b..600949b5d 100644 --- a/DevProxy.Integration.Tests/MockStdioResponsePluginTests.cs +++ b/DevProxy.Integration.Tests/MockStdioResponsePluginTests.cs @@ -85,6 +85,31 @@ public async Task Nth_EffectiveMatcher_SharesFallbackCount(string? bodyRegex, st } } + [Fact] + public async Task StdinBodyIdPlaceholder_NewlineTerminatedJsonRpcMessage_IsReplaced() + { + var config = PluginConfig.FromJson(""" + { + "mocks": [ + { + "request": { "bodyFragment": "accept_eula" }, + "response": { "stdout": "{\"jsonrpc\":\"2.0\",\"id\":@stdin.body.id,\"result\":{}}\n" } + } + ] + } + """); + using var httpClient = new HttpClient(); + using var plugin = new MockStdioResponsePlugin( + httpClient, NullLogger.Instance, new HashSet(), + new TestProxyConfiguration(), config); + + var request = CreateRequest("{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\",\"params\":{\"name\":\"accept_eula\"}}\n"); + await plugin.BeforeStdinAsync(request, CancellationToken.None); + + Assert.True(request.ResponseState.HasBeenSet); + Assert.Equal("{\"jsonrpc\":\"2.0\",\"id\":2,\"result\":{}}\n", request.StdoutResponse); + } + private static StdioRequestArgs CreateRequest(string body) => new() { Body = Encoding.UTF8.GetBytes(body), diff --git a/DevProxy.Tests/StdioLineFramerTests.cs b/DevProxy.Tests/StdioLineFramerTests.cs new file mode 100644 index 000000000..b3e3a7d29 --- /dev/null +++ b/DevProxy.Tests/StdioLineFramerTests.cs @@ -0,0 +1,79 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +using System.Text; +using DevProxy.Stdio; +using Xunit; + +namespace DevProxy.Tests; + +public class StdioLineFramerTests +{ + private static string[] AsStrings(IEnumerable messages) => + [.. messages.Select(m => Encoding.UTF8.GetString(m))]; + + [Fact] + public void Append_MultipleLinesInOneChunk_ReturnsEachLine() + { + var framer = new StdioLineFramer(); + + var messages = framer.Append(Encoding.UTF8.GetBytes( + "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n{\"jsonrpc\":\"2.0\",\"id\":2}\n")); + + Assert.Equal( + ["{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n", "{\"jsonrpc\":\"2.0\",\"id\":2}\n"], + AsStrings(messages)); + Assert.Null(framer.Flush()); + } + + [Fact] + public void Append_LineSplitAcrossChunks_ReturnsLineOnceComplete() + { + var framer = new StdioLineFramer(); + + Assert.Empty(framer.Append(Encoding.UTF8.GetBytes("{\"id\":"))); + Assert.Empty(framer.Append(Encoding.UTF8.GetBytes("1"))); + var messages = framer.Append(Encoding.UTF8.GetBytes("}\n{\"id\":2")); + + Assert.Equal(["{\"id\":1}\n"], AsStrings(messages)); + Assert.Equal("{\"id\":2", Encoding.UTF8.GetString(framer.Flush()!)); + Assert.Null(framer.Flush()); + } + + [Fact] + public void Append_CrLf_KeepsLineEndingIntact() + { + var framer = new StdioLineFramer(); + + var messages = framer.Append(Encoding.UTF8.GetBytes("a\r\nb\r\n")); + + Assert.Equal(["a\r\n", "b\r\n"], AsStrings(messages)); + } + + [Fact] + public void Append_EmptyInput_ReturnsNothing() + { + var framer = new StdioLineFramer(); + + Assert.Empty(framer.Append([])); + Assert.Null(framer.Flush()); + } + + [Fact] + public void Append_LineLargerThanReadBuffer_IsReturnedAsSingleMessage() + { + var framer = new StdioLineFramer(); + var line = new string('x', 10_000) + "\n"; + var bytes = Encoding.UTF8.GetBytes(line); + + var messages = new List(); + for (var offset = 0; offset < bytes.Length; offset += 4096) + { + var length = Math.Min(4096, bytes.Length - offset); + messages.AddRange(framer.Append(bytes.AsSpan(offset, length))); + } + + Assert.Equal([line], AsStrings(messages)); + } +} diff --git a/DevProxy/Stdio/ProxySession.cs b/DevProxy/Stdio/ProxySession.cs index fe091d230..4934dbcaf 100644 --- a/DevProxy/Stdio/ProxySession.cs +++ b/DevProxy/Stdio/ProxySession.cs @@ -196,6 +196,7 @@ public async Task RunAsync() private async Task ForwardStdinAsync() { var buffer = new byte[4096]; + var framer = new StdioLineFramer(); using var stdin = Console.OpenStandardInput(); try @@ -228,60 +229,19 @@ private async Task ForwardStdinAsync() var bytesRead = await readTask; if (bytesRead == 0) { - break; - } - - var data = buffer[..bytesRead]; - Log("STDIN >>>", buffer, bytesRead); - - // Log intercepted stdin (similar to HTTP InterceptedRequest) - await LogInterceptedStdioAsync(data, StdioMessageDirection.Stdin); - - // Execute plugins and check if message was consumed - var responseState = new ResponseState(); - var requestArgs = new StdioRequestArgs - { - Body = data, - ResponseState = responseState, - Session = _stdioSession!, - SessionData = _sessionData, - GlobalData = _globalData - }; - - await ExecuteBeforeStdinPluginsAsync(requestArgs); - - // Only forward to child if not consumed - if (!responseState.HasBeenSet) - { - await _writeSemaphore.WaitAsync(_cts.Token); - try - { - await _childStdin!.WriteAsync(buffer.AsMemory(0, bytesRead), _cts.Token); - await _childStdin.FlushAsync(_cts.Token); - } - finally + // Process any trailing data that wasn't terminated by a newline + var remainder = framer.Flush(); + if (remainder is not null) { - _ = _writeSemaphore.Release(); + await ProcessStdinMessageAsync(remainder); } - } - else - { - // Send mock response if set by a plugin - await SendMockResponseAsync(requestArgs); + break; } - // Create request log entry for this stdin message - var requestLog = new StdioRequestLog + foreach (var message in framer.Append(buffer.AsSpan(0, bytesRead))) { - Command = _stdioSession!.Command, - StdinBody = data, - StdinTimestamp = DateTimeOffset.UtcNow, - MessageType = responseState.HasBeenSet ? MessageType.Mocked : MessageType.PassedThrough - }; - _requestLogs.Add(requestLog); - - // Notify plugins of the request log - await NotifyStdioRequestLogAsync(requestLog); + await ProcessStdinMessageAsync(message); + } } } catch (OperationCanceledException) @@ -305,6 +265,60 @@ private async Task ForwardStdinAsync() } } + private async Task ProcessStdinMessageAsync(byte[] message) + { + Log("STDIN >>>", message, message.Length); + + // Log intercepted stdin (similar to HTTP InterceptedRequest) + await LogInterceptedStdioAsync(message, StdioMessageDirection.Stdin); + + // Execute plugins and check if message was consumed + var responseState = new ResponseState(); + var requestArgs = new StdioRequestArgs + { + Body = message, + ResponseState = responseState, + Session = _stdioSession!, + SessionData = _sessionData, + GlobalData = _globalData + }; + + await ExecuteBeforeStdinPluginsAsync(requestArgs); + + // Only forward to child if not consumed + if (!responseState.HasBeenSet) + { + await _writeSemaphore.WaitAsync(_cts.Token); + try + { + await _childStdin!.WriteAsync(message, _cts.Token); + await _childStdin.FlushAsync(_cts.Token); + } + finally + { + _ = _writeSemaphore.Release(); + } + } + else + { + // Send mock response if set by a plugin + await SendMockResponseAsync(requestArgs); + } + + // Create request log entry for this stdin message + var requestLog = new StdioRequestLog + { + Command = _stdioSession!.Command, + StdinBody = message, + StdinTimestamp = DateTimeOffset.UtcNow, + MessageType = responseState.HasBeenSet ? MessageType.Mocked : MessageType.PassedThrough + }; + _requestLogs.Add(requestLog); + + // Notify plugins of the request log + await NotifyStdioRequestLogAsync(requestLog); + } + private async Task ForwardStdoutAsync() { var buffer = new byte[4096]; diff --git a/DevProxy/Stdio/StdioLineFramer.cs b/DevProxy/Stdio/StdioLineFramer.cs new file mode 100644 index 000000000..01a0e8a50 --- /dev/null +++ b/DevProxy/Stdio/StdioLineFramer.cs @@ -0,0 +1,68 @@ +// Licensed to the .NET Foundation under one or more agreements. +// The .NET Foundation licenses this file to you under the MIT license. +// See the LICENSE file in the project root for more information. + +namespace DevProxy.Stdio; + +/// +/// Splits a byte stream into newline-delimited messages (as used by JSON-RPC +/// over stdio, e.g. MCP). Partial lines are buffered until the terminating +/// newline arrives or the stream ends. +/// +internal sealed class StdioLineFramer +{ + private readonly List _pending = []; + + /// + /// Appends data and returns all complete messages. Each message includes + /// its trailing newline so that forwarded bytes are unchanged. + /// + public IReadOnlyList Append(ReadOnlySpan data) + { + var messages = new List(); + + while (!data.IsEmpty) + { + var newlineIndex = data.IndexOf((byte)'\n'); + if (newlineIndex < 0) + { + _pending.AddRange(data); + break; + } + + var line = data[..(newlineIndex + 1)]; + byte[] message; + if (_pending.Count == 0) + { + message = line.ToArray(); + } + else + { + _pending.AddRange(line); + message = [.. _pending]; + _pending.Clear(); + } + + messages.Add(message); + data = data[(newlineIndex + 1)..]; + } + + return messages; + } + + /// + /// Returns any buffered data that wasn't terminated by a newline, or + /// null if there is none. + /// + public byte[]? Flush() + { + if (_pending.Count == 0) + { + return null; + } + + byte[] remainder = [.. _pending]; + _pending.Clear(); + return remainder; + } +} From 6418cf61d9f327c4d7c405f04aa93622f44968e4 Mon Sep 17 00:00:00 2001 From: waldekmastykarz Date: Sat, 31 Oct 2026 10:14:11 +0100 Subject: [PATCH 2/3] Add end-to-end regression test for back-to-back stdio messages (#1757) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ProcessSmokeTests.cs | 97 +++++++++++++++++++ 1 file changed, 97 insertions(+) diff --git a/DevProxy.Integration.Tests/ProcessSmokeTests.cs b/DevProxy.Integration.Tests/ProcessSmokeTests.cs index 8c6e8c85f..73e6db6d0 100644 --- a/DevProxy.Integration.Tests/ProcessSmokeTests.cs +++ b/DevProxy.Integration.Tests/ProcessSmokeTests.cs @@ -403,6 +403,74 @@ public async Task Host_LoadsMockPluginFromConfig_AndServesMock() } } + [Fact] + public async Task Stdio_BackToBackJsonRpcMessages_AreMockedIndividually() + { + var hostDll = LocateHostDll(); + Assert.True(File.Exists(hostDll), $"DevProxy host not built at {hostDll}. Run `dotnet build DevProxy` first."); + + var workDir = Directory.CreateTempSubdirectory("devproxy-stdio-"); + try + { + await File.WriteAllTextAsync(Path.Combine(workDir.FullName, "stdio-mocks.json"), StdioMocksFile); + await File.WriteAllTextAsync(Path.Combine(workDir.FullName, "devproxyrc.json"), StdioConfigFile); + + var psi = new ProcessStartInfo + { + FileName = "dotnet", + WorkingDirectory = workDir.FullName, + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + }; + psi.ArgumentList.Add(hostDll); + psi.ArgumentList.Add("stdio"); + psi.ArgumentList.Add("--config-file"); + psi.ArgumentList.Add(Path.Combine(workDir.FullName, "devproxyrc.json")); + psi.ArgumentList.Add("--"); + // Any child that stays alive while reading stdin; unmocked messages are blocked + // so it never receives input. + if (OperatingSystem.IsWindows()) + { + psi.ArgumentList.Add("cmd.exe"); + psi.ArgumentList.Add("/c"); + psi.ArgumentList.Add("more"); + } + else + { + psi.ArgumentList.Add("cat"); + } + + using var process = new Process { StartInfo = psi }; + Assert.True(process.Start(), "Failed to start DevProxy stdio process."); + try + { + // Write an id-less notification and an id-bearing request in a single write so + // that they arrive in the same stdin read (#1757). + await process.StandardInput.WriteAsync( + "{\"jsonrpc\":\"2.0\",\"method\":\"notifications/initialized\"}\n" + + "{\"jsonrpc\":\"2.0\",\"id\":2,\"method\":\"tools/call\",\"params\":{\"name\":\"accept_eula\"}}\n"); + await process.StandardInput.FlushAsync(); + + using var cts = new CancellationTokenSource(s_startupTimeout); + var line = await ReadLineAsync(process.StandardOutput, cts.Token); + Assert.NotNull(line); + + using var response = JsonDocument.Parse(line); + Assert.Equal(2, response.RootElement.GetProperty("id").GetInt32()); + } + finally + { + StopHost(process); + } + } + finally + { + workDir.Delete(recursive: true); + } + } + private static Process StartHost(string hostDll, string workDir) { var psi = new ProcessStartInfo @@ -524,6 +592,35 @@ private static string LocateHostDll() return Path.Combine(AppContext.BaseDirectory, "devproxy.dll"); } + private const string StdioConfigFile = """ + { + "plugins": [ + { + "name": "MockStdioResponsePlugin", + "enabled": true, + "pluginPath": "~appFolder/plugins/DevProxy.Plugins.dll", + "configSection": "mockStdioResponsePlugin" + } + ], + "mockStdioResponsePlugin": { + "mocksFile": "stdio-mocks.json", + "blockUnmockedRequests": true + }, + "validateSchemas": false + } + """; + + private const string StdioMocksFile = """ + { + "mocks": [ + { + "request": { "bodyFragment": "accept_eula" }, + "response": { "stdout": "{\"jsonrpc\":\"2.0\",\"id\":@stdin.body.id,\"result\":{}}\n" } + } + ] + } + """; + private const string ConfigFile = """ { "$schema": "https://raw.githubusercontent.com/dotnet/dev-proxy/main/schemas/v3.1.0/rc.schema.json", From 8b950e93d3f514881638badc6799055aa8fdfd7c Mon Sep 17 00:00:00 2001 From: waldekmastykarz Date: Sat, 31 Oct 2026 10:53:23 +0100 Subject: [PATCH 3/3] Use separate write gates for child stdin and parent stdout (#1757) Writing a large framed stdin message while holding a shared semaphore could deadlock with streaming children because stdout couldn't drain. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../ProcessSmokeTests.cs | 59 +++++++++++++++++++ DevProxy/Stdio/ProxySession.cs | 23 ++++---- 2 files changed, 72 insertions(+), 10 deletions(-) diff --git a/DevProxy.Integration.Tests/ProcessSmokeTests.cs b/DevProxy.Integration.Tests/ProcessSmokeTests.cs index 73e6db6d0..75ea5ae26 100644 --- a/DevProxy.Integration.Tests/ProcessSmokeTests.cs +++ b/DevProxy.Integration.Tests/ProcessSmokeTests.cs @@ -471,6 +471,65 @@ await process.StandardInput.WriteAsync( } } + [Fact] + public async Task Stdio_LargeMessageThroughStreamingChild_DoesNotDeadlock() + { + if (OperatingSystem.IsWindows()) + { + return; + } + + var hostDll = LocateHostDll(); + Assert.True(File.Exists(hostDll), $"DevProxy host not built at {hostDll}. Run `dotnet build DevProxy` first."); + + var workDir = Directory.CreateTempSubdirectory("devproxy-stdio-large-"); + try + { + await File.WriteAllTextAsync(Path.Combine(workDir.FullName, "stdio-mocks.json"), StdioMocksFile); + await File.WriteAllTextAsync(Path.Combine(workDir.FullName, "devproxyrc.json"), + StdioConfigFile.Replace("\"blockUnmockedRequests\": true", "\"blockUnmockedRequests\": false", StringComparison.Ordinal)); + + var psi = new ProcessStartInfo + { + FileName = "dotnet", + WorkingDirectory = workDir.FullName, + RedirectStandardInput = true, + RedirectStandardOutput = true, + RedirectStandardError = true, + UseShellExecute = false, + }; + psi.ArgumentList.Add(hostDll); + psi.ArgumentList.Add("stdio"); + psi.ArgumentList.Add("--config-file"); + psi.ArgumentList.Add(Path.Combine(workDir.FullName, "devproxyrc.json")); + psi.ArgumentList.Add("--"); + psi.ArgumentList.Add("cat"); + + using var process = new Process { StartInfo = psi }; + Assert.True(process.Start(), "Failed to start DevProxy stdio process."); + try + { + // Larger than typical pipe buffers so the child blocks writing stdout + // while the proxy is still writing the message to its stdin. + var line = new string('x', 1024 * 1024); + using var cts = new CancellationTokenSource(s_startupTimeout); + var readTask = ReadLineAsync(process.StandardOutput, cts.Token); + await process.StandardInput.WriteAsync(line + "\n"); + await process.StandardInput.FlushAsync(); + + Assert.Equal(line, await readTask); + } + finally + { + StopHost(process); + } + } + finally + { + workDir.Delete(recursive: true); + } + } + private static Process StartHost(string hostDll, string workDir) { var psi = new ProcessStartInfo diff --git a/DevProxy/Stdio/ProxySession.cs b/DevProxy/Stdio/ProxySession.cs index 4934dbcaf..d07757682 100644 --- a/DevProxy/Stdio/ProxySession.cs +++ b/DevProxy/Stdio/ProxySession.cs @@ -40,7 +40,9 @@ internal sealed class ProxySession( private Stream? _childStdin; private StdioSession? _stdioSession; - private readonly SemaphoreSlim _writeSemaphore = new(1, 1); + // Separate gates so stdout can drain while a large stdin message is being written + private readonly SemaphoreSlim _childStdinSemaphore = new(1, 1); + private readonly SemaphoreSlim _parentStdoutSemaphore = new(1, 1); private bool _disposed; /// @@ -55,7 +57,7 @@ public async Task WriteToChildStdinAsync(string message) } var bytes = Encoding.UTF8.GetBytes(message); - await _writeSemaphore.WaitAsync(); + await _childStdinSemaphore.WaitAsync(); try { Log("INJECT >>>", bytes, bytes.Length); @@ -68,7 +70,7 @@ public async Task WriteToChildStdinAsync(string message) } finally { - _ = _writeSemaphore.Release(); + _ = _childStdinSemaphore.Release(); } } @@ -84,7 +86,7 @@ public async Task WriteToParentStdoutAsync(string message) } var bytes = Encoding.UTF8.GetBytes(message); - await _writeSemaphore.WaitAsync(); + await _parentStdoutSemaphore.WaitAsync(); try { Log("INJECT <<< STDOUT", bytes, bytes.Length); @@ -97,7 +99,7 @@ public async Task WriteToParentStdoutAsync(string message) } finally { - _ = _writeSemaphore.Release(); + _ = _parentStdoutSemaphore.Release(); } } @@ -288,7 +290,7 @@ private async Task ProcessStdinMessageAsync(byte[] message) // Only forward to child if not consumed if (!responseState.HasBeenSet) { - await _writeSemaphore.WaitAsync(_cts.Token); + await _childStdinSemaphore.WaitAsync(_cts.Token); try { await _childStdin!.WriteAsync(message, _cts.Token); @@ -296,7 +298,7 @@ private async Task ProcessStdinMessageAsync(byte[] message) } finally { - _ = _writeSemaphore.Release(); + _ = _childStdinSemaphore.Release(); } } else @@ -356,7 +358,7 @@ private async Task ForwardStdoutAsync() // Only forward to parent if not consumed if (!responseState.HasBeenSet) { - await _writeSemaphore.WaitAsync(_cts.Token); + await _parentStdoutSemaphore.WaitAsync(_cts.Token); try { await _parentStdout!.WriteAsync(buffer.AsMemory(0, bytesRead), _cts.Token); @@ -364,7 +366,7 @@ private async Task ForwardStdoutAsync() } finally { - _ = _writeSemaphore.Release(); + _ = _parentStdoutSemaphore.Release(); } } @@ -653,6 +655,7 @@ public void Dispose() _parentStdout?.Dispose(); _process?.Dispose(); _cts.Dispose(); - _writeSemaphore.Dispose(); + _childStdinSemaphore.Dispose(); + _parentStdoutSemaphore.Dispose(); } }