                                                                  [new-clause-id "variable-b" "variable" val]})
                   (swap! rule-id->clauses update-in [rule-id] conj [rule-id "when" new-clause-id "=variable"]))
                 (swap! var->key assoc val key)))))))
+
+    ;; rewrite aggregates
+    (doseq [[rule-id clauses] @rule-id->clauses]
+      (when (seq (@rule-id->has-agg rule-id))
+        (let [agg-rule-id (str rule-id "-agg-rule")
+              agg-clause-id (str rule-id "-agg-clause")
+              agg-index-id (str rule-id "-agg-index")]
+
+          ;; remove output clauses
+          (doseq [[_ clause-type clause-id name] clauses
+                  :when (not= clause-type "when")]
+            (swap! rule-id->clauses update-in [rule-id] disj [rule-id clause-type clause-id name]))
+
+          ;; add a clause to output used vars to agg-index-ix
+          (swap! rule-id->clauses update-in [rule-id] conj [rule-id "know" agg-clause-id agg-index-id])
+          (doseq [[_ var] (@rule-id->group-by rule-id)]
+            (swap! clause-id->fields update-in [agg-clause-id] conj [agg-clause-id "variable" var var]))
+          (doseq [[_ in-var agg-fun out-var] (@rule-id->agg-over rule-id)]
+            (swap! clause-id->fields update-in [agg-clause-id] conj [agg-clause-id "variable" in-var in-var]))
+
+          ;; create an aggregate flow to read from agg-index-id
+          (let [group-by-vars (for [[_ var] (@rule-id->group-by rule-id)]
+                                var)
+                agg-in-vars (for [[_ in-var agg-fun out-var] (@rule-id->agg-over rule-id)]
+                              in-var)
+                agg-out-vars (for [[_ in-var agg-fun out-var] (@rule-id->agg-over rule-id)]
+                               out-var)
+                in-vars (into-array (concat group-by-vars agg-in-vars))
+                out-vars (into-array (concat group-by-vars agg-in-vars agg-out-vars))
+                var->ix (into {}
+                              (for [i (range (alength out-vars))]
+                                [(aget out-vars i) i]))
+                index (.get-or-create-index kn "know" agg-index-id in-vars)
+                group-len (count group-by-vars)
+                agg-ixes (into-array
+                          (for [[_ in-var agg-fun out-var] (@rule-id->agg-over rule-id)]
+                            (var->ix in-var)))
+                agg-funs (into-array
+                          (for [[_ in-var agg-fun out-var] (@rule-id->agg-over rule-id)]
+                            (aget js/aurora.aggregates agg-fun)))
+                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 agg-ixes agg-funs sinks))
+            (swap! kind->name->rules update-in ["know" agg-index-id] conj agg-rule-id)))))
 
     (doseq [[rule-id clauses] @rule-id->clauses]
       (let [var->when-count (atom {})]

        name->transient? (atom {})
        clauses (.keys (.get-or-create-index kn "know" "clauses" #js ["rule-id" "when|know|remember|forget" "clause-id" "name"]))
        fields (.keys (.get-or-create-index kn "know" "clause-fields" #js ["clause-id" "constant|variable" "key" "val"]))
        rule-id->clauses (atom (into {} (for [[k vs] (group-by #(nth % 0) clauses)] [k (set (map vec vs))])))
        clause-id->fields (atom (into {} (for [[k vs] (group-by #(nth % 0) fields)] [k (set (map vec vs))])))]

    ;; rewrite clauses
    (doseq [[rule-id clauses] @rule-id->clauses]


;; FLOWS

(deftype Sink [kind name fields]
  Object
  (run [this kn rule->dirty? kind->name->rules facts&vals]
       (when (true? (.update-facts kn kind name fields facts&vals))
         (let [dirtied-rules (aget kind->name->rules kind name)]
           (dotimes [j (alength dirtied-rules)]
             (aset rule->dirty? (aget dirtied-rules j) true))))))

(deftype SolverFlow [solver sinks]
  Object
  (run [this kn rule->dirty? kind->name->rules]
       (.reset solver)
       (let [facts&vals (.elems solver)]
         (dotimes [i (alength sinks)]
           (.run (aget sinks i) kn rule->dirty? kind->name->rules facts&vals)))))

(deftype AggregateFlow [index group-len agg-ixes agg-funs sinks]
  Object
  (run [this kn rule->dirty? kind->name->rules]
       (let [current-key nil
             inputs #js []
             aggs (make-array (alength agg-ixes))
             facts&vals #js []
             push-input (fn [key]
                          (when (nil? current-key)
                            (set!! current-key key))
                          (when (btree/prefix-not= key current-key group-len)
                            (dotimes [i (alength aggs)]
                              (aset aggs i ((aget agg-funs i) (aget agg-ixes i) inputs)))
                            (dotimes [i (alength inputs)]
                              (let [output (aclone (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))
                          (apush inputs key))]
         (.foreach 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->transient?]
  Object
