          (dotimes [i (alength sinks)]
            (.run (aget sinks i) kn rule->dirty? kind->name->rules facts&vals)))))
 
-(deftype AggregateFlow [index group-len limit-ix ascending? agg-ixes agg-funs sinks]
+(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))
-             facts&vals #js []
+             new-output (btree/tree 10 (+ group-len 1 (alength aggs)))
              push-input (fn [key]
                           (when (nil? current-key)
                             (set!! current-key key)

           (.foreach index push-input)
           (.foreach-reverse index push-input))
         (push-input (btree/greatest-key group-len))
         (dotimes [i (alength sinks)]
           (.run (aget sinks i) kn rule->dirty? kind->name->rules facts&vals)))))

(deftype Flows [rules rule->flow rule->dirty? kind->name->rules name->lifetime]
  Object

                              (for [i (range (alength out-vars))]
                                [(aget out-vars i) i]))
                index (.get-or-create-index kn "know" agg-index-id in-vars)
                delta-index (.get-or-create-index kn "know" (str "delta-" agg-index-id) in-vars)
                group-len (count (conj (set group-by-vars) limit))
                limit-ix (var->ix limit)
                ascending? (= ascending|descending "ascending")
