anchor.diff:                              {   Index = iterator.Index; 
                                   Func = 
                                       (fun value -> 
  -                                        let (Stream streamf) = f value
  +                                        let (Stream streamf) = f ctx value
                                           let cts = CancellationTokenSource.CreateLinkedTokenSource(iterator.Cts.Token)
                                           let { Bulk = bulk; Iterator = _ } = streamf { Complete = (fun () -> ()); Cont = iter; Cts = cts } in bulk ());
                                   Cts = iterator.Cts }
  
positive.code:                 }  }
  
  
      /// <summary>
      /// Applies a key-generating function to each element of a CloudStream and return a CloudStream yielding unique keys and their number of occurrences in the original sequence.
      /// </summary>
      /// <param name="projection">A function that maps items from the input CloudStream to keys.</param>
      /// <param name="stream">The input CloudStream.</param>
      let inline countBy (projection : 'T -> 'Key) (stream : CloudStream<'T>) : CloudStream<'Key * int64> =
          foldBy projection (fun state _ -> state + 1L) (+) (fun () -> 0L) stream
  
      /// <summary>Runs the action on each element. The actions are not necessarily performed in order.</summary>
      /// <param name="stream">The input CloudStream.</param>
  
negative.code:     /// <returns>The final result.</returns>
      let inline fold (folder : 'State -> 'T -> 'State) (combiner : 'State -> 'State -> 'State) 
                      (state : unit -> 'State) (stream : CloudStream<'T>) : Cloud<'State> =
          foldGen (fun ctx x y -> folder x y) (fun ctx x y -> combiner x y) (fun ctx -> state ()) stream
  
      let inline private foldByGen (projection : ExecutionContext -> 'T -> 'Key) 
                                   (folder : ExecutionContext -> 'State -> 'T -> 'State) 
                                   (combiner : ExecutionContext -> 'State -> 'State -> 'State) 
                                   (state : ExecutionContext -> 'State) (stream : CloudStream<'T>) : CloudStream<'Key * 'State> = 
          let collectorf (totalWorkers : int) = local {
              let dict = new ConcurrentDictionary<'Key, 'State ref>()
              let! ctx = Cloud.GetExecutionContext()
              return
                { new Collector<'T,  seq<int * seq<'Key * 'State>>> with
                  member self.DegreeOfParallelism = stream.DegreeOfParallelism 
  