Merge pull request #299 from fsprojects/repo-assist/feat-reduce-2026-03-4df851dee5ec5f9d

[Repo Assist] feat: add TaskSeq.reduce and TaskSeq.reduceAsync (ref #289)

Author: dsyme

src/FSharp.Control.TaskSeq.Test/FSharp.Control.TaskSeq.Test.fsproj

@@ -26,6 +26,7 @@
     <Compile Include="TaskSeq.FindIndex.Tests.fs" />
     <Compile Include="TaskSeq.Find.Tests.fs" />
     <Compile Include="TaskSeq.Fold.Tests.fs" />
+    <Compile Include="TaskSeq.Reduce.Tests.fs" />
     <Compile Include="TaskSeq.Forall.Tests.fs" />
     <Compile Include="TaskSeq.Head.Tests.fs" />
     <Compile Include="TaskSeq.Indexed.Tests.fs" />

src/FSharp.Control.TaskSeq.Test/TaskSeq.Reduce.Tests.fs

@@ -0,0 +1,156 @@
+module TaskSeq.Tests.Reduce
+
+open Xunit
+open FsUnit.Xunit
+
+open FSharp.Control
+
+//
+// TaskSeq.reduce
+// TaskSeq.reduceAsync
+//
+
+module EmptySeq =
+    [<Fact>]
+    let ``Null source is invalid`` () =
+        assertNullArg
+        <| fun () -> TaskSeq.reduce (fun a _ -> a) null
+
+        assertNullArg
+        <| fun () -> TaskSeq.reduceAsync (fun a _ -> Task.fromResult a) null
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-reduce raises on empty`` variant =
+        fun () ->
+            Gen.getEmptyVariant variant
+            |> TaskSeq.reduce (fun a b -> a + b)
+            |> Task.ignore
+
+        |> should throwAsyncExact typeof<System.ArgumentException>
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-reduceAsync raises on empty`` variant =
+        fun () ->
+            Gen.getEmptyVariant variant
+            |> TaskSeq.reduceAsync (fun a b -> task { return a + b })
+            |> Task.ignore
+
+        |> should throwAsyncExact typeof<System.ArgumentException>
+
+module Immutable =
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-reduce folds from first element`` variant = task {
+        // items are 1..10; sum = 55
+        let! sum =
+            Gen.getSeqImmutable variant
+            |> TaskSeq.reduce (fun acc item -> acc + item)
+
+        sum |> should equal 55
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-reduceAsync folds from first element`` variant = task {
+        let! sum =
+            Gen.getSeqImmutable variant
+            |> TaskSeq.reduceAsync (fun acc item -> task { return acc + item })
+
+        sum |> should equal 55
+    }
+
+    [<Fact>]
+    let ``TaskSeq-reduce returns single element without calling folder`` () = task {
+        let mutable called = false
+
+        let! result =
+            TaskSeq.singleton 42
+            |> TaskSeq.reduce (fun _ _ ->
+                called <- true
+                failwith "should not be called")
+
+        result |> should equal 42
+        called |> should equal false
+    }
+
+    [<Fact>]
+    let ``TaskSeq-reduceAsync returns single element without calling folder`` () = task {
+        let mutable called = false
+
+        let! result =
+            TaskSeq.singleton 42
+            |> TaskSeq.reduceAsync (fun _ _ -> task {
+                called <- true
+                return failwith "should not be called"
+            })
+
+        result |> should equal 42
+        called |> should equal false
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-reduce uses first element as initial accumulator`` variant = task {
+        // reduce must use element[0] as initial state; for 1..10 summing gives 55
+        // if it used 0 as initial, sum would also be 55 — but we verify the folder is called n-1 times
+        let mutable callCount = 0
+
+        let! sum =
+            Gen.getSeqImmutable variant
+            |> TaskSeq.reduce (fun acc item ->
+                callCount <- callCount + 1
+                acc + item)
+
+        sum |> should equal 55
+        callCount |> should equal 9 // 10 elements => 9 reduce calls
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-reduce can concatenate strings`` variant = task {
+        // items 1..10 as chars: ABCDEFGHIJ
+        let! letters =
+            Gen.getSeqImmutable variant
+            |> TaskSeq.map (fun i -> string (char (i + 64)))
+            |> TaskSeq.reduce (fun acc item -> acc + item)
+
+        letters |> should equal "ABCDEFGHIJ"
+    }
+
+    [<Theory; ClassData(typeof<TestImmTaskSeq>)>]
+    let ``TaskSeq-reduceAsync can concatenate strings`` variant = task {
+        let! letters =
+            Gen.getSeqImmutable variant
+            |> TaskSeq.map (fun i -> string (char (i + 64)))
+            |> TaskSeq.reduceAsync (fun acc item -> task { return acc + item })
+
+        letters |> should equal "ABCDEFGHIJ"
+    }
+
+module SideEffects =
+    [<Theory; ClassData(typeof<TestSideEffectTaskSeq>)>]
+    let ``TaskSeq-reduce folds correctly with side-effecting sequences`` variant = task {
+        let ts = Gen.getSeqWithSideEffect variant
+
+        let! sum = ts |> TaskSeq.reduce (fun acc item -> acc + item)
+
+        sum |> should equal 55
+
+        // second enumeration produces next 10 elements: 11..20, sum = 155
+        let! sum2 = ts |> TaskSeq.reduce (fun acc item -> acc + item)
+
+        sum2 |> should equal 155
+    }
+
+    [<Theory; ClassData(typeof<TestSideEffectTaskSeq>)>]
+    let ``TaskSeq-reduceAsync folds correctly with side-effecting sequences`` variant = task {
+        let ts = Gen.getSeqWithSideEffect variant
+
+        let! sum =
+            ts
+            |> TaskSeq.reduceAsync (fun acc item -> task { return acc + item })
+
+        sum |> should equal 55
+
+        let! sum2 =
+            ts
+            |> TaskSeq.reduceAsync (fun acc item -> task { return acc + item })
+
+        sum2 |> should equal 155
+    }

src/FSharp.Control.TaskSeq/TaskSeq.fs

@@ -407,3 +407,5 @@ type TaskSeq private () =
     static member zip source1 source2 = Internal.zip source1 source2
     static member fold folder state source = Internal.fold (FolderAction folder) state source
     static member foldAsync folder state source = Internal.fold (AsyncFolderAction folder) state source
+    static member reduce folder source = Internal.reduce (FolderAction folder) source
+    static member reduceAsync folder source = Internal.reduce (AsyncFolderAction folder) source

src/FSharp.Control.TaskSeq/TaskSeq.fsi

@@ -1361,6 +1361,38 @@ type TaskSeq =
     static member foldAsync:
         folder: ('State -> 'T -> #Task<'State>) -> state: 'State -> source: TaskSeq<'T> -> Task<'State>
 
+    /// <summary>
+    /// Applies the function <paramref name="folder" /> to each element of the task sequence, threading an accumulator
+    /// argument through the computation. The first element is used as the initial state. If the input function is
+    /// <paramref name="f" /> and the elements are <paramref name="i0...iN" />, then computes
+    /// <paramref name="f (... (f i0 i1)...) iN" />. Raises <see cref="T:System.ArgumentException" /> when the
+    /// sequence is empty.
+    /// If the accumulator function <paramref name="folder" /> is asynchronous, consider using <see cref="TaskSeq.reduceAsync" />.
+    /// </summary>
+    ///
+    /// <param name="folder">A function that updates the state with each element from the sequence.</param>
+    /// <param name="source">The input sequence.</param>
+    /// <returns>The final state value after applying the reduction function to all elements.</returns>
+    /// <exception cref="T:ArgumentNullException">Thrown when the input task sequence is null.</exception>
+    /// <exception cref="T:ArgumentException">Thrown when the input task sequence is empty.</exception>
+    static member reduce: folder: ('T -> 'T -> 'T) -> source: TaskSeq<'T> -> Task<'T>
+
+    /// <summary>
+    /// Applies the asynchronous function <paramref name="folder" /> to each element of the task sequence, threading
+    /// an accumulator argument through the computation. The first element is used as the initial state. If the input
+    /// function is <paramref name="f" /> and the elements are <paramref name="i0...iN" />, then computes
+    /// <paramref name="f (... (f i0 i1)...) iN" />. Raises <see cref="T:System.ArgumentException" /> when the
+    /// sequence is empty.
+    /// If the accumulator function <paramref name="folder" /> is synchronous, consider using <see cref="TaskSeq.reduce" />.
+    /// </summary>
+    ///
+    /// <param name="folder">A function that updates the state with each element from the sequence.</param>
+    /// <param name="source">The input sequence.</param>
+    /// <returns>The final state value after applying the reduction function to all elements.</returns>
+    /// <exception cref="T:ArgumentNullException">Thrown when the input task sequence is null.</exception>
+    /// <exception cref="T:ArgumentException">Thrown when the input task sequence is empty.</exception>
+    static member reduceAsync: folder: ('T -> 'T -> #Task<'T>) -> source: TaskSeq<'T> -> Task<'T>
+
     /// <summary>
     /// Return a new task sequence with a new item inserted before the given index.
     /// </summary>

src/FSharp.Control.TaskSeq/TaskSeqInternal.fs

@@ -371,6 +371,37 @@ module internal TaskSeqInternal =
             return result
         }
 
+    let reduce folder (source: TaskSeq<_>) =
+        checkNonNull (nameof source) source
+
+        task {
+            use e = source.GetAsyncEnumerator CancellationToken.None
+            let! hasFirst = e.MoveNextAsync()
+
+            if not hasFirst then
+                raiseEmptySeq ()
+
+            let mutable result = e.Current
+            let! step = e.MoveNextAsync()
+            let mutable go = step
+
+            match folder with
+            | FolderAction folder ->
+                while go do
+                    result <- folder result e.Current
+                    let! step = e.MoveNextAsync()
+                    go <- step
+
+            | AsyncFolderAction folder ->
+                while go do
+                    let! tempResult = folder result e.Current
+                    result <- tempResult
+                    let! step = e.MoveNextAsync()
+                    go <- step
+
+            return result
+        }
+
     let toResizeArrayAsync source =
         checkNonNull (nameof source) source