             restart-pred-fn (operation/resolve-restart-pred-fn task-map)
             ex-f (fn [e] (handle-exception log restart-pred-fn e restart-ch outbox-ch job-id))
             _ (while (and (first (alts!! [kill-ch task-kill-ch] :default true))
-                          (not (start-lifecycle? pipeline-data)))
+                          (not (munge-start-lifecycle pipeline-data)))
                 (Thread/sleep (arg-or-default :onyx.peer/peer-not-ready-back-off opts)))
 
             pipeline-data (-> pipeline-data

             (gen-lifecycle-id)
             (lc/invoke-before-batch compiled-before-batch-fn pipeline)
             (read-batch task-type replica peer-replica-view job-id pipeline)
             (lc/invoke-after-read-batch)
             (back-off-after-read)
             (tag-messages task-type replica peer-replica-view id)
             (add-messages-to-timeout-pool task-type state)
             (process-sentinel task-type pipeline monitoring)

(defn gen-lifecycle-id [event]
  (assoc event :onyx.core/lifecycle-id (uuid/random-uuid)))

(defn handle-backoff! [event]
  (let [batch (:onyx.core/batch event)]
    (when (and (= (count batch) 1)
               (= (:message (first batch)) :done))
      (Thread/sleep (:onyx.core/drained-back-off event)))))

(defn read-batch [task-type replica peer-replica-view job-id pipeline event]
  (if (and (= task-type :input) (:backpressure? peer-replica-view))
    (assoc event :onyx.core/batch '())
    (let [rets (merge event (p-ext/read-batch pipeline event))
          rets (lc/invoke-after-read-batch rets)]
      (handle-backoff! event)
      rets)))

(defn validate-ackable! [peers event]
  (when-not (seq peers)
