   (if (= :input (:onyx/type task-map))
     (p/start (instantiate-plugin event) event)))
 
-(defn build-output-pipeline [{:keys [onyx.core/task-map] :as event}]
+(defn build-output-pipeline [event]
   (cond (source-task? event) 
         (p/start (mo/new-messenger-output event) event)
 

                state-fn-builders)))

(defn build-task-fns
  [{:keys [onyx.core/task-map] :as event}]
  (let [task-types (cond-> #{(node-type event)}
                     (windowed-task? event) (conj :windowed))
        phase-order [:recover :start-iteration :barriers :process-batch :heartbeat]]


(defn recover-state
  [state]
  (let [{:keys [onyx.core/peer-opts] :as event} (get-event state)
        _ (cleanup-previous-state-store! state)
        state-serializers (onyx.state.serializers.utils/event->state-serializers event)
        db-name (db-name event)
