   (let [matched (filter #(= (:lifecycle/task %) task-name) lifecycles)]
     (reduce
      (fn [f lifecycle]
-       (comp (operation/kw->fn (get lifecycle kw)) f))
+       (comp (fn [x] ((operation/kw->fn (get lifecycle kw)) x lifecycle)) f))
      identity
      matched)))
 

  (merge event (p-ext/write-batch event)))

(defn close-batch-resources [event]
  (merge event (l-ext/close-batch-resources* event)))

(defn launch-aux-threads!
  [messenger event outbox-ch seal-ch completion-ch task-kill-ch]

                           :onyx.core/state (atom {:timeout-pool buckets})}

            ex-f (fn [e] (handle-exception e restart-ch outbox-ch job-id))
            pipeline-data (merge pipeline-data
                                 ((:onyx.core/compiled-pre-fn pipeline-data)
                                  (l-ext/inject-lifecycle-resources* pipeline-data)))]

        (while (and (first (alts!! [kill-ch task-kill-ch] :default true))
                    (not (:onyx.core/start-lifecycle? (munge-start-lifecycle pipeline-data))))
