feat: add TaskSeq.foldUntil and foldUntilAsync with FoldStep<'State> DU

Fold over a task sequence with early termination. The folder returns
`Continue newState` to keep consuming, or `Halt newState` to stop
immediately; in either case the state is updated. No elements past the
one that caused Halt are enumerated from the input.

Motivation: reaching for `fold`/`foldAsync` often ends with a mutable
flag or a `match state with Decided -> state | _ -> ...` guard at the
top of the folder when the caller wants to short-circuit on some
condition discovered mid-stream. `tryPickAsync` handles the "find and
halt" subcase but cannot also thread an accumulator. `foldUntil` fills
the gap and removes the impedance mismatch.

Includes:
  - FoldStep<'State> DU (Continue / Halt) in public API
  - Internal FoldUntilAction DU unifying sync/async folder dispatch
  - Public TaskSeq.foldUntil and TaskSeq.foldUntilAsync
  - XML docs on both overloads mirroring fold/foldAsync style
  - Release-notes entry
  - 22 tests covering: null source, empty sequence (no folder call,
    initial state preserved), all-Continue (equals fold), Halt-on-first
    (1 folder call, no further pulls), Halt mid-sequence (correct count,
    source not pulled past Halt), Halt-on-last (equals fold)

Author: Onur Gumus

release-notes.txt

@@ -2,6 +2,7 @@
 Release notes:
 
 Unreleased
+    - adds TaskSeq.foldUntil and TaskSeq.foldUntilAsync: fold with early termination via a FoldStep<'State> DU (Continue / Halt). The underlying sequence is not enumerated past the element that caused Halt.
     - test: add SideEffects module and ImmTaskSeq variant tests to TaskSeq.ChunkBy.Tests.fs, improving coverage for chunkBy and chunkByAsync
     - fixes: `Async.bind` signature corrected from `(Async<'T> -> Async<'U>)` to `('T -> Async<'U>)` to match standard monadic bind semantics (same as `Task.bind`); the previous signature made the function effectively equivalent to direct application
     - refactor: simplify splitAt 'rest' taskSeq to use while!, removing redundant go2 mutable and manual MoveNextAsync pre-advance

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

@@ -28,6 +28,7 @@
     <Compile Include="TaskSeq.FindIndex.Tests.fs" />
     <Compile Include="TaskSeq.Find.Tests.fs" />
     <Compile Include="TaskSeq.Fold.Tests.fs" />
+    <Compile Include="TaskSeq.FoldUntil.Tests.fs" />
     <Compile Include="TaskSeq.Scan.Tests.fs" />
     <Compile Include="TaskSeq.MapFold.Tests.fs" />
     <Compile Include="TaskSeq.Reduce.Tests.fs" />

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

@@ -0,0 +1,226 @@
+module TaskSeq.Tests.FoldUntil
+
+open Xunit
+open FsUnit.Xunit
+
+open FSharp.Control
+
+//
+// TaskSeq.foldUntil
+// TaskSeq.foldUntilAsync
+//
+
+module EmptySeq =
+    [<Fact>]
+    let ``Null source is invalid`` () =
+        assertNullArg
+        <| fun () -> TaskSeq.foldUntil (fun _ _ -> Continue 42) 0 null
+
+        assertNullArg
+        <| fun () -> TaskSeq.foldUntilAsync (fun _ _ -> Task.fromResult (Continue 42)) 0 null
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-foldUntil returns initial state when empty`` variant = task {
+        let! result =
+            Gen.getEmptyVariant variant
+            |> TaskSeq.foldUntil (fun _ item -> Continue(item + 1)) -1
+
+        result |> should equal -1
+    }
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-foldUntilAsync returns initial state when empty`` variant = task {
+        let! result =
+            Gen.getEmptyVariant variant
+            |> TaskSeq.foldUntilAsync (fun _ item -> task { return Continue(item + 1) }) -1
+
+        result |> should equal -1
+    }
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-foldUntil does not call folder when empty`` variant = task {
+        let mutable called = false
+
+        let! _ =
+            Gen.getEmptyVariant variant
+            |> TaskSeq.foldUntil
+                (fun state _ ->
+                    called <- true
+                    Continue state)
+                0
+
+        called |> should be False
+    }
+
+    [<Theory; ClassData(typeof<TestEmptyVariants>)>]
+    let ``TaskSeq-foldUntilAsync does not call folder when empty`` variant = task {
+        let mutable called = false
+
+        let! _ =
+            Gen.getEmptyVariant variant
+            |> TaskSeq.foldUntilAsync
+                (fun state _ -> task {
+                    called <- true
+                    return Continue state
+                })
+                0
+
+        called |> should be False
+    }
+
+module Functionality =
+    [<Fact>]
+    let ``TaskSeq-foldUntil with all Continue behaves like fold`` () = task {
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntil (fun acc item -> Continue(acc + item)) 0
+
+        result |> should equal 15
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntilAsync with all Continue behaves like foldAsync`` () = task {
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntilAsync (fun acc item -> task { return Continue(acc + item) }) 0
+
+        result |> should equal 15
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntil is left-associative like fold`` () = task {
+        let! result =
+            TaskSeq.ofList [ "b"; "c"; "d" ]
+            |> TaskSeq.foldUntil (fun acc item -> Continue(acc + item)) "a"
+
+        result |> should equal "abcd"
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntilAsync is left-associative like foldAsync`` () = task {
+        let! result =
+            TaskSeq.ofList [ "b"; "c"; "d" ]
+            |> TaskSeq.foldUntilAsync (fun acc item -> task { return Continue(acc + item) }) "a"
+
+        result |> should equal "abcd"
+    }
+
+module Halt =
+    [<Fact>]
+    let ``TaskSeq-foldUntil Halt on first element stops immediately`` () = task {
+        let mutable callCount = 0
+
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntil
+                (fun _ item ->
+                    callCount <- callCount + 1
+                    Halt item)
+                0
+
+        result |> should equal 1
+        callCount |> should equal 1
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntilAsync Halt on first element stops immediately`` () = task {
+        let mutable callCount = 0
+
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntilAsync
+                (fun _ item -> task {
+                    callCount <- callCount + 1
+                    return Halt item
+                })
+                0
+
+        result |> should equal 1
+        callCount |> should equal 1
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntil halts mid-sequence, preserving halt state`` () = task {
+        // Sum until running total exceeds 5, then halt with the overshoot.
+        let mutable callCount = 0
+
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntil
+                (fun acc item ->
+                    callCount <- callCount + 1
+                    let next = acc + item
+                    if next > 5 then Halt next else Continue next)
+                0
+
+        // 1, 3, 6 — halts on the third element (6 > 5)
+        result |> should equal 6
+        callCount |> should equal 3
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntilAsync halts mid-sequence, preserving halt state`` () = task {
+        let mutable callCount = 0
+
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntilAsync
+                (fun acc item -> task {
+                    callCount <- callCount + 1
+                    let next = acc + item
+                    return if next > 5 then Halt next else Continue next
+                })
+                0
+
+        result |> should equal 6
+        callCount |> should equal 3
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntil does not enumerate past the halt element`` () = task {
+        // Source has a side effect per pulled element; halt on the 2nd.
+        let mutable pulled = 0
+
+        let source = taskSeq {
+            for i in 1..5 do
+                pulled <- pulled + 1
+                yield i
+        }
+
+        let! _ =
+            source
+            |> TaskSeq.foldUntil (fun _ item -> if item = 2 then Halt item else Continue item) 0
+
+        // The source yielded 1 (Continue), then 2 (Halt) — we must not pull 3.
+        pulled |> should equal 2
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntilAsync does not enumerate past the halt element`` () = task {
+        let mutable pulled = 0
+
+        let source = taskSeq {
+            for i in 1..5 do
+                pulled <- pulled + 1
+                yield i
+        }
+
+        let! _ =
+            source
+            |> TaskSeq.foldUntilAsync (fun _ item -> task { return if item = 2 then Halt item else Continue item }) 0
+
+        pulled |> should equal 2
+    }
+
+    [<Fact>]
+    let ``TaskSeq-foldUntil on last-element Halt is equivalent to fold`` () = task {
+        let! result =
+            TaskSeq.ofList [ 1; 2; 3; 4; 5 ]
+            |> TaskSeq.foldUntil
+                (fun acc item ->
+                    let next = acc + item
+                    if item = 5 then Halt next else Continue next)
+                0
+
+        result |> should equal 15
+    }

src/FSharp.Control.TaskSeq/TaskSeq.fs

@@ -540,6 +540,10 @@ type TaskSeq private () =
     static member compareWithAsync comparer source1 source2 = Internal.compareWithAsync comparer 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 foldUntil folder state source = Internal.foldUntil (FoldUntilAction folder) state source
+
+    static member foldUntilAsync folder state source = Internal.foldUntil (AsyncFoldUntilAction folder) state source
+
     static member scan folder state source = Internal.scan (FolderAction folder) state source
     static member scanAsync folder state source = Internal.scan (AsyncFolderAction folder) state source
     static member reduce folder source = Internal.reduce (FolderAction folder) source

src/FSharp.Control.TaskSeq/TaskSeq.fsi

@@ -1865,6 +1865,42 @@ 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 in the task sequence, threading an
+    /// accumulator of type <paramref name="'State" /> through the computation, with the ability to stop
+    /// early. The folder returns <c>Continue newState</c> to keep consuming or <c>Halt newState</c> to
+    /// stop iteration immediately; in either case the state is updated. When the folder halts, no further
+    /// elements of the input are enumerated.
+    /// If the folder function <paramref name="folder" /> is asynchronous, consider using
+    /// <see cref="TaskSeq.foldUntilAsync" />.
+    /// </summary>
+    ///
+    /// <param name="folder">A function that updates the state and decides whether to continue or halt.</param>
+    /// <param name="state">The initial state.</param>
+    /// <param name="source">The input sequence.</param>
+    /// <returns>The state object, either after the folder halts or after the whole sequence has been consumed.</returns>
+    /// <exception cref="T:ArgumentNullException">Thrown when the input task sequence is null.</exception>
+    static member foldUntil:
+        folder: ('State -> 'T -> FoldStep<'State>) -> state: 'State -> source: TaskSeq<'T> -> Task<'State>
+
+    /// <summary>
+    /// Applies the asynchronous function <paramref name="folder" /> to each element in the task sequence,
+    /// threading an accumulator of type <paramref name="'State" /> through the computation, with the ability
+    /// to stop early. The folder returns <c>Continue newState</c> to keep consuming or <c>Halt newState</c>
+    /// to stop iteration immediately; in either case the state is updated. When the folder halts, no further
+    /// elements of the input are enumerated.
+    /// If the folder function <paramref name="folder" /> is synchronous, consider using
+    /// <see cref="TaskSeq.foldUntil" />.
+    /// </summary>
+    ///
+    /// <param name="folder">A function that updates the state and decides whether to continue or halt.</param>
+    /// <param name="state">The initial state.</param>
+    /// <param name="source">The input sequence.</param>
+    /// <returns>The state object, either after the folder halts or after the whole sequence has been consumed.</returns>
+    /// <exception cref="T:ArgumentNullException">Thrown when the input task sequence is null.</exception>
+    static member foldUntilAsync:
+        folder: ('State -> 'T -> #Task<FoldStep<'State>>) -> state: 'State -> source: TaskSeq<'T> -> Task<'State>
+
     /// <summary>
     /// Like <see cref="TaskSeq.fold" />, but returns the sequence of intermediate results and the final result.
     /// The first element of the output sequence is always the initial state. If the input task sequence

src/FSharp.Control.TaskSeq/TaskSeqInternal.fs

@@ -34,6 +34,19 @@ type internal FolderAction<'T, 'State, 'TaskState when 'TaskState :> Task<'State
     | FolderAction of state_action: ('State -> 'T -> 'State)
     | AsyncFolderAction of async_state_action: ('State -> 'T -> 'TaskState)
 
+/// The result of a single folder step in <see cref="TaskSeq.foldUntil" /> or
+/// <see cref="TaskSeq.foldUntilAsync" />: <c>Continue</c> threads the new state and
+/// keeps consuming, <c>Halt</c> records the final state and stops iteration.
+[<Struct>]
+type FoldStep<'State> =
+    | Continue of continue_state: 'State
+    | Halt of halt_state: 'State
+
+[<Struct>]
+type internal FoldUntilAction<'T, 'State, 'TaskStep when 'TaskStep :> Task<FoldStep<'State>>> =
+    | FoldUntilAction of fold_until_action: ('State -> 'T -> FoldStep<'State>)
+    | AsyncFoldUntilAction of async_fold_until_action: ('State -> 'T -> 'TaskStep)
+
 [<Struct>]
 type internal ChooserAction<'T, 'U, 'TaskOption when 'TaskOption :> Task<'U option>> =
     | TryPick of try_pick: ('T -> 'U option)
@@ -409,6 +422,44 @@ module internal TaskSeqInternal =
             return result
         }
 
+    let foldUntil folder initial (source: TaskSeq<_>) =
+        checkNonNull (nameof source) source
+
+        task {
+            use e = source.GetAsyncEnumerator CancellationToken.None
+            let mutable result = initial
+            let mutable running = true
+
+            match folder with
+            | FoldUntilAction folder ->
+                while running do
+                    let! hasNext = e.MoveNextAsync()
+
+                    if hasNext then
+                        match folder result e.Current with
+                        | Continue s -> result <- s
+                        | Halt s ->
+                            result <- s
+                            running <- false
+                    else
+                        running <- false
+
+            | AsyncFoldUntilAction folder ->
+                while running do
+                    let! hasNext = e.MoveNextAsync()
+
+                    if hasNext then
+                        match! folder result e.Current with
+                        | Continue s -> result <- s
+                        | Halt s ->
+                            result <- s
+                            running <- false
+                    else
+                        running <- false
+
+            return result
+        }
+
     let scan folder initial (source: TaskSeq<_>) =
         checkNonNull (nameof source) source