diff --git a/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableExceptionFidelityTests.cs b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableExceptionFidelityTests.cs
new file mode 100644
index 0000000..377ab41
--- /dev/null
+++ b/EnumerableAsyncProcessor.UnitTests/AsyncEnumerableExceptionFidelityTests.cs
@@ -0,0 +1,265 @@
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Extensions;
+using TUnit.Assertions;
+using TUnit.Core;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+/// <summary>
+/// Guards exception fidelity and abandonment semantics on the IAsyncEnumerable paths:
+/// - the Task from void ExecuteAsync must carry every failure via Task.Exception.InnerExceptions,
+///   matching the IEnumerable processors (issue #362),
+/// - a mid-enumeration source failure must stay the primary exception instead of being masked
+///   by an in-flight task failure (issue #363),
+/// - breaking out of an unbounded result stream must cancel in-flight work instead of silently
+///   blocking until it finishes naturally (issue #363),
+/// - the unbounded ProcessInParallel extension must drain started tasks on mid-enumeration
+///   failure instead of abandoning them fire-and-forget (issue #364).
+/// </summary>
+public class AsyncEnumerableExceptionFidelityTests
+{
+    private static async IAsyncEnumerable<int> Source(int count)
+    {
+        for (var i = 0; i < count; i++)
+        {
+            await Task.Yield();
+            yield return i;
+        }
+    }
+
+    private static async IAsyncEnumerable<int> ThrowingSource(int yieldBeforeThrow)
+    {
+        for (var i = 0; i < yieldBeforeThrow; i++)
+        {
+            await Task.Yield();
+            yield return i;
+        }
+
+        throw new InvalidOperationException("source failed");
+    }
+
+    [Test]
+    public async Task Bounded_Void_ExecuteAsync_Task_Carries_Every_Failure()
+    {
+        var executeTask = Source(10)
+            .ForEachAsync(i => (i is 2 or 5 or 8)
+                ? Task.FromException(new InvalidOperationException($"item {i} failed"))
+                : Task.CompletedTask)
+            .ProcessInParallel(maxConcurrency: 2)
+            .ExecuteAsync();
+
+        Exception? caught = null;
+        try
+        {
+            await executeTask;
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is InvalidOperationException).IsTrue();
+
+        var messages = executeTask.Exception!.InnerExceptions.Select(e => e.Message).OrderBy(m => m).ToList();
+        await Assert.That(messages.Count).IsEqualTo(3);
+        await Assert.That(messages).IsEquivalentTo(new[] { "item 2 failed", "item 5 failed", "item 8 failed" });
+    }
+
+    [Test]
+    public async Task Batch_Void_ExecuteAsync_Task_Carries_Every_Failure_In_The_Batch()
+    {
+        var executeTask = Source(2)
+            .ForEachAsync(i => Task.FromException(new InvalidOperationException($"item {i} failed")))
+            .ProcessInBatches(2)
+            .ExecuteAsync();
+
+        try
+        {
+            await executeTask;
+        }
+        catch (Exception)
+        {
+            // Expected
+        }
+
+        var messages = executeTask.Exception!.InnerExceptions.Select(e => e.Message).ToList();
+        await Assert.That(messages.Count).IsEqualTo(2);
+        await Assert.That(messages).IsEquivalentTo(new[] { "item 0 failed", "item 1 failed" });
+    }
+
+    [Test]
+    public async Task Batch_Void_All_Items_Cancelled_Completes_As_Cancelled_Not_Successful()
+    {
+        using var itemCancellation = new CancellationTokenSource();
+        itemCancellation.Cancel();
+
+        var executeTask = Source(2)
+            .ForEachAsync(_ => Task.FromCanceled(itemCancellation.Token))
+            .ProcessInBatches(2)
+            .ExecuteAsync();
+
+        Exception? caught = null;
+        try
+        {
+            await executeTask;
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is OperationCanceledException).IsTrue();
+        await Assert.That(executeTask.IsCanceled).IsTrue();
+    }
+
+    [Test]
+    public async Task Unbounded_Void_Source_Failure_Stays_Primary_When_InFlight_Tasks_Also_Fail()
+    {
+        var taskStarted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+
+        var executeTask = ThrowingSource(1)
+            .ForEachAsync(async _ =>
+            {
+                taskStarted.TrySetResult();
+                await Task.Yield();
+                throw new ApplicationException("in-flight task failed");
+            })
+            .ProcessInParallel()
+            .ExecuteAsync();
+
+        await taskStarted.Task;
+
+        Exception? caught = null;
+        try
+        {
+            await executeTask;
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        // The source failure is primary; the in-flight failure is preserved alongside it
+        await Assert.That(caught is InvalidOperationException).IsTrue();
+        await Assert.That(caught!.Message).IsEqualTo("source failed");
+
+        var messages = executeTask.Exception!.InnerExceptions.Select(e => e.Message).ToList();
+        await Assert.That(messages.Count).IsEqualTo(2);
+        await Assert.That(messages.Contains("in-flight task failed")).IsTrue();
+    }
+
+    [Test]
+    public async Task Unbounded_Result_Stream_Early_Break_Cancels_InFlight_Work()
+    {
+        var cancelledCount = 0;
+
+        var processor = Source(10)
+            .SelectAsync(async (i, cancellationToken) =>
+            {
+                if (i == 0)
+                {
+                    return i;
+                }
+
+                try
+                {
+                    await Task.Delay(Timeout.Infinite, cancellationToken);
+                }
+                catch (OperationCanceledException)
+                {
+                    Interlocked.Increment(ref cancelledCount);
+                    throw;
+                }
+
+                return i;
+            })
+            .ProcessInParallel();
+
+        var stopwatch = Stopwatch.StartNew();
+
+        await foreach (var _ in processor.ExecuteAsync())
+        {
+            break;
+        }
+
+        stopwatch.Stop();
+
+        // Before the fix this blocked forever (the drain waited for every in-flight task and
+        // nothing cancelled them). Well under the 30s disposal window proves cancellation ran.
+        await Assert.That(stopwatch.Elapsed < TimeSpan.FromSeconds(15)).IsTrue();
+        await Assert.That(cancelledCount).IsEqualTo(9);
+    }
+
+    [Test]
+    public async Task Bounded_Result_Stream_Early_Break_Cancels_InFlight_Work()
+    {
+        var cancelledCount = 0;
+
+        var processor = Source(10)
+            .SelectAsync(async (i, cancellationToken) =>
+            {
+                if (i == 0)
+                {
+                    return i;
+                }
+
+                try
+                {
+                    await Task.Delay(Timeout.Infinite, cancellationToken);
+                }
+                catch (OperationCanceledException)
+                {
+                    Interlocked.Increment(ref cancelledCount);
+                    throw;
+                }
+
+                return i;
+            })
+            .ProcessInParallel(maxConcurrency: 3);
+
+        var stopwatch = Stopwatch.StartNew();
+
+        await foreach (var _ in processor.ExecuteAsync())
+        {
+            break;
+        }
+
+        stopwatch.Stop();
+
+        await Assert.That(stopwatch.Elapsed < TimeSpan.FromSeconds(15)).IsTrue();
+        await Assert.That(cancelledCount > 0).IsTrue();
+    }
+
+    [Test]
+    public async Task Unbounded_Extension_Drains_Started_Tasks_On_MidEnumeration_Failure()
+    {
+        var completedCount = 0;
+
+        Exception? caught = null;
+        try
+        {
+            _ = await ThrowingSource(3).ProcessInParallel(async i =>
+            {
+                await Task.Delay(100);
+                Interlocked.Increment(ref completedCount);
+                return i;
+            });
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is InvalidOperationException).IsTrue();
+        await Assert.That(caught!.Message).IsEqualTo("source failed");
+
+        // Before the fix the extension rethrew immediately and abandoned the started tasks
+        // fire-and-forget; now they are drained before the exception leaves the method.
+        await Assert.That(completedCount).IsEqualTo(3);
+    }
+}
diff --git a/EnumerableAsyncProcessor.UnitTests/PreCancelledTokenTests.cs b/EnumerableAsyncProcessor.UnitTests/PreCancelledTokenTests.cs
new file mode 100644
index 0000000..0dea1d1
--- /dev/null
+++ b/EnumerableAsyncProcessor.UnitTests/PreCancelledTokenTests.cs
@@ -0,0 +1,91 @@
+using System;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Extensions;
+using TUnit.Assertions;
+using TUnit.Core;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+/// <summary>
+/// Guards the pre-cancelled token contract (issue #367):
+/// building a processor with an already-cancelled token must produce a processor whose
+/// per-item tasks are cancelled - matching TPL convention - instead of throwing
+/// ArgumentException from an internal constructor parameter the caller never passed.
+/// </summary>
+public class PreCancelledTokenTests
+{
+    [Test]
+    public async Task Void_Processor_Built_With_PreCancelled_Token_Yields_Cancelled_Tasks()
+    {
+        using var cancellationTokenSource = new CancellationTokenSource();
+        cancellationTokenSource.Cancel();
+
+        var invoked = 0;
+
+        await using var processor = Enumerable.Range(0, 5).ToList()
+            .ForEachAsync(_ =>
+            {
+                Interlocked.Increment(ref invoked);
+                return Task.CompletedTask;
+            }, cancellationTokenSource.Token)
+            .ProcessInParallel(maxConcurrency: 2);
+
+        Exception? caught = null;
+        try
+        {
+            await processor.WaitAsync();
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is OperationCanceledException).IsTrue();
+        await Assert.That(processor.GetEnumerableTasks().All(t => t.IsCanceled)).IsTrue();
+        await Assert.That(invoked).IsEqualTo(0);
+    }
+
+    [Test]
+    public async Task Result_Processor_Built_With_PreCancelled_Token_Yields_Cancelled_Tasks()
+    {
+        using var cancellationTokenSource = new CancellationTokenSource();
+        cancellationTokenSource.Cancel();
+
+        await using var processor = Enumerable.Range(0, 5).ToList()
+            .SelectAsync(i => Task.FromResult(i), cancellationTokenSource.Token)
+            .ProcessInParallel(maxConcurrency: 2);
+
+        Exception? caught = null;
+        try
+        {
+            _ = await processor.GetResultsAsync();
+        }
+        catch (Exception exception)
+        {
+            caught = exception;
+        }
+
+        await Assert.That(caught is OperationCanceledException).IsTrue();
+        await Assert.That(processor.GetEnumerableTasks().All(t => t.IsCanceled)).IsTrue();
+    }
+
+    [Test]
+    public async Task OneAtATime_And_Batch_Built_With_PreCancelled_Token_Yield_Cancelled_Tasks()
+    {
+        using var cancellationTokenSource = new CancellationTokenSource();
+        cancellationTokenSource.Cancel();
+
+        await using var oneAtATime = new[] { 1, 2, 3 }
+            .ForEachAsync(_ => Task.CompletedTask, cancellationTokenSource.Token)
+            .ProcessOneAtATime();
+
+        await using var batched = new[] { 1, 2, 3 }
+            .ForEachAsync(_ => Task.CompletedTask, cancellationTokenSource.Token)
+            .ProcessInBatches(2);
+
+        await Assert.That(oneAtATime.GetEnumerableTasks().All(t => t.IsCanceled)).IsTrue();
+        await Assert.That(batched.GetEnumerableTasks().All(t => t.IsCanceled)).IsTrue();
+    }
+}
diff --git a/EnumerableAsyncProcessor.UnitTests/SingleUseExecutionTests.cs b/EnumerableAsyncProcessor.UnitTests/SingleUseExecutionTests.cs
new file mode 100644
index 0000000..bf368bf
--- /dev/null
+++ b/EnumerableAsyncProcessor.UnitTests/SingleUseExecutionTests.cs
@@ -0,0 +1,107 @@
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+using EnumerableAsyncProcessor.Extensions;
+using TUnit.Assertions;
+using TUnit.Core;
+
+namespace EnumerableAsyncProcessor.UnitTests;
+
+/// <summary>
+/// Guards the single-use contract of the IAsyncEnumerable processors (issue #369):
+/// a second ExecuteAsync call must throw InvalidOperationException naming the contract,
+/// and ExecuteAsync after disposal must throw ObjectDisposedException naming the processor -
+/// never a bare ObjectDisposedException from the internal CancellationTokenSource.
+/// </summary>
+public class SingleUseExecutionTests
+{
+    private static async IAsyncEnumerable<int> Source(int count)
+    {
+        for (var i = 0; i < count; i++)
+        {
+            await Task.Yield();
+            yield return i;
+        }
+    }
+
+    [Test]
+    public async Task Void_Processor_Second_ExecuteAsync_Throws_InvalidOperationException()
+    {
+        var processor = Source(3).ForEachAsync(_ => Task.CompletedTask).ProcessInParallel(2);
+        await processor.ExecuteAsync();
+
+        var caught = Catch(() => processor.ExecuteAsync());
+
+        await Assert.That(caught is InvalidOperationException).IsTrue();
+        await Assert.That(caught!.Message).Contains("single-use");
+    }
+
+    [Test]
+    public async Task Result_Processor_Second_ExecuteAsync_Throws_InvalidOperationException_Eagerly()
+    {
+        var processor = Source(3).SelectAsync(i => Task.FromResult(i)).ProcessInParallel(2);
+
+        await foreach (var _ in processor.ExecuteAsync())
+        {
+        }
+
+        // The guard must fire at the ExecuteAsync call, not on first MoveNextAsync
+        var caught = Catch(() => processor.ExecuteAsync());
+
+        await Assert.That(caught is InvalidOperationException).IsTrue();
+        await Assert.That(caught!.Message).Contains("single-use");
+    }
+
+    [Test]
+    public async Task Void_Processor_ExecuteAsync_After_Dispose_Throws_ObjectDisposedException()
+    {
+        var processor = Source(3).ForEachAsync(_ => Task.CompletedTask).ProcessInParallel(2);
+        processor.Dispose();
+
+        var caught = Catch(() => processor.ExecuteAsync());
+
+        await Assert.That(caught is ObjectDisposedException).IsTrue();
+        await Assert.That(caught!.Message).Contains("AsyncEnumerableParallelProcessor");
+    }
+
+    [Test]
+    public async Task Result_Processor_ExecuteAsync_After_Dispose_Throws_ObjectDisposedException()
+    {
+        var processor = Source(3).SelectAsync(i => Task.FromResult(i)).ProcessInParallel(2);
+        await processor.DisposeAsync();
+
+        var caught = Catch(() => processor.ExecuteAsync());
+
+        await Assert.That(caught is ObjectDisposedException).IsTrue();
+        await Assert.That(caught!.Message).Contains("ResultAsyncEnumerableParallelProcessor");
+    }
+
+    [Test]
+    public async Task Batch_And_OneAtATime_Processors_Enforce_Single_Use()
+    {
+        var batchProcessor = Source(3).ForEachAsync(_ => Task.CompletedTask).ProcessInBatches(2);
+        await batchProcessor.ExecuteAsync();
+        await Assert.That(Catch(() => batchProcessor.ExecuteAsync()) is InvalidOperationException).IsTrue();
+
+        var oneAtATimeProcessor = Source(3).SelectAsync(i => Task.FromResult(i)).ProcessOneAtATime();
+        await foreach (var _ in oneAtATimeProcessor.ExecuteAsync())
+        {
+        }
+
+        await Assert.That(Catch(() => oneAtATimeProcessor.ExecuteAsync()) is InvalidOperationException).IsTrue();
+    }
+
+    private static Exception? Catch(Func<object> action)
+    {
+        try
+        {
+            _ = action();
+            return null;
+        }
+        catch (Exception exception)
+        {
+            return exception;
+        }
+    }
+}
diff --git a/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs b/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs
index 209b55d..c12da9e 100644
--- a/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs
+++ b/EnumerableAsyncProcessor.UnitTests/ValidationRegressionTests.cs
@@ -13,7 +13,9 @@ namespace EnumerableAsyncProcessor.UnitTests;
 ///   (the result variants previously skipped validation, and maxConcurrency: 0 previously surfaced
 ///   as a semaphore error mid-processing),
 /// - the old arbitrary caps (10,000 tasks / 10,000 batch size) must stay removed,
-/// - an already-cancelled token must fail cleanly at build time.
+/// - an already-cancelled token must build a processor with cancelled tasks - never throw
+///   ArgumentException for an internal parameter, and never NullReferenceException from a
+///   cancellation callback firing on a partially constructed processor (see PreCancelledTokenTests).
 /// </summary>
 public class ValidationRegressionTests
 {
@@ -99,15 +101,18 @@ public async Task Large_Batch_Sizes_Are_Not_Capped()
     }
 
     [Test]
-    public async Task Already_Cancelled_Token_Fails_Cleanly_At_Build_Time()
+    public async Task Already_Cancelled_Token_Builds_A_Cancelled_Processor()
     {
         using var cancellationTokenSource = new CancellationTokenSource();
         cancellationTokenSource.Cancel();
 
-        // Must throw a clear ArgumentException - never a NullReferenceException from a
-        // cancellation callback firing on a partially constructed processor
-        await AssertThrows<ArgumentException>(() =>
-            new[] { 1 }.ForEachAsync(_ => Task.CompletedTask, cancellationTokenSource.Token).ProcessInParallel());
+        // Building must not throw; the cancellation registration fires on the fully built
+        // instance and cancels every per-item task before the processor is returned.
+        await using var processor = new[] { 1 }
+            .ForEachAsync(_ => Task.CompletedTask, cancellationTokenSource.Token)
+            .ProcessInParallel();
+
+        await Assert.That(processor.GetEnumerableTasks().All(t => t.IsCanceled)).IsTrue();
     }
 
     private static async Task AssertThrows<TException>(Func<object> action) where TException : Exception
diff --git a/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs b/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs
index d1163cd..0de33ed 100644
--- a/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs
+++ b/EnumerableAsyncProcessor/AsyncEnumerableWorkerPool.cs
@@ -1,6 +1,5 @@
 using System.Collections.Concurrent;
 using System.Runtime.CompilerServices;
-using System.Runtime.ExceptionServices;
 using System.Threading.Channels;
 
 namespace EnumerableAsyncProcessor;
@@ -11,49 +10,86 @@ namespace EnumerableAsyncProcessor;
 /// </summary>
 internal static class AsyncEnumerableWorkerPool
 {
-    internal static async Task ProcessAsync<TInput>(
+    // Returned via a TaskCompletionSource rather than as the async method's own task so the
+    // result carries Task.WhenAll fidelity: awaiting it throws the first failure while
+    // Task.Exception.InnerExceptions preserves every queued failure.
+    internal static Task ProcessAsync<TInput>(
         IAsyncEnumerable<TInput> items,
         Func<TInput, Task> taskSelector,
         int workerCount,
         CancellationToken cancellationToken)
     {
-        using var pipelineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
-        var pipelineToken = pipelineCancellation.Token;
-        var channel = CreateChannel<TInput>(workerCount);
-        var exceptions = new ConcurrentQueue<Exception>();
-        var wasCanceled = 0;
-        var workers = StartWorkers(channel.Reader, taskSelector, workerCount, exceptions, () => Interlocked.Exchange(ref wasCanceled, 1), pipelineToken);
+        var completionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        _ = RunAsync();
+        return completionSource.Task;
 
-        try
+        async Task RunAsync()
         {
             try
             {
-                await foreach (var item in items.WithCancellation(pipelineToken).ConfigureAwait(false))
+                using var pipelineCancellation = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+                var pipelineToken = pipelineCancellation.Token;
+                var channel = CreateChannel<TInput>(workerCount);
+                var exceptions = new ConcurrentQueue<Exception>();
+                var wasCanceled = 0;
+                var workers = StartWorkers(channel.Reader, taskSelector, workerCount, exceptions, () => Interlocked.Exchange(ref wasCanceled, 1), pipelineToken);
+
+                try
+                {
+                    try
+                    {
+                        await foreach (var item in items.WithCancellation(pipelineToken).ConfigureAwait(false))
+                        {
+                            await channel.Writer.WriteAsync(item, pipelineToken).ConfigureAwait(false);
+                        }
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        Interlocked.Exchange(ref wasCanceled, 1);
+                    }
+                    catch (Exception exception)
+                    {
+                        exceptions.Enqueue(exception);
+                    }
+                    finally
+                    {
+                        channel.Writer.TryComplete();
+                    }
+
+                    try
+                    {
+                        await Task.WhenAll(workers).ConfigureAwait(false);
+                    }
+                    catch (OperationCanceledException)
+                    {
+                        Interlocked.Exchange(ref wasCanceled, 1);
+                    }
+
+                    if (!exceptions.IsEmpty)
+                    {
+                        completionSource.TrySetException(exceptions);
+                    }
+                    else if (wasCanceled != 0)
+                    {
+                        completionSource.TrySetCanceled(
+                            cancellationToken.IsCancellationRequested ? cancellationToken : new CancellationToken(canceled: true));
+                    }
+                    else
+                    {
+                        completionSource.TrySetResult();
+                    }
+                }
+                finally
                 {
-                    await channel.Writer.WriteAsync(item, pipelineToken).ConfigureAwait(false);
+                    pipelineCancellation.Cancel();
+                    channel.Writer.TryComplete();
                 }
             }
-            catch (OperationCanceledException)
-            {
-                Interlocked.Exchange(ref wasCanceled, 1);
-            }
             catch (Exception exception)
             {
-                exceptions.Enqueue(exception);
+                // Defensive: nothing above should throw, but the caller's task must always complete.
+                completionSource.TrySetException(exception);
             }
-            finally
-            {
-                channel.Writer.TryComplete();
-            }
-
-            await Task.WhenAll(workers).ConfigureAwait(false);
-
-            ThrowIfFailed(exceptions, wasCanceled, cancellationToken);
-        }
-        finally
-        {
-            pipelineCancellation.Cancel();
-            channel.Writer.TryComplete();
         }
     }
 
@@ -107,12 +143,18 @@ internal static async IAsyncEnumerable<TOutput> ProcessResultsAsync<TInput, TOut
 
             try
             {
-                await Task.WhenAll(workers).ConfigureAwait(false);
+                // Bounded: a non-cancellation-aware in-flight item must not block iterator
+                // disposal indefinitely when the consumer abandons the stream early.
+                await Task.WhenAll(workers).WaitAsync(ProcessorLifecycle.DisposalTimeout).ConfigureAwait(false);
             }
             catch (OperationCanceledException) when (pipelineToken.IsCancellationRequested)
             {
                 // Expected when enumeration is canceled or the consumer stops early.
             }
+            catch (TimeoutException)
+            {
+                // Work still running after the disposal window; abandoned results below stay observed.
+            }
 
             // Results abandoned by cancellation or an earlier failure may still fault;
             // observe them so they cannot surface as UnobservedTaskException.
@@ -240,22 +282,6 @@ private static void EnqueueExceptions(
         exceptions.Enqueue(exception);
     }
 
-    private static void ThrowIfFailed(
-        ConcurrentQueue<Exception> exceptions,
-        int wasCanceled,
-        CancellationToken cancellationToken)
-    {
-        if (exceptions.TryDequeue(out var firstException))
-        {
-            ExceptionDispatchInfo.Capture(firstException).Throw();
-        }
-
-        if (wasCanceled != 0)
-        {
-            throw new OperationCanceledException(cancellationToken);
-        }
-    }
-
     private readonly record struct ResultWorkItem<TInput, TOutput>(
         TInput Input,
         TaskCompletionSource<TOutput> CompletionSource);
diff --git a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
index 654f751..3533cd4 100644
--- a/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
+++ b/EnumerableAsyncProcessor/Builders/AsyncEnumerableActionAsyncProcessorBuilder_2.cs
@@ -37,6 +37,10 @@ public AsyncEnumerableActionAsyncProcessorBuilder(
     /// <param name="maxConcurrency">Maximum concurrent operations, or null for unbounded concurrency.</param>
     /// <param name="scheduleOnThreadPool">For unbounded processing, schedules tasks on the thread pool when true.</param>
     /// <returns>An async processor configured for parallel execution.</returns>
+    /// <remarks>
+    /// Bounded processing (<paramref name="maxConcurrency"/> set) streams results in source order
+    /// with source backpressure; unbounded processing streams results in completion order.
+    /// </remarks>
     public IAsyncEnumerableProcessor<TOutput> ProcessInParallel(int? maxConcurrency = null, bool scheduleOnThreadPool = false)
     {
         return new ResultAsyncEnumerableParallelProcessor<TInput, TOutput>(
diff --git a/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs b/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs
index 82e211b..32b61ac 100644
--- a/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs
+++ b/EnumerableAsyncProcessor/Extensions/AsyncEnumerableExtensions.cs
@@ -53,7 +53,7 @@ public static async IAsyncEnumerable<TOutput> SelectMany<T, TOutput>(
         Func<T, TOutput[]> selector,
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        await foreach (var item in items.WithCancellation(cancellationToken))
+        await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
             foreach (var result in selector(item))
             {
@@ -70,7 +70,7 @@ public static async IAsyncEnumerable<TOutput> SelectMany<T, TOutput>(
         Func<T, IEnumerable<TOutput>> selector,
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        await foreach (var item in items.WithCancellation(cancellationToken))
+        await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
             foreach (var result in selector(item))
             {
@@ -87,9 +87,9 @@ public static async IAsyncEnumerable<TOutput> SelectMany<T, TOutput>(
         Func<T, IAsyncEnumerable<TOutput>> selector,
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        await foreach (var item in items.WithCancellation(cancellationToken))
+        await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
-            await foreach (var result in selector(item).WithCancellation(cancellationToken))
+            await foreach (var result in selector(item).WithCancellation(cancellationToken).ConfigureAwait(false))
             {
                 yield return result;
             }
@@ -104,7 +104,7 @@ public static async IAsyncEnumerable<TOutput> SelectManyAsync<T, TOutput>(
         Func<T, Task<TOutput[]>> selector,
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        await foreach (var item in items.WithCancellation(cancellationToken))
+        await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
             var results = await selector(item).ConfigureAwait(false);
             foreach (var result in results)
@@ -122,7 +122,7 @@ public static async IAsyncEnumerable<TOutput> SelectManyAsync<T, TOutput>(
         Func<T, Task<IEnumerable<TOutput>>> selector,
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        await foreach (var item in items.WithCancellation(cancellationToken))
+        await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
             var results = await selector(item).ConfigureAwait(false);
             foreach (var result in results)
@@ -140,10 +140,10 @@ public static async IAsyncEnumerable<TOutput> SelectManyAsync<T, TOutput>(
         Func<T, Task<IAsyncEnumerable<TOutput>>> selector,
         [System.Runtime.CompilerServices.EnumeratorCancellation] CancellationToken cancellationToken = default)
     {
-        await foreach (var item in items.WithCancellation(cancellationToken))
+        await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
         {
             var asyncEnum = await selector(item).ConfigureAwait(false);
-            await foreach (var result in asyncEnum.WithCancellation(cancellationToken))
+            await foreach (var result in asyncEnum.WithCancellation(cancellationToken).ConfigureAwait(false))
             {
                 yield return result;
             }
@@ -195,6 +195,7 @@ public static async Task<IEnumerable<TOutput>> ProcessInParallel<T, TOutput>(
     
     /// <summary>
     /// Process items in parallel with transformation, optional concurrency and thread pool scheduling, return all results as IEnumerable when awaited.
+    /// Results are returned in source order.
     /// </summary>
     public static async Task<IEnumerable<TOutput>> ProcessInParallel<T, TOutput>(
         this IAsyncEnumerable<T> items,
@@ -224,24 +225,48 @@ public static async Task<IEnumerable<TOutput>> ProcessInParallel<T, TOutput>(
         {
             // Unbounded parallel processing
             var tasks = new List<Task<TOutput>>();
-            
-            await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
+
+            try
             {
-                var capturedItem = item;
-                
-                Task<TOutput> task;
-                if (scheduleOnThreadPool)
+                await foreach (var item in items.WithCancellation(cancellationToken).ConfigureAwait(false))
                 {
-                    task = Task.Run(() => taskSelector(capturedItem), cancellationToken);
+                    var capturedItem = item;
+
+                    Task<TOutput> task;
+                    if (scheduleOnThreadPool)
+                    {
+                        task = Task.Run(() => taskSelector(capturedItem), cancellationToken);
+                    }
+                    else
+                    {
+                        task = taskSelector(capturedItem);
+                    }
+
+                    tasks.Add(task);
                 }
-                else
+            }
+            catch
+            {
+                // Mid-enumeration failure or cancellation: drain started work within the
+                // disposal window and observe its failures so nothing runs on fire-and-forget
+                // or surfaces as UnobservedTaskException; the enumeration failure stays primary.
+                if (tasks.Count > 0)
                 {
-                    task = taskSelector(capturedItem);
+                    try
+                    {
+                        await Task.WhenAll(tasks).WaitAsync(ProcessorLifecycle.DisposalTimeout).ConfigureAwait(false);
+                    }
+                    catch
+                    {
+                        // Observed below.
+                    }
+
+                    StreamingExecution.ObserveFailures(tasks);
                 }
-                
-                tasks.Add(task);
+
+                throw;
             }
-            
+
             if (tasks.Count > 0)
             {
                 var taskResults = await Task.WhenAll(tasks).ConfigureAwait(false);
diff --git a/EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs b/EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs
index c811400..58b7ef5 100644
--- a/EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs
+++ b/EnumerableAsyncProcessor/Extensions/EnumerableExtensions.cs
@@ -74,7 +74,7 @@ public static async IAsyncEnumerable<TOutput> SelectManyAsync<T, TOutput>(
     {
         foreach (var item in items)
         {
-            await foreach (var result in selector(item).WithCancellation(cancellationToken))
+            await foreach (var result in selector(item).WithCancellation(cancellationToken).ConfigureAwait(false))
             {
                 yield return result;
             }
@@ -131,7 +131,7 @@ public static async IAsyncEnumerable<TOutput> SelectManyAsync<T, TOutput>(
         {
             cancellationToken.ThrowIfCancellationRequested();
             var asyncEnum = await selector(item).ConfigureAwait(false);
-            await foreach (var result in asyncEnum.WithCancellation(cancellationToken))
+            await foreach (var result in asyncEnum.WithCancellation(cancellationToken).ConfigureAwait(false))
             {
                 yield return result;
             }
diff --git a/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs b/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs
index 728824f..8f6f5ce 100644
--- a/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs
+++ b/EnumerableAsyncProcessor/Interfaces/IAsyncEnumerableProcessor.cs
@@ -10,6 +10,12 @@ public interface IAsyncEnumerableProcessor : IAsyncDisposable, IDisposable
     /// Processes the source. The processor is single-use; it disposes its internal
     /// resources when processing completes.
     /// </summary>
+    /// <returns>
+    /// A task that completes when processing finishes. When multiple items fail, awaiting the
+    /// task throws the first failure while <c>Task.Exception.InnerExceptions</c> carries every failure.
+    /// </returns>
+    /// <exception cref="InvalidOperationException">Thrown when called a second time.</exception>
+    /// <exception cref="ObjectDisposedException">Thrown when the processor was disposed before ever executing.</exception>
     Task ExecuteAsync();
 }
 
@@ -23,5 +29,14 @@ public interface IAsyncEnumerableProcessor<out TOutput> : IAsyncDisposable, IDis
     /// Processes the source, streaming results as they become available. The processor is
     /// single-use; it disposes its internal resources when enumeration finishes.
     /// </summary>
+    /// <remarks>
+    /// Result ordering depends on the processor configuration: one-at-a-time, batch, and
+    /// bounded parallel (<c>maxConcurrency</c> set) processors yield results in source order;
+    /// unbounded parallel processors yield results in completion order.
+    /// Abandoning the stream early (breaking out of enumeration) cancels the processor's
+    /// remaining work.
+    /// </remarks>
+    /// <exception cref="InvalidOperationException">Thrown when called a second time.</exception>
+    /// <exception cref="ObjectDisposedException">Thrown when the processor was disposed before ever executing.</exception>
     IAsyncEnumerable<TOutput> ExecuteAsync();
 }
diff --git a/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor_1.cs b/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor_1.cs
index 6d0a46a..09e6f59 100644
--- a/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor_1.cs
+++ b/EnumerableAsyncProcessor/Interfaces/IAsyncProcessor_1.cs
@@ -15,22 +15,27 @@ public interface IAsyncProcessor<TOutput> : IAsyncDisposable, IDisposable
 {
     /// <summary>
     /// Gets a collection of all the asynchronous Tasks, which could be pending or complete.
+    /// Tasks appear in source order.
     /// </summary>
     /// <returns>An enumerable of tasks that may be running, completed, or faulted.</returns>
     IEnumerable<Task<TOutput>> GetEnumerableTasks();
-    
+
     /// <summary>
     /// Gets a task that will contain all mapped results when complete.
+    /// Results appear in source order.
     /// </summary>
     /// <returns>A task that completes when all processing is done, containing an array of results.</returns>
     Task<TOutput[]> GetResultsAsync();
 
     /// <summary>
     /// Gets results as an async enumerable that yields items as they complete.
+    /// Results are yielded in completion order, not source order.
     /// </summary>
     /// <returns>An async enumerable that yields results as they become available.</returns>
     /// <remarks>
     /// This method allows you to process results as they complete rather than waiting for all tasks to finish.
+    /// Because items are yielded as they finish, the order differs from <see cref="GetResultsAsync"/>,
+    /// which preserves source order.
     /// The processor should still be disposed after consuming the async enumerable.
     /// </remarks>
     IAsyncEnumerable<TOutput> GetResultsAsyncEnumerable();
diff --git a/EnumerableAsyncProcessor/ProcessorLifecycle.cs b/EnumerableAsyncProcessor/ProcessorLifecycle.cs
index df9cfcf..2c18135 100644
--- a/EnumerableAsyncProcessor/ProcessorLifecycle.cs
+++ b/EnumerableAsyncProcessor/ProcessorLifecycle.cs
@@ -22,7 +22,7 @@ internal sealed class ProcessorLifecycle
 
     public ProcessorLifecycle(CancellationTokenSource cancellationTokenSource, Action trySetCanceledAll, Action<Exception> trySetExceptionAll)
     {
-        ValidationHelper.ValidateCancellationTokenSource(cancellationTokenSource);
+        ValidationHelper.ThrowIfNull(cancellationTokenSource);
 
         _cancellationTokenSource = cancellationTokenSource;
         _trySetCanceledAll = trySetCanceledAll;
@@ -32,6 +32,8 @@ public ProcessorLifecycle(CancellationTokenSource cancellationTokenSource, Actio
 
     // Cancellation is registered here rather than at construction so that a token cancelled
     // while the processor is still being constructed can never fire on a partially built instance.
+    // An already-cancelled token invokes CancelAll synchronously on this line, so every per-item
+    // completion source is canceled before the caller ever observes the processor.
     public void Start(Func<Task> process)
     {
         _cancellationTokenRegistration = Token.Register(CancelAll);
@@ -66,9 +68,17 @@ public void CancelAll()
 
     private void CancelAllCore()
     {
-        if (!_cancellationTokenSource.IsCancellationRequested)
+        try
+        {
+            if (!_cancellationTokenSource.IsCancellationRequested)
+            {
+                _cancellationTokenSource.Cancel();
+            }
+        }
+        catch (ObjectDisposedException)
         {
-            _cancellationTokenSource.Cancel();
+            // A concurrent Dispose won the race and disposed the source; the per-item
+            // completion sources below are still canceled.
         }
 
         _trySetCanceledAll();
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs
index 82b79f5..071d642 100644
--- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableBatchProcessor.cs
@@ -9,6 +9,7 @@ public sealed class AsyncEnumerableBatchProcessor<TInput> : IAsyncEnumerableProc
     private readonly int _batchSize;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private int _disposed;
+    private int _executionState;
     private Task? _executionTask;
 
     internal AsyncEnumerableBatchProcessor(
@@ -25,17 +26,26 @@ internal AsyncEnumerableBatchProcessor(
 
     public Task ExecuteAsync()
     {
-        var executionTask = ExecuteCoreAsync();
-        _executionTask = executionTask;
-        return executionTask;
+        StreamingExecution.GuardSingleUse(ref _executionState, ref _disposed, this);
+
+        // A TaskCompletionSource rather than the async method's own task, so the returned task
+        // carries every failure from a failing batch (Task.WhenAll fidelity) instead of only
+        // the first one awaited.
+        var completionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        _executionTask = completionSource.Task;
+        _ = ExecuteCoreAsync(completionSource);
+        return completionSource.Task;
     }
 
-    private async Task ExecuteCoreAsync()
+    private async Task ExecuteCoreAsync(TaskCompletionSource completionSource)
     {
-        var cancellationToken = _cancellationTokenSource.Token;
+        var exceptions = new List<Exception>();
+        var wasCanceled = false;
+        var cancellationToken = CancellationToken.None;
 
         try
         {
+            cancellationToken = _cancellationTokenSource.Token;
             var batch = new List<TInput>(_batchSize);
 
             await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
@@ -44,27 +54,63 @@ private async Task ExecuteCoreAsync()
 
                 if (batch.Count >= _batchSize)
                 {
-                    await ProcessBatch(batch).ConfigureAwait(false);
+                    if (!await ProcessBatch(batch, exceptions).ConfigureAwait(false))
+                    {
+                        break;
+                    }
+
                     batch = new List<TInput>(_batchSize);
                 }
             }
 
             // Process any remaining items in the final batch
-            if (batch.Count > 0)
+            if (exceptions.Count == 0 && batch.Count > 0)
             {
-                await ProcessBatch(batch).ConfigureAwait(false);
+                await ProcessBatch(batch, exceptions).ConfigureAwait(false);
             }
         }
+        catch (OperationCanceledException)
+        {
+            wasCanceled = true;
+        }
+        catch (Exception exception)
+        {
+            exceptions.Add(exception);
+        }
         finally
         {
             DisposeCancellationSource(cancelFirst: false);
+            StreamingExecution.Complete(completionSource, exceptions, wasCanceled, cancellationToken);
         }
     }
 
-    private async Task ProcessBatch(List<TInput> batch)
+    // Returns false when the batch failed, which stops subsequent batches (a failing batch has
+    // always halted the run); every failure in the batch is collected, not just the first.
+    // Pure cancellation (no fault) is excluded by the filter and propagates to the caller's
+    // OperationCanceledException handler, which completes the execution task as canceled.
+    private async Task<bool> ProcessBatch(List<TInput> batch, List<Exception> exceptions)
     {
         var tasks = batch.Select(item => _taskSelector(item)).ToArray();
-        await Task.WhenAll(tasks).ConfigureAwait(false);
+        var whenAll = Task.WhenAll(tasks);
+
+        try
+        {
+            await whenAll.ConfigureAwait(false);
+            return true;
+        }
+        catch (Exception exception) when (exception is not OperationCanceledException || whenAll.IsFaulted)
+        {
+            if (whenAll.IsFaulted)
+            {
+                exceptions.AddRange(whenAll.Exception!.InnerExceptions);
+            }
+            else
+            {
+                exceptions.Add(exception);
+            }
+
+            return false;
+        }
     }
 
     public void Dispose()
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs
index 28c0936..02d8a6c 100644
--- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableOneAtATimeProcessor.cs
@@ -11,6 +11,7 @@ public sealed class AsyncEnumerableOneAtATimeProcessor<TInput> : IAsyncEnumerabl
     private readonly Func<TInput, Task> _taskSelector;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private int _disposed;
+    private int _executionState;
     private Task? _executionTask;
 
     internal AsyncEnumerableOneAtATimeProcessor(
@@ -25,25 +26,53 @@ internal AsyncEnumerableOneAtATimeProcessor(
 
     public Task ExecuteAsync()
     {
-        var executionTask = ExecuteCoreAsync();
-        _executionTask = executionTask;
-        return executionTask;
+        StreamingExecution.GuardSingleUse(ref _executionState, ref _disposed, this);
+
+        // A TaskCompletionSource rather than the async method's own task, so a multi-fault
+        // item task surfaces every inner exception instead of only the first one awaited.
+        var completionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        _executionTask = completionSource.Task;
+        _ = ExecuteCoreAsync(completionSource);
+        return completionSource.Task;
     }
 
-    private async Task ExecuteCoreAsync()
+    private async Task ExecuteCoreAsync(TaskCompletionSource completionSource)
     {
-        var cancellationToken = _cancellationTokenSource.Token;
+        var exceptions = new List<Exception>();
+        var wasCanceled = false;
+        var cancellationToken = CancellationToken.None;
 
         try
         {
+            cancellationToken = _cancellationTokenSource.Token;
+
             await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
             {
-                await _taskSelector(item).ConfigureAwait(false);
+                var task = _taskSelector(item);
+
+                try
+                {
+                    await task.ConfigureAwait(false);
+                }
+                catch (Exception exception) when (exception is not OperationCanceledException || task.IsFaulted)
+                {
+                    StreamingExecution.CollectFailures(task, exception, exceptions, ref wasCanceled);
+                    break;
+                }
             }
         }
+        catch (OperationCanceledException)
+        {
+            wasCanceled = true;
+        }
+        catch (Exception exception)
+        {
+            exceptions.Add(exception);
+        }
         finally
         {
             DisposeCancellationSource(cancelFirst: false);
+            StreamingExecution.Complete(completionSource, exceptions, wasCanceled, cancellationToken);
         }
     }
 
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs
index e824987..da02593 100644
--- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/AsyncEnumerableParallelProcessor.cs
@@ -10,6 +10,7 @@ public sealed class AsyncEnumerableParallelProcessor<TInput> : IAsyncEnumerableP
     private readonly bool _scheduleOnThreadPool;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private int _disposed;
+    private int _executionState;
     private Task? _executionTask;
 
     internal AsyncEnumerableParallelProcessor(
@@ -28,65 +29,97 @@ internal AsyncEnumerableParallelProcessor(
 
     public Task ExecuteAsync()
     {
-        var executionTask = ExecuteCoreAsync();
-        _executionTask = executionTask;
-        return executionTask;
+        StreamingExecution.GuardSingleUse(ref _executionState, ref _disposed, this);
+
+        // A TaskCompletionSource rather than the async method's own task, so the returned task
+        // carries every failure (Task.WhenAll fidelity) instead of only the first one awaited.
+        var completionSource = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
+        _executionTask = completionSource.Task;
+        _ = ExecuteCoreAsync(completionSource);
+        return completionSource.Task;
     }
 
-    private async Task ExecuteCoreAsync()
+    private async Task ExecuteCoreAsync(TaskCompletionSource completionSource)
     {
-        var cancellationToken = _cancellationTokenSource.Token;
+        var exceptions = new List<Exception>();
+        var wasCanceled = false;
+        var cancellationToken = CancellationToken.None;
 
         try
         {
+            cancellationToken = _cancellationTokenSource.Token;
+
             if (_maxConcurrency.HasValue)
             {
                 Func<TInput, Task> taskSelector = _scheduleOnThreadPool
                     ? item => Task.Run(() => _taskSelector(item), cancellationToken)
                     : _taskSelector;
 
-                await AsyncEnumerableWorkerPool.ProcessAsync(
+                var poolTask = AsyncEnumerableWorkerPool.ProcessAsync(
                     _items,
                     taskSelector,
                     _maxConcurrency.Value,
-                    cancellationToken).ConfigureAwait(false);
+                    cancellationToken);
 
-                return;
+                try
+                {
+                    await poolTask.ConfigureAwait(false);
+                }
+                catch (Exception exception)
+                {
+                    StreamingExecution.CollectFailures(poolTask, exception, exceptions, ref wasCanceled);
+                }
             }
+            else
+            {
+                // Unbounded parallel processing
+                var tasks = new List<Task>();
 
-            // Unbounded parallel processing
-            var tasks = new List<Task>();
+                try
+                {
+                    await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
+                    {
+                        var capturedItem = item;
 
-            try
-            {
-                await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
+                        tasks.Add(_scheduleOnThreadPool
+                            ? Task.Run(() => _taskSelector(capturedItem), cancellationToken)
+                            : _taskSelector(capturedItem));
+                    }
+                }
+                catch (OperationCanceledException)
+                {
+                    wasCanceled = true;
+                }
+                catch (Exception exception)
                 {
-                    var capturedItem = item;
+                    // Recorded before the drain below so a mid-enumeration source failure stays
+                    // the primary exception; started-task failures append rather than replace it.
+                    exceptions.Add(exception);
+                }
 
-                    Task task;
-                    if (_scheduleOnThreadPool)
+                if (tasks.Count > 0)
+                {
+                    var whenAll = Task.WhenAll(tasks);
+
+                    try
                     {
-                        task = Task.Run(() => _taskSelector(capturedItem), cancellationToken);
+                        await whenAll.ConfigureAwait(false);
                     }
-                    else
+                    catch (Exception exception)
                     {
-                        task = _taskSelector(capturedItem);
+                        StreamingExecution.CollectFailures(whenAll, exception, exceptions, ref wasCanceled);
                     }
-
-                    tasks.Add(task);
-                }
-            }
-            finally
-            {
-                if (tasks.Count > 0)
-                {
-                    await Task.WhenAll(tasks).ConfigureAwait(false);
                 }
             }
         }
+        catch (Exception exception)
+        {
+            exceptions.Add(exception);
+        }
         finally
         {
             DisposeCancellationSource(cancelFirst: false);
+            StreamingExecution.Complete(completionSource, exceptions, wasCanceled, cancellationToken);
         }
     }
 
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs
index fadeb29..ee4840e 100644
--- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableBatchProcessor.cs
@@ -10,6 +10,7 @@ public sealed class ResultAsyncEnumerableBatchProcessor<TInput, TOutput> : IAsyn
     private readonly int _batchSize;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private int _disposed;
+    private int _executionState;
     private TaskCompletionSource? _executionCompleted;
 
     internal ResultAsyncEnumerableBatchProcessor(
@@ -24,7 +25,13 @@ internal ResultAsyncEnumerableBatchProcessor(
         _cancellationTokenSource = cancellationTokenSource;
     }
 
-    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    public IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        StreamingExecution.GuardSingleUse(ref _executionState, ref _disposed, this);
+        return ExecuteCoreAsync();
+    }
+
+    private async IAsyncEnumerable<TOutput> ExecuteCoreAsync()
     {
         var executionCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         _executionCompleted = executionCompleted;
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs
index a519693..ba32fdf 100644
--- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableOneAtATimeProcessor.cs
@@ -11,6 +11,7 @@ public sealed class ResultAsyncEnumerableOneAtATimeProcessor<TInput, TOutput> :
     private readonly Func<TInput, Task<TOutput>> _taskSelector;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private int _disposed;
+    private int _executionState;
     private TaskCompletionSource? _executionCompleted;
 
     internal ResultAsyncEnumerableOneAtATimeProcessor(
@@ -23,7 +24,13 @@ internal ResultAsyncEnumerableOneAtATimeProcessor(
         _cancellationTokenSource = cancellationTokenSource;
     }
 
-    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    public IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        StreamingExecution.GuardSingleUse(ref _executionState, ref _disposed, this);
+        return ExecuteCoreAsync();
+    }
+
+    private async IAsyncEnumerable<TOutput> ExecuteCoreAsync()
     {
         var executionCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         _executionCompleted = executionCompleted;
diff --git a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs
index aa5af63..5d17466 100644
--- a/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs
+++ b/EnumerableAsyncProcessor/RunnableProcessors/AsyncEnumerable/ResultProcessors/ResultAsyncEnumerableParallelProcessor.cs
@@ -3,6 +3,12 @@
 
 namespace EnumerableAsyncProcessor.RunnableProcessors.AsyncEnumerable.ResultProcessors;
 
+/// <summary>
+/// Streams one result per item from an <see cref="IAsyncEnumerable{T}"/> source, processing in
+/// parallel. Bounded (<c>maxConcurrency</c> set) runs yield results in source order with source
+/// backpressure; unbounded runs yield results in completion order. Abandoning the stream early
+/// cancels remaining work.
+/// </summary>
 public sealed class ResultAsyncEnumerableParallelProcessor<TInput, TOutput> : IAsyncEnumerableProcessor<TOutput>
 {
     private readonly IAsyncEnumerable<TInput> _items;
@@ -11,6 +17,7 @@ public sealed class ResultAsyncEnumerableParallelProcessor<TInput, TOutput> : IA
     private readonly bool _scheduleOnThreadPool;
     private readonly CancellationTokenSource _cancellationTokenSource;
     private int _disposed;
+    private int _executionState;
     private TaskCompletionSource? _executionCompleted;
 
     internal ResultAsyncEnumerableParallelProcessor(
@@ -27,11 +34,18 @@ internal ResultAsyncEnumerableParallelProcessor(
         _cancellationTokenSource = cancellationTokenSource;
     }
 
-    public async IAsyncEnumerable<TOutput> ExecuteAsync()
+    public IAsyncEnumerable<TOutput> ExecuteAsync()
+    {
+        StreamingExecution.GuardSingleUse(ref _executionState, ref _disposed, this);
+        return ExecuteCoreAsync();
+    }
+
+    private async IAsyncEnumerable<TOutput> ExecuteCoreAsync()
     {
         var executionCompleted = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously);
         _executionCompleted = executionCompleted;
         var cancellationToken = _cancellationTokenSource.Token;
+        var completedNormally = false;
 
         try
         {
@@ -41,57 +55,79 @@ public async IAsyncEnumerable<TOutput> ExecuteAsync()
                     ? item => Task.Run(() => _taskSelector(item), cancellationToken)
                     : _taskSelector;
 
-                await foreach (var result in AsyncEnumerableWorkerPool.ProcessResultsAsync(
-                                   _items,
-                                   taskSelector,
-                                   _maxConcurrency.Value,
-                                   cancellationToken).ConfigureAwait(false))
+                // Enumerated manually so that on abandonment or failure the processor's token is
+                // cancelled BEFORE the pool enumerator's disposal drains its in-flight work -
+                // token-aware selectors then abort instead of being waited on to natural completion.
+                var enumerator = AsyncEnumerableWorkerPool.ProcessResultsAsync(
+                        _items,
+                        taskSelector,
+                        _maxConcurrency.Value,
+                        cancellationToken)
+                    .GetAsyncEnumerator(CancellationToken.None);
+
+                try
                 {
-                    yield return result;
+                    while (await enumerator.MoveNextAsync().ConfigureAwait(false))
+                    {
+                        yield return enumerator.Current;
+                    }
+
+                    completedNormally = true;
                 }
+                finally
+                {
+                    if (!completedNormally)
+                    {
+                        CancelForDisposal();
+                    }
 
-                yield break;
+                    await enumerator.DisposeAsync().ConfigureAwait(false);
+                }
             }
-
-            var tasks = new List<Task<TOutput>>();
-
-            // Unbounded parallel processing
-            try
+            else
             {
-                await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
-                {
-                    var capturedItem = item;
+                // Unbounded parallel processing
+                var tasks = new List<Task<TOutput>>();
 
-                    Task<TOutput> task;
-                    if (_scheduleOnThreadPool)
+                try
+                {
+                    await foreach (var item in _items.WithCancellation(cancellationToken).ConfigureAwait(false))
                     {
-                        task = Task.Run(() => _taskSelector(capturedItem), cancellationToken);
+                        var capturedItem = item;
+
+                        tasks.Add(_scheduleOnThreadPool
+                            ? Task.Run(() => _taskSelector(capturedItem), cancellationToken)
+                            : _taskSelector(capturedItem));
                     }
-                    else
+
+                    // Yield all results as they complete
+                    await foreach (var result in tasks.ToIAsyncEnumerable(cancellationToken).ConfigureAwait(false))
                     {
-                        task = _taskSelector(capturedItem);
+                        yield return result;
                     }
 
-                    tasks.Add(task);
-                }
-
-                // Yield all results as they complete
-                await foreach (var result in tasks.ToIAsyncEnumerable(cancellationToken).ConfigureAwait(false))
-                {
-                    yield return result;
+                    completedNormally = true;
                 }
-            }
-            finally
-            {
-                if (tasks.Count > 0)
+                finally
                 {
-                    try
-                    {
-                        await Task.WhenAll(tasks).ConfigureAwait(false);
-                    }
-                    catch
+                    if (!completedNormally && tasks.Count > 0)
                     {
-                        // Preserve the exception already propagating from enumeration or result consumption.
+                        // Abandonment (early break) or a propagating failure: cancel in-flight
+                        // work, drain it within the disposal window rather than for as long as it
+                        // takes, and observe its failures so they can neither mask the propagating
+                        // exception nor surface as UnobservedTaskException.
+                        CancelForDisposal();
+
+                        try
+                        {
+                            await Task.WhenAll(tasks).WaitAsync(ProcessorLifecycle.DisposalTimeout).ConfigureAwait(false);
+                        }
+                        catch
+                        {
+                            // Failures are observed below; the propagating exception stays primary.
+                        }
+
+                        StreamingExecution.ObserveFailures(tasks);
                     }
                 }
             }
diff --git a/EnumerableAsyncProcessor/StreamingExecution.cs b/EnumerableAsyncProcessor/StreamingExecution.cs
new file mode 100644
index 0000000..002baa2
--- /dev/null
+++ b/EnumerableAsyncProcessor/StreamingExecution.cs
@@ -0,0 +1,94 @@
+namespace EnumerableAsyncProcessor;
+
+/// <summary>
+/// Shared plumbing for the single-use IAsyncEnumerable processors: the single-use execution
+/// guard, and exception collection that preserves every failure from a task the way the
+/// enumerable processors' completion sources do.
+/// </summary>
+internal static class StreamingExecution
+{
+    /// <summary>
+    /// Marks the processor as executed. A second call throws <see cref="InvalidOperationException"/>;
+    /// a first call on a disposed processor throws <see cref="ObjectDisposedException"/>.
+    /// </summary>
+    public static void GuardSingleUse(ref int executionState, ref int disposed, object processor)
+    {
+        if (Interlocked.Exchange(ref executionState, 1) != 0)
+        {
+            throw new InvalidOperationException(
+                $"{processor.GetType().Name} is single-use; ExecuteAsync may only be called once.");
+        }
+
+        // Read after the exchange (a full fence) so a disposal that completed before this
+        // call claimed execution is always seen, not a stale pre-claim snapshot.
+        if (Volatile.Read(ref disposed) != 0)
+        {
+            throw new ObjectDisposedException(processor.GetType().Name);
+        }
+    }
+
+    /// <summary>
+    /// Records the failure(s) behind an awaited task, classifying by task state so a
+    /// multi-fault task contributes every inner exception rather than only the first one
+    /// the await rethrew.
+    /// </summary>
+    public static void CollectFailures(Task task, Exception thrown, List<Exception> exceptions, ref bool wasCanceled)
+    {
+        if (task.IsFaulted)
+        {
+            exceptions.AddRange(task.Exception!.InnerExceptions);
+        }
+        else if (thrown is OperationCanceledException)
+        {
+            wasCanceled = true;
+        }
+        else
+        {
+            exceptions.Add(thrown);
+        }
+    }
+
+    /// <summary>
+    /// Completes the execution task with Task.WhenAll fidelity: awaiting it throws the first
+    /// exception while <c>Task.Exception.InnerExceptions</c> carries the full set.
+    /// </summary>
+    public static void Complete(TaskCompletionSource completionSource, List<Exception> exceptions, bool wasCanceled, CancellationToken cancellationToken)
+    {
+        if (exceptions.Count > 0)
+        {
+            completionSource.TrySetException(exceptions);
+        }
+        else if (wasCanceled)
+        {
+            completionSource.TrySetCanceled(
+                cancellationToken.IsCancellationRequested ? cancellationToken : new CancellationToken(canceled: true));
+        }
+        else
+        {
+            completionSource.TrySetResult();
+        }
+    }
+
+    /// <summary>
+    /// Observes the exception of every task, attaching a continuation to those still running,
+    /// so abandoned work can never surface as <c>TaskScheduler.UnobservedTaskException</c>.
+    /// </summary>
+    public static void ObserveFailures(IEnumerable<Task> tasks)
+    {
+        foreach (var task in tasks)
+        {
+            if (task.IsCompleted)
+            {
+                _ = task.Exception;
+            }
+            else
+            {
+                _ = task.ContinueWith(
+                    static t => _ = t.Exception,
+                    CancellationToken.None,
+                    TaskContinuationOptions.ExecuteSynchronously,
+                    TaskScheduler.Default);
+            }
+        }
+    }
+}
diff --git a/EnumerableAsyncProcessor/Validation/ValidationHelper.cs b/EnumerableAsyncProcessor/Validation/ValidationHelper.cs
index 35495da..3575812 100644
--- a/EnumerableAsyncProcessor/Validation/ValidationHelper.cs
+++ b/EnumerableAsyncProcessor/Validation/ValidationHelper.cs
@@ -56,20 +56,4 @@ public static void ThrowIfNegative(TimeSpan value, [CallerArgumentExpression(nam
         }
     }
 
-    /// <summary>
-    /// Validates that a CancellationTokenSource is not null and not already cancelled.
-    /// </summary>
-    /// <param name="cancellationTokenSource">The CancellationTokenSource to validate.</param>
-    /// <param name="paramName">The parameter name for the exception.</param>
-    /// <exception cref="ArgumentNullException">Thrown when the CancellationTokenSource is null.</exception>
-    /// <exception cref="ArgumentException">Thrown when the CancellationTokenSource has already been cancelled.</exception>
-    public static void ValidateCancellationTokenSource([NotNull] CancellationTokenSource? cancellationTokenSource, [CallerArgumentExpression(nameof(cancellationTokenSource))] string? paramName = null)
-    {
-        ThrowIfNull(cancellationTokenSource, paramName);
-
-        if (cancellationTokenSource.IsCancellationRequested)
-        {
-            throw new ArgumentException($"'{paramName}' has already been cancelled.", paramName);
-        }
-    }
 }
diff --git a/README.md b/README.md
index 857f9d0..586daf4 100644
--- a/README.md
+++ b/README.md
@@ -36,6 +36,10 @@ Version 4 is a major release. Review these source and behaviour changes before u
 - **Validation is consistent and eager.** Invalid `maxConcurrency`, parallelism, batch, and timed-rate arguments now throw while the processor is built for both action and result variants.
 - **`TaskWrapper` types and processor plumbing are internal.** The `ActionTaskWrapper`/`ItemTaskWrapper` structs and the previously `protected` members of the abstract processor base classes (`TaskWrappers`, `EnumerableTaskCompletionSources`, `CancellationToken`, constructors) were implementation details that could not be meaningfully used outside the library, and are no longer public.
 - **Synchronous `InParallelAsync` delegates now run concurrently.** The `Func<TSource, TResult>` and `Action<TSource>` overloads use thread-pool workers instead of executing delegates sequentially on the caller. Do not depend on their previous thread affinity or execution order.
+- **An already-cancelled token yields cancelled tasks, not `ArgumentException`.** Building a processor with a token that is already cancelled (or that cancels between building and the terminal call) now produces a processor whose per-item tasks are cancelled; `WaitAsync`/`GetResultsAsync` throw `OperationCanceledException`. Previously this threw `ArgumentException` from an internal constructor.
+- **`IAsyncEnumerable<T>` processors preserve every failure.** The `Task` returned by `ExecuteAsync()` now carries all failures via `Task.Exception.InnerExceptions` (awaiting still throws the first), matching the `IEnumerable<T>` processors. Previously only the first failure was observable and the rest were lost.
+- **`ExecuteAsync()` enforces single use.** Calling it a second time throws `InvalidOperationException`; calling it after disposal throws `ObjectDisposedException`. Previously a second call failed with a confusing `ObjectDisposedException` from internal plumbing.
+- **Abandoning a result stream cancels remaining work.** Breaking out of `await foreach` over `ExecuteAsync()` now cancels the processor's in-flight work and bounds the drain to the 30-second disposal window, instead of silently blocking until every started task finished naturally.
 
 Version 4 also adds cancellation-aware selectors. Use `(item, cancellationToken) => ...` when in-flight work must observe external cancellation, `CancelAll()`, or disposal:
 
@@ -56,6 +60,13 @@ Bounded parallel processors use a fixed set of workers, so coordination work sca
 
 Timed processors acquire a shared token-bucket permit before starting each operation. The permit rate and maximum in-flight concurrency are separate controls; long-running operations therefore do not reduce permit replenishment. A zero-length window disables start-rate throttling but retains the concurrency limit.
 
+### Result ordering
+
+- `GetResultsAsync()` and `GetEnumerableTasks()` preserve source order.
+- `GetResultsAsyncEnumerable()` yields results in completion order.
+- `IAsyncEnumerable<T>` streaming (`ExecuteAsync()`): one-at-a-time, batch, and bounded parallel (`maxConcurrency` set) processors yield in source order; unbounded parallel yields in completion order.
+- The awaitable `IAsyncEnumerable<T>.ProcessInParallel(selector, ...)` extension returns results in source order.
+
 ## Why I built this
 
 Because I've come across situations where you need to fine tune the rate at which you do things.