Add AsyncSeq.reduce and AsyncSeq.reduceAsync (#242)

Adds two new combinators:
- reduce: ('T -> 'T -> 'T) -> AsyncSeq<'T> -> Async<'T>
- reduceAsync: ('T -> 'T -> Async<'T>) -> AsyncSeq<'T> -> Async<'T>

These mirror Seq.reduce: aggregate without an initial state, raising
InvalidOperationException on empty sequences. reduce is implemented
in terms of reduceAsync for consistency.

5 new tests added covering: non-empty sequences, single-element sequences,
empty sequence exception, async reduction, and parity with Seq.reduce.

Co-authored-by: Repo Assist <[email protected]>
Co-authored-by: Copilot <[email protected]>

Author: 3 people

src/FSharp.Control.AsyncSeq/AsyncSeq.fs

@@ -1160,6 +1160,25 @@ module AsyncSeq =
   let fold f (state:'State) (source : AsyncSeq<'T>) =
     foldAsync (fun st v -> f st v |> async.Return) state source
 
+  let reduceAsync (f: 'T -> 'T -> Async<'T>) (source: AsyncSeq<'T>) : Async<'T> = async {
+      use ie = source.GetEnumerator()
+      let! first = ie.MoveNext()
+      match first with
+      | None -> return raise (InvalidOperationException("The input sequence was empty."))
+      | Some v ->
+          let acc = ref v
+          let! next = ie.MoveNext()
+          let b = ref next
+          while b.Value.IsSome do
+              let! newAcc = f acc.Value b.Value.Value
+              acc := newAcc
+              let! more = ie.MoveNext()
+              b := more
+          return acc.Value }
+
+  let reduce (f: 'T -> 'T -> 'T) (source: AsyncSeq<'T>) : Async<'T> =
+      reduceAsync (fun a b -> f a b |> async.Return) source
+
   let length (source : AsyncSeq<'T>) =
     fold (fun st _ -> st + 1L) 0L source
 

src/FSharp.Control.AsyncSeq/AsyncSeq.fsi

@@ -214,6 +214,14 @@ module AsyncSeq =
     /// specified 'aggregation' function.
     val fold : folder:('State -> 'T -> 'State) -> state:'State -> source:AsyncSeq<'T> -> Async<'State>
 
+    /// Asynchronously reduce the elements of the input asynchronous sequence using the
+    /// specified asynchronous 'reduction' function. Raises InvalidOperationException if the sequence is empty.
+    val reduceAsync : reduction:('T -> 'T -> Async<'T>) -> source:AsyncSeq<'T> -> Async<'T>
+
+    /// Asynchronously reduce the elements of the input asynchronous sequence using the
+    /// specified 'reduction' function. Raises InvalidOperationException if the sequence is empty.
+    val reduce : reduction:('T -> 'T -> 'T) -> source:AsyncSeq<'T> -> Async<'T>
+
     /// Asynchronously sum the elements of the input asynchronous sequence using the specified function.
     val inline sum : source:AsyncSeq< ^T > -> Async< ^T>
                                 when ^T : (static member ( + ) : ^T * ^T -> ^T)

tests/FSharp.Control.AsyncSeq.Tests/AsyncSeqTests.fs

@@ -2375,6 +2375,45 @@ let ``AsyncSeq.fold with empty sequence should return seed``() =
                |> Async.RunSynchronously
   Assert.AreEqual(10, result)
 
+[<Test>]
+let ``AsyncSeq.reduce sums non-empty sequence`` () =
+  let result = asyncSeq { yield 1; yield 2; yield 3; yield 4; yield 5 }
+               |> AsyncSeq.reduce (+)
+               |> Async.RunSynchronously
+  Assert.AreEqual(15, result)
+
+[<Test>]
+let ``AsyncSeq.reduce single element returns that element`` () =
+  let result = asyncSeq { yield 42 }
+               |> AsyncSeq.reduce (+)
+               |> Async.RunSynchronously
+  Assert.AreEqual(42, result)
+
+[<Test>]
+let ``AsyncSeq.reduce empty sequence raises InvalidOperationException`` () =
+  Assert.Throws<InvalidOperationException>(fun () ->
+    AsyncSeq.empty<int>
+    |> AsyncSeq.reduce (+)
+    |> Async.RunSynchronously
+    |> ignore) |> ignore
+
+[<Test>]
+let ``AsyncSeq.reduceAsync accumulates with async function`` () =
+  async {
+    let result =
+      asyncSeq { yield 10; yield 3; yield 2 }
+      |> AsyncSeq.reduceAsync (fun a b -> async { return a - b })
+      |> Async.RunSynchronously
+    Assert.AreEqual(5, result)
+  } |> Async.RunSynchronously
+
+[<Test>]
+let ``AsyncSeq.reduce matches Seq.reduce`` () =
+  for ls in [ [1]; [1;2]; [3;1;4;1;5;9;2;6] ] do
+    let expected = Seq.reduce (+) ls
+    let actual = AsyncSeq.ofSeq ls |> AsyncSeq.reduce (+) |> Async.RunSynchronously
+    Assert.AreEqual(expected, actual)
+
 [<Test>]
 let ``AsyncSeq.ofSeq should work with large sequence``() =
   let largeSeq = seq { 1 .. 100 }