anchor.diff:      let ofArray (source : 'T []) : CloudStream<'T> =
           { new CloudStream<'T> with
               member self.DegreeOfParallelism = None
  -            member self.Apply<'S, 'R> (collectorf : Local<Collector<'T, 'S>>) (projection : 'S -> Local<'R>) (combiner : 'R -> 'R -> 'R) =
  +            member self.Apply<'S, 'R> (collectorf : Local<Collector<'T, 'S>>) (projection : 'S -> Local<'R>) (combiner : 'R [] -> Local<'R>) =
                   cloud {
                       let! collector = collectorf 
                       let! workerCount = 
  
positive.code:         // Phase 1
          let shuffling = 
              cloud {
                  let combiner' (left : _ []) (right : _ []) =  Array.append left right 
                  let! totalWorkers = match stream.DegreeOfParallelism with Some n -> local { return n } | None -> Cloud.GetWorkerCount()
                  let! keyValueArray = stream.Apply (collectorf totalWorkers) 
                                                    (fun keyValues -> local {
  
negative.code:                                     let! collector = collectorf
                                      return! projection collector.Result
                                  else
                                      return! combiner (result.ToArray())
                              }
  
                          let partitions = sources |> Seq.toArray |> Partitions.ofArray workerCount
                          let! results = partitions |> Array.map (fun cfiles -> createTask cfiles collectorf) |> Cloud.Parallel
                          return! combiner results
                  } }
  
      /// <summary>
  