Enforce process ownership: guarantee process termination across all execution paths

CliWrap.Tests/EventStreamSpecs.cs

@@ -120,6 +120,23 @@ public async Task I_can_execute_a_command_as_a_push_based_event_stream_and_not_h
         await cmd.Observe().ToArray();
     }
 
+    [Fact(Timeout = 15000)]
+    public async Task I_can_execute_a_command_as_a_push_based_event_stream_and_the_underlying_process_is_killed_if_the_subscription_is_abandoned()
+    {
+        // Arrange
+        var cmd = Cli.Wrap(Dummy.Program.FilePath).WithArguments(["sleep", "00:00:20"]);
+
+        // Act: subscribe but take only the first event (StartedCommandEvent) then dispose
+        var startedEvent = await cmd.Observe().OfType<StartedCommandEvent>().FirstAsync();
+
+        // Assert: the process is not left running in the background.
+        // Kill() is system-level asynchronous, so we poll until it takes effect.
+        while (Process.IsRunning(startedEvent.ProcessId))
+            await Task.Delay(10);
+
+        Process.IsRunning(startedEvent.ProcessId).Should().BeFalse();
+    }
+
     [Fact(Timeout = 15000)]
     public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_the_underlying_process_is_killed_if_the_iterator_is_abandoned()
     {
@@ -136,7 +153,11 @@ public async Task I_can_execute_a_command_as_a_pull_based_event_stream_and_the_u
             break;
         }
 
-        // Assert: the process is not left running in the background
+        // Assert: the process is not left running in the background.
+        // Kill() is system-level asynchronous, so we poll until it takes effect.
+        while (Process.IsRunning(processId))
+            await Task.Delay(10);
+
         Process.IsRunning(processId).Should().BeFalse();
     }
 }

CliWrap/EventStream/PushEventStreamCommandExtensions.cs

@@ -31,6 +31,12 @@ CancellationToken gracefulCancellationToken
         ) =>
             Observable.CreateSynchronized<CommandEvent>(observer =>
             {
+                // Used to kill the process if the subscription is disposed (abandoned)
+                // before the command finishes executing.
+                var killCts = CancellationTokenSource.CreateLinkedTokenSource(
+                    forcefulCancellationToken
+                );
+
                 var stdOutPipe = PipeTarget.Merge(
                     command.StandardOutputPipe,
                     PipeTarget.ToDelegate(
@@ -47,11 +53,12 @@ CancellationToken gracefulCancellationToken
                     )
                 );
 
-                // Execute the command with the pipes extended to push events to the observer
+                // Execute the command with killCts.Token as the forceful cancellation token,
+                // so that disposing the subscription (which cancels killCts) also kills the process.
                 var commandTask = command
                     .WithStandardOutputPipe(stdOutPipe)
                     .WithStandardErrorPipe(stdErrPipe)
-                    .ExecuteAsync(forcefulCancellationToken, gracefulCancellationToken);
+                    .ExecuteAsync(killCts.Token, gracefulCancellationToken);
 
                 observer.OnNext(new StartedCommandEvent(commandTask.ProcessId));
 
@@ -86,7 +93,16 @@ CancellationToken gracefulCancellationToken
                     // doesn't get reported to the finalizer thread and crash the process.
                     .Task.ObserveException();
 
-                return Disposable.Null;
+                // Return a disposable that cancels killCts to terminate the process if the
+                // subscription is disposed before the command completes.
+                // This is also triggered on normal completion (OnCompleted/OnError dispose
+                // the subscription), but by then the process has already exited, so Kill()
+                // is a no-op.
+                return Disposable.Create(() =>
+                {
+                    killCts.Cancel();
+                    killCts.Dispose();
+                });
             });
 
         /// <summary>