   be used as a fallback, and then two more values will be dropped before fresh
   values start being emitted."
   [xform base sig]
-  (let [reducer ((comp (core/map :value)
+  (let [reducer ((comp (core/map value)
                        xform
                        (core/map ->Fresh))
                  conj)

        msgs->events (comp cat
                           (filter fresh?)
                           (core/map (fn [msg]
                                       (->Event topic (:value msg)))))
        events-channel-fn (fn [live-graph _]
                            (async/tap (get (:mult-map live-graph) source)
                                       (chan 1 msgs->events)))]

    msg-batches
    (let [max-count (reduce max (core/map core/count msg-batches))
          pad (fn [msgs]
                (->> [msgs (-> msgs last value ->Cached repeat)]
                     (into [] (comp cat (take max-count)))))]
      (core/map pad msg-batches))))

