Enforce process ownership: guarantee process termination across all execution paths

CliWrap.Tests/EventStreamSpecs.cs

@@ -1,9 +1,11 @@
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.Linq;
 using System.Reactive.Linq;
 using System.Threading.Tasks;
 using CliWrap.EventStream;
 using FluentAssertions;
+using PowerKit.Extensions;
 using Xunit;
 
 namespace CliWrap.Tests;
@@ -31,6 +33,26 @@ public async Task I_can_execute_a_command_as_a_pull_based_event_stream()
         events.OfType<ExitedCommandEvent>().Single().ExitCode.Should().Be(0);
     }
 
+    [Fact(Timeout = 15000)]
+    public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_break_out_early()
+    {
+        // Arrange
+        var cmd = Cli.Wrap(Dummy.Program.FilePath).WithArguments(["sleep", "00:00:20"]);
+
+        // Act
+        var processId = 0;
+        await foreach (var cmdEvent in cmd.ListenAsync())
+        {
+            if (cmdEvent is StartedCommandEvent startedEvent)
+                processId = startedEvent.ProcessId;
+
+            break;
+        }
+
+        // Assert
+        Process.IsRunning(processId).Should().BeFalse();
+    }
+
     [Fact(Timeout = 15000)]
     public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_not_hang_on_large_stdout_and_stderr()
     {
@@ -99,6 +121,22 @@ public async Task I_can_execute_a_command_as_a_push_based_event_stream()
         events.OfType<ExitedCommandEvent>().Single().ExitCode.Should().Be(0);
     }
 
+    [Fact(Timeout = 15000)]
+    public async Task I_can_execute_a_command_as_a_push_based_event_stream_and_abandon_it_early()
+    {
+        // Arrange
+        var cmd = Cli.Wrap(Dummy.Program.FilePath).WithArguments(["sleep", "00:00:20"]);
+
+        // Act
+        var startedEvent = await cmd.Observe().OfType<StartedCommandEvent>().FirstAsync();
+
+        // Assert
+        // Abandoning the subscription triggers the kill asynchronously, so poll
+        // until the process has terminated (bounded by the test timeout)
+        while (Process.IsRunning(startedEvent.ProcessId))
+            await Task.Delay(100);
+    }
+
     [Fact(Timeout = 15000)]
     public async Task I_can_execute_a_command_as_a_push_based_event_stream_and_not_hang_on_large_stdout_and_stderr()
     {

CliWrap.Tests/PipingSpecs.cs

@@ -1,13 +1,15 @@
 using System;
 using System.Buffers;
 using System.Collections.Generic;
+using System.Diagnostics;
 using System.IO;
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
 using CliWrap.Buffered;
 using FluentAssertions;
 using PowerKit;
+using PowerKit.Extensions;
 using Xunit;
 
 namespace CliWrap.Tests;
@@ -634,10 +636,14 @@ public async Task I_can_try_to_execute_a_command_and_get_an_error_if_the_pipe_so
             | Cli.Wrap(Dummy.Program.FilePath).WithArguments("echo stdin");
 
         // Act
-        var act = async () => await cmd.ExecuteAsync();
+        var task = cmd.ExecuteAsync();
+        var act = async () => await task;
 
         // Assert
         await act.Should().ThrowAsync<Exception>();
+
+        // Assert: the process is not left running in the background
+        Process.IsRunning(task.ProcessId).Should().BeFalse();
     }
 
     [Fact(Timeout = 15000)]
@@ -650,10 +656,14 @@ public async Task I_can_try_to_execute_a_command_and_get_an_error_if_the_pipe_ta
             | PipeTarget.ToFile("non-existing-directory/file.txt");
 
         // Act
-        var act = async () => await cmd.ExecuteAsync();
+        var task = cmd.ExecuteAsync();
+        var act = async () => await task;
 
         // Assert
         await act.Should().ThrowAsync<Exception>();
+
+        // Assert: the process is not left running in the background
+        Process.IsRunning(task.ProcessId).Should().BeFalse();
     }
 
     [Fact(Timeout = 15000)]

CliWrap/Command.Execution.cs

@@ -215,46 +215,72 @@ private async Task<CommandResult> ExecuteAsync(
     {
         using var _ = process;
 
+        // CliWrap owns the process lifecycle and guarantees that the process is terminated before
+        // this method returns or throws. All forceful termination — whether requested by the user
+        // or triggered internally (for example, when a pipe fails) — is funneled through this
+        // single cancellation source, which is also linked to the user-provided forceful
+        // cancellation token. This ensures that every kill goes through the same flow: the process
+        // is killed and then awaited with a timeout.
+        using var forcefulCancellationOrPanicCts = CancellationTokenSource.CreateLinkedTokenSource(
+            forcefulCancellationToken
+        );
+
         // Ideally, we don't want ExecuteAsync() to return or throw before the process actually
         // exits, but it's theoretically possible that an attempt to kill the process may fail,
         // so we need a fallback. This cancellation token is triggered after a timeout once
-        // forceful cancellation is requested, and ensures that we don't wait forever.
+        // forceful termination is requested, and ensures that we don't wait forever.
         using var waitTimeoutCts = new CancellationTokenSource();
-        await using var _1 = forcefulCancellationToken
-            .Register(() => waitTimeoutCts.CancelAfter(TimeSpan.FromSeconds(3)))
+        await using var _1 = forcefulCancellationOrPanicCts
+            .Token.Register(() => waitTimeoutCts.CancelAfter(TimeSpan.FromSeconds(3)))
             .ToAsyncDisposable();
 
         // The process may exit without fully consuming the data from the stdin pipe, in which
         // case we need a separate cancellation signal that will abort the piping operation.
-        using var stdInCts = CancellationTokenSource.CreateLinkedTokenSource(
-            forcefulCancellationToken
-        );
-
-        // Bind user-provided cancellation tokens to the process
-        await using var _2 = forcefulCancellationToken.Register(process.Kill).ToAsyncDisposable();
+        // This source is linked to the forceful termination flow (forceful cancellation or panic)
+        // and is additionally triggered when the process exits.
+        using var forcefulCancellationOrPanicOrExitCts =
+            CancellationTokenSource.CreateLinkedTokenSource(forcefulCancellationOrPanicCts.Token);
+
+        // Kill the process when forceful termination is requested
+        await using var _2 = forcefulCancellationOrPanicCts
+            .Token.Register(process.Kill)
+            .ToAsyncDisposable();
         await using var _3 = gracefulCancellationToken
             .Register(process.Interrupt)
             .ToAsyncDisposable();
 
         // Start piping streams in the background
         var pipingTask = Task.WhenAll(
-            PipeStandardInputAsync(process, stdInCts.Token),
+            PipeStandardInputAsync(process, forcefulCancellationOrPanicOrExitCts.Token),
             // Output pipe may outlive the process, so don't cancel it on process exit
             PipeStandardOutputAsync(process, forcefulCancellationToken),
             // Error pipe may outlive the process, so don't cancel it on process exit
             PipeStandardErrorAsync(process, forcefulCancellationToken)
         );
 
+        // Start waiting for the process to exit
+        var waitTask = process.WaitUntilExitAsync(waitTimeoutCts.Token);
+
         try
         {
-            // Wait until the process exits normally or gets killed.
+            // Wait until the process exits normally or gets killed, OR until piping completes/fails.
             // The timeout is started after the execution is forcefully canceled and ensures
             // that we don't wait forever in case the attempt to kill the process failed.
-            await process.WaitUntilExitAsync(waitTimeoutCts.Token).ConfigureAwait(false);
+            await Task.WhenAny(waitTask, pipingTask).ConfigureAwait(false);
+
+            // If piping failed before the process exited, request forceful termination.
+            // This prevents the deadlock where the process is blocked trying to write to a
+            // pipe that nobody is reading anymore, and — unlike a bare Kill() — it routes
+            // through the forceful-termination flow, which awaits the exit with a timeout.
+            if (!waitTask.IsCompleted && !pipingTask.IsCompletedSuccessfully)
+                await forcefulCancellationOrPanicCts.CancelAsync();
+
+            // Wait for the process to fully exit
+            await waitTask.ConfigureAwait(false);
 
             // Send the cancellation signal to the stdin pipe since the process has exited
             // and won't need it anymore. This should prevent it from hanging in some edge cases.
-            await stdInCts.CancelAsync();
+            await forcefulCancellationOrPanicOrExitCts.CancelAsync();
 
             // Wait until piping is done and propagate exceptions
             await pipingTask.ConfigureAwait(false);
@@ -269,25 +295,40 @@ private async Task<CommandResult> ExecuteAsync(
             );
         }
         catch (OperationCanceledException)
-            // Not checking ex.CancellationToken here because it will always be stdInCts.Token
+            // Not checking ex.CancellationToken here because it will always be forcefulCancellationOrPanicOrExitCts.Token
             // at this point due to the link.
             when (forcefulCancellationToken.IsCancellationRequested)
         {
             // The operation was cancelled forcefully by the user. Suppress this exception as we'll throw
             // a more meaningful one later.
         }
         catch (OperationCanceledException)
-            // Not checking ex.CancellationToken here because it will always be stdInCts.Token
+            // Not checking ex.CancellationToken here because it will always be forcefulCancellationOrPanicOrExitCts.Token
             // at this point due to the link.
             when (gracefulCancellationToken.IsCancellationRequested)
         {
             // The operation was cancelled gracefully by the user. Suppress this exception as we'll throw
             // a more meaningful one later.
         }
-        catch (OperationCanceledException ex) when (ex.CancellationToken == stdInCts.Token)
+        catch (OperationCanceledException ex)
+            when (ex.CancellationToken == forcefulCancellationOrPanicOrExitCts.Token)
         {
             // The process exited before consuming all stdin, ignore this internal cancellation
         }
+        finally
+        {
+            // Guarantee that the process is terminated on any remaining exit path (for example,
+            // an unexpected exception). Route this through the forceful-termination source rather
+            // than calling Kill() directly, so that it goes through the same flow as user
+            // cancellation: the process is killed and then awaited with a timeout. Kill() is
+            // asynchronous at the system level, so returning right after it could otherwise leave
+            // the process still running.
+            if (!waitTask.IsCompletedSuccessfully)
+            {
+                await forcefulCancellationOrPanicCts.CancelAsync();
+                await waitTask.ObserveException().ConfigureAwait(false);
+            }
+        }
 
         // Handle forceful cancellation
         if (forcefulCancellationToken.IsCancellationRequested)

CliWrap/EventStream/PullEventStreamCommandExtensions.cs

@@ -34,45 +34,23 @@ CancellationToken gracefulCancellationToken
         {
             using var channel = new Channel<CommandEvent>();
 
-            // The consumer may abandon the iterator, leaving it in a hanging but uncanceled state.
-            // In that case, we want the process to continue running in the background, but we also
-            // need to bypass the channel to drain the pipes without waiting for transmit/receive locks.
-            using var abandonCts = CancellationTokenSource.CreateLinkedTokenSource(
-                forcefulCancellationToken
-            );
+            // Used to kill the process if the consumer abandons the iterator or cancels forcefully
+            using var forcefulCancellationOrAbandonCts =
+                CancellationTokenSource.CreateLinkedTokenSource(forcefulCancellationToken);
 
+            // The delegate's cancellation token is derived from the token passed to ExecuteAsync
+            // below (forcefulCancellationOrAbandonCts.Token), so abandoning the iterator cancels
+            // the in-flight transmit as well. Any resulting cancellation is handled by ExecuteAsync.
             var stdOutPipe = PipeTarget.Merge(
                 command.StandardOutputPipe,
                 PipeTarget.ToDelegate(
                     async (line, innerCancellationToken) =>
-                    {
-                        // If the iterator was abandoned, then just turn this pipe into a no-op
-                        // so that it drains the process's output stream without deadlocking on the channel.
-                        if (abandonCts.IsCancellationRequested)
-                            return;
-
-                        try
-                        {
-                            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
-                                innerCancellationToken,
-                                abandonCts.Token
-                            );
-
-                            await channel
-                                .TransmitAsync(
-                                    new StandardOutputCommandEvent(line),
-                                    linkedCts.Token
-                                )
-                                .ConfigureAwait(false);
-                        }
-                        catch (Exception ex)
-                            when ((ex is OperationCanceledException or ObjectDisposedException)
-                                && abandonCts.IsCancellationRequested
+                        await channel
+                            .TransmitAsync(
+                                new StandardOutputCommandEvent(line),
+                                innerCancellationToken
                             )
-                        {
-                            // The iterator was abandoned during transmit, ignore
-                        }
-                    },
+                            .ConfigureAwait(false),
                     standardOutputEncoding
                 )
             );
@@ -81,40 +59,23 @@ await channel
                 command.StandardErrorPipe,
                 PipeTarget.ToDelegate(
                     async (line, innerCancellationToken) =>
-                    {
-                        // If the iterator was abandoned, then just turn this pipe into a no-op
-                        // so that it drains the process's error stream without deadlocking on the channel.
-                        if (abandonCts.IsCancellationRequested)
-                            return;
-
-                        try
-                        {
-                            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
-                                innerCancellationToken,
-                                abandonCts.Token
-                            );
-
-                            await channel
-                                .TransmitAsync(new StandardErrorCommandEvent(line), linkedCts.Token)
-                                .ConfigureAwait(false);
-                        }
-                        catch (Exception ex)
-                            when ((ex is OperationCanceledException or ObjectDisposedException)
-                                && abandonCts.IsCancellationRequested
+                        await channel
+                            .TransmitAsync(
+                                new StandardErrorCommandEvent(line),
+                                innerCancellationToken
                             )
-                        {
-                            // The iterator was abandoned during transmit, ignore
-                        }
-                    },
+                            .ConfigureAwait(false),
                     standardErrorEncoding
                 )
             );
 
-            // Execute the command with the pipes extended to transmit events to the channel
+            // Execute the command with the pipes extended to transmit events to the channel.
+            // We pass forcefulCancellationOrAbandonCts.Token as the forceful cancellation token so that abandoning the
+            // iterator (which cancels forcefulCancellationOrAbandonCts) also kills the underlying process.
             var commandTask = command
                 .WithStandardOutputPipe(stdOutPipe)
                 .WithStandardErrorPipe(stdErrPipe)
-                .ExecuteAsync(forcefulCancellationToken, gracefulCancellationToken)
+                .ExecuteAsync(forcefulCancellationOrAbandonCts.Token, gracefulCancellationToken)
                 .Bind(async task =>
                 {
                     try
@@ -127,11 +88,13 @@ await channel
                         // so that the consumer can stop listening.
                         try
                         {
-                            await channel.CloseAsync(abandonCts.Token).ConfigureAwait(false);
+                            await channel
+                                .CloseAsync(forcefulCancellationOrAbandonCts.Token)
+                                .ConfigureAwait(false);
                         }
                         catch (Exception ex)
                             when ((ex is OperationCanceledException or ObjectDisposedException)
-                                && abandonCts.IsCancellationRequested
+                                && forcefulCancellationOrAbandonCts.IsCancellationRequested
                             )
                         {
                             // The iterator was abandoned as the channel was closing, ignore
@@ -159,15 +122,23 @@ var cmdEvent in channel
             finally
             {
                 // The code after the yield return statements may not execute if the consumer
-                // breaks out of the iterator early. Because of that, the pipes will stop
-                // draining properly and the execution may deadlock. To avoid that, we trigger
-                // a token to stop transmitting events so that the command can keep draining its
-                // output without waiting for the consumer to read from the channel.
-                await abandonCts.CancelAsync();
-
-                // The task will remain detached, so observe its exception so it
-                // doesn't get reported to the finalizer thread and crash the process.
-                _ = commandTask.Task.ObserveException();
+                // breaks out of the iterator early. Cancelling forcefulCancellationOrAbandonCts
+                // terminates the underlying process and stops the pipes.
+                await forcefulCancellationOrAbandonCts.CancelAsync();
+
+                // Wait for the command to finish executing before returning, so that the process
+                // is fully terminated by the time the method returns, per the CliWrap convention.
+                // Any exception is swallowed here (rather than surfaced) because on the abandon path
+                // the command task faults with the expected forceful-cancellation exception, and on
+                // the normal path the task was already awaited above.
+                try
+                {
+                    await commandTask.ConfigureAwait(false);
+                }
+                catch
+                {
+                    // The process has been terminated; the exception is expected here
+                }
             }
         }