anchor.diff:                          Func =
                               (fun value -> 
                                       let mutable grouping = Unchecked.defaultof<_>
  -                                    let key = projection value
  +                                    let key = projection ctx value
                                       if dict.TryGetValue(key, &grouping) then
                                           let acc = grouping
  -                                        lock grouping (fun () -> acc := folder !acc value) 
  +                                        lock grouping (fun () -> acc := folder ctx !acc value) 
                                       else
  -                                        grouping <- ref <| state ()
  +                                        grouping <- ref <| state ctx
                                           if not <| dict.TryAdd(key, grouping) then
                                               dict.TryGetValue(key, &grouping) |> ignore
                                           let acc = grouping
  -                                        lock grouping (fun () -> acc := folder !acc value) 
  +                                        lock grouping (fun () -> acc := folder ctx !acc value) 
                                       ());
                           Cts = new CancellationTokenSource() }
                   member self.Result = 
  
positive.code: 
      //#region Intermediate functions
  
      /// <summary>Transforms each element of the input CloudStream.</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 map (f : 'T -> 'R) (stream : CloudStream<'T>) : CloudStream<'R> =
          { new CloudStream<'R> with
              member self.DegreeOfParallelism = stream.DegreeOfParallelism
              member self.Apply<'S, 'Result> (collectorf : Local<Collector<'R, 'S>>) (projection : 'S -> Local<'Result>) combiner =
                  let collectorf' = local {
                      let! collector = collectorf
                      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 -> iter (f value));
                                  Cts = iterator.Cts }
                          member self.Result = collector.Result  }
                  }
                  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> =
          { new CloudStream<'R> with
              member self.DegreeOfParallelism = stream.DegreeOfParallelism
              member self.Apply<'S, 'Result> (collectorf : Local<Collector<'R, 'S>>) (projection : 'S -> Local<'Result>) combiner =
                  let collectorf' = local {
                      let! collector = collectorf
                      return 
                        { new Collector<'T, 'S> with
                          member self.DegreeOfParallelism = collector.DegreeOfParallelism
  
negative.code: 
      // terminal functions
  
      let inline private foldGen (folder : ExecutionContext -> 'State -> 'T -> 'State) (combiner : ExecutionContext -> 'State -> 'State -> 'State) 
                                 (state : ExecutionContext -> 'State) (stream : CloudStream<'T>) : Cloud<'State> =
          let collectorf = local {  
              let results = new List<'State ref>()
              let! ctx = Cloud.GetExecutionContext()
              return
                { new Collector<'T, 'State> with
                  member self.DegreeOfParallelism = stream.DegreeOfParallelism 
                  member self.Iterator() = 
                      let accRef = ref <| state ctx 
                      results.Add(accRef)
                      {   Index = ref -1;
                          Func = (fun value -> accRef := folder ctx !accRef value);
                          Cts = new CancellationTokenSource() }
                  member self.Result = 
                      let mutable acc = state ctx
                      for result in results do
                              acc <- combiner ctx acc !result
                      acc }
          }
          stream.Apply 
             collectorf 
             (fun x -> local { return x }) 
             (fun values -> local { 
                 let! ctx = Cloud.GetExecutionContext()
                 return Array.reduce (combiner ctx) values })
  
  
  
      /// <summary>Applies a locally executing 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 locally executing cloud function that updates the state with each element from the CloudStream.</param>
      /// <param name="combiner">A locally executing cloud function that combines partial states into a new state.</param>
      /// <param name="state">A locally executing 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> =
          foldGen (fun ctx x y -> run ctx (folder x y)) (fun ctx x y -> run ctx (combiner x y)) (fun ctx -> run ctx (state ())) stream
  
      /// <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>
  