   (f/collect-pure
    (fn [_ incoming]
      (let [res (vals (peek incoming))]
-       ;;(info :agg-incoming res)
+       ;;(debug :agg-incoming res)
        (delay
         (when (every? #(not= [nil] %) res)
           (->> res

          inferred (mapcat production adds)]
      (debug :additions adds)
      (doseq [t inferred]
        (info :add-triple t)
        (add-triple! g t))
      (swap! (:state vertex) update :val set/union in (set inferred)))))


  [spo]
  (f/collect-pure
   (fn [val incoming]
     ;;(debug :update-index spo incoming)
     (transduce
      (map (fn [[op id t]] [op id (nth t spo)]))
      (completing
