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: 
      // terminal functions
  
  
      /// <summary>Applies a local cloud function to each element of the CloudStream, threading an accumulator argument through the computation. If the input function is f and the elements are i0...iN, then this function computes f (... (f s i0)...) iN.</summary>
      /// <param name="folder">A local cloud function that updates the state with each element from the CloudStream.</param>
      /// <param name="combiner">A local cloud function that combines partial states into a new state.</param>
      /// <param name="state">A local cloud function that produces the initial state.</param>
      /// <param name="stream">The input CloudStream.</param>
      /// <returns>The final result.</returns>
      let inline foldLocal (folder : 'State -> 'T -> Local<'State>) (combiner : 'State -> 'State -> Local<'State>) 
                           (state : unit -> Local<'State>) (stream : CloudStream<'T>) : Cloud<'State> =
          let collectorf = local {  
              let results = new List<'State ref>()
              let! ctx = Cloud.GetExecutionContext()
              let run a = Async.RunSync(Cloud.ToAsync(a, ctx.Resources,ctx.CancellationToken))
              return
                { new Collector<'T, 'State> with
                  member self.DegreeOfParallelism = stream.DegreeOfParallelism 
                  member self.Iterator() = 
                      let accRef = ref <| run (state ())
                      results.Add(accRef)
                      {   Index = ref -1;
                          Func = (fun value -> accRef := run (folder !accRef value));
                          Cts = new CancellationTokenSource() }
                  member self.Result = 
                      let mutable acc = run (state ())
                      for result in results do
                              acc <- run (combiner acc !result)
                      acc }
          }
          stream.Apply collectorf (fun x -> local { return x }) (fun values -> local { return Array.reduce combiner values })
  
      /// <summary>Applies a function to each element of the CloudStream, threading an accumulator argument through the computation. If the input function is f and the elements are i0...iN, then this function computes f (... (f s i0)...) iN.</summary>
      /// <param name="folder">A function that updates the state with each element from the CloudStream.</param>
  
negative.code:                 }
                  stream.Apply collectorf' projection combiner }
  
      /// <summary>Transforms each element of the input CloudStream to a new stream and flattens its elements.</summary>
      /// <param name="f">A function to transform items from the input CloudStream.</param>
      /// <param name="stream">The input CloudStream.</param>
      /// <returns>The result CloudStream.</returns>
      let inline collect (f : 'T -> Stream<'R>) (stream : CloudStream<'T>) : CloudStream<'R> =
          collectGen (fun ctx x -> f x) stream 
  
      /// <summary>Transforms each element of the input CloudStream to a new stream and flattens its elements using a locally executing cloud function.</summary>
      /// <param name="f">A locally executing cloud function to transform items from the input CloudStream.</param>
      /// <param name="stream">The input CloudStream.</param>
      /// <returns>The result CloudStream.</returns>
      let inline collectLocal (f : 'T -> Local<Stream<'R>>) (stream : CloudStream<'T>) : CloudStream<'R> =
          collectGen (fun ctx x -> f x |> run ctx) stream 
  
      let inline private filterGen (predicate : ExecutionContext -> 'T -> bool) (stream : CloudStream<'T>) : CloudStream<'T> =
          { new CloudStream<'T> with
              member self.DegreeOfParallelism = stream.DegreeOfParallelism
              member self.Apply<'S, 'R> (collectorf : Local<Collector<'T, 'S>>) (projection : 'S -> Local<'R>) combiner =
                  let collectorf' = local {
                      let! collector = collectorf
                      let! ctx = Cloud.GetExecutionContext()
                      return { new Collector<'T, 'S> with
                          member self.DegreeOfParallelism = collector.DegreeOfParallelism
                          member self.Iterator() = 
                              let { Func = iter } as iterator = collector.Iterator()
                              {   Index = iterator.Index; 
                                  Func = (fun value -> if predicate ctx value then iter value else ());
                                  Cts = iterator.Cts }
                          member self.Result = collector.Result }
                  }
                  stream.Apply collectorf' projection combiner }
  
      /// <summary>Filters the elements of the input CloudStream.</summary>
      /// <param name="predicate">A function to test each source element for a condition.</param>
      /// <param name="stream">The input CloudStream.</param>
      /// <returns>The result CloudStream.</returns>
      let inline filter (predicate : 'T -> bool) (stream : CloudStream<'T>) : CloudStream<'T> =
          filterGen (fun ctx x -> predicate x) stream 
  
      /// <summary>Filters the elements of the input CloudStream using a locally executing cloud function.</summary>
      /// <param name="predicate">A locally executing cloud function to test each source element for a condition.</param>
      /// <param name="stream">The input CloudStream.</param>
      /// <returns>The result CloudStream.</returns>
      let inline filterLocal (predicate : 'T -> Local<bool>) (stream : CloudStream<'T>) : CloudStream<'T> =
          filterGen (fun ctx x -> predicate x |> run ctx) stream 
  
      /// <summary>Returns a cloud stream with a new degree of parallelism.</summary>
      /// <param name="degreeOfParallelism">The degree of parallelism.</param>
  