                 sinks (into-array
                        (for [[_ clause-type clause-id name] clauses
                              :when (not= clause-type "when")]
-                         (sink-of rule-id clause-type clause-id name var->ix)))]
-            (swap! rule->flow assoc agg-rule-id (AggregateFlow. index group-len limit-ix ascending? agg-ixes agg-funs sinks))
+                         (sink-of rule-id clause-type clause-id (str "delta-" name) var->ix)))]
+            (swap! rule->flow assoc agg-rule-id (AggregateFlow. index delta-index group-len limit-ix ascending? agg-ixes agg-funs sinks))
             (swap! kind->name->rules update-in ["know" (str "delta-" agg-index-id)] conj agg-rule-id)))))
 
     (doseq [[rule-id clauses] @rule-id->clauses]

                              (let [output (aget inputs i)]
                                (dotimes [j (alength aggs)]
                                  (apush output (aget aggs j)))
                                (apush facts&vals output)
                                (apush facts&vals 1)))
                            (aclear inputs)
                            (set!! current-key key)
                            (set!! current-limit (aget key limit-ix))

         (dotimes [i (alength sinks)]
           (.run (aget sinks i) kn rule->dirty? kind->name->rules facts&vals)))))

(deftype AggregateFlow [index delta-index group-len limit-ix ascending? agg-ixes agg-funs sinks]
  Object
  (run [this kn rule->dirty? kind->name->rules]
       ;; TODO this is kind of hacky - we shouldn't be directly messing with input state or storing things in anon fields

       ;; update input
       (.foreach delta-index
                 (fn [k v]
                   (.update index k v)))
       (.reset delta-index)

       ;; figure out new output
       (let [current-key nil
             current-limit nil
             current-index 1
             inputs #js []
             aggs (make-array (alength agg-ixes))
             new-output (btree/tree 10 (+ group-len 1 (alength aggs)))
             push-input (fn [key]
                          (when (nil? current-key)
                            (set!! current-key key)
