 (defn only-relevant-branches [flow task]
   (filter #(= (:flow/from %) task) flow))
 
-(defn compile-flow-conditions [flow-conditions task-name]
-  (let [conditions (only-relevant-branches flow-conditions task-name)]
+(defn compile-flow-conditions [flow-conditions task-name f]
+  (let [conditions (filter f (only-relevant-branches flow-conditions task-name))]
     (map
      (fn [condition]
        (assoc condition :flow/predicate (build-pred-fn (:flow/predicate condition) condition)))
-          conditions)))
+     conditions)))
+
+(defn compile-fc-norms [flow-conditions task-name]
+  (compile-flow-conditions flow-conditions task-name (comp not :flow/thrown-exception?)))
+
+(defn compile-fc-exs [flow-conditions task-name]
+  (compile-flow-conditions flow-conditions task-name :flow/thrown-exception?))
 
 (defn run-task-lifecycle [init-event kill-ch ex-f]
   (try
-    (loop [event init-event]
-      (when (first (alts!! [kill-ch] :default true))
-        (-> event
-            (inject-batch-resources)
-            (read-batch)
-            (decompress-batch)
-            (apply-fn)
-            (compress-batch)
-            (write-batch)
-            (close-batch-resources))
-        (recur init-event)))
+    (while (first (alts!! [kill-ch] :default true))
+      (-> init-event
+          (inject-batch-resources)
+          (read-batch)
+          (tag-messages)
+          (decompress-batch)
+          (start-message-timeout-monitors)
+          (try-complete-job)
+          (strip-sentinel)
+          (apply-fn)
+          (ack-messages)
+          (route-output-paths)
+          (compress-batch)
+          (write-batch)
+          (close-batch-resources)))
     (catch Exception e
       (ex-f e))))
 

        (= to-add :none) #{}
        :else (clojure.set/union (into #{} all) (into #{} to-add))))

(defn choose-output-paths [flow-conditions event new downstream]
  (if (seq flow-conditions)
    (reduce
     (fn [{:keys [paths exclusions] :as all} entry]
       (if ((:flow/predicate entry) [event new])
         (if (:flow/short-circuit? entry)
           (reduced {:paths (join-output-paths paths (:flow/to entry) downstream)
                     :exclusions (clojure.set/union (into #{} exclusions) (into #{} (:flow/exclude-keys entry)))})
           {:paths (join-output-paths paths (:flow/to entry) downstream)
            :exclusions (clojure.set/union (into #{} exclusions) (into #{} (:flow/exclude-keys entry)))})
         all))
     {:paths #{} :exclusions #{}}
     flow-conditions)
    {:paths downstream}))

(defn output-paths
  [{:keys [onyx.core/serialized-task onyx.core/compiled-flow-conditions] :as event}]
  (merge
   event
   {:onyx.core/result-paths
    (map
     (fn [segment]
       (choose-output-paths compiled-flow-conditions event segment
                            (keys (:egress-ids serialized-task))))
     (:onyx.core/results event))}))

(defn inject-batch-resources [event]
  (let [cycle-params {:onyx.core/lifecycle-id (java.util.UUID/randomUUID)}]
    (merge event cycle-params (l-ext/inject-batch-resources* event))))

(defn read-batch [event]
  (let [rets (tag-each-message (merge event (p-ext/read-batch event)))]
    (when (= (:onyx/type (:onyx.core/task-map event)) :input)
      (doseq [m (:onyx.core/batch rets)]
        (go (try (<! (timeout (or (:onyx/pending-timeout (:onyx.core/task-map event)) 60000)))
                 (when (p-ext/pending? rets (:id m))
                   (p-ext/retry-message event (:id m)))
                 (catch Exception e
                   (taoensso.timbre/warn e))))))
    rets))

(defn decompress-batch [event]
  ;; TODO: Messages are already decompressed, need to rework the interface.
  (let [rets (merge event {:onyx.core/decompressed (:onyx.core/batch event)})]
    (if (sentinel-found? rets)
      (do (if (p-ext/drained? rets)
            (complete-job rets)
            (p-ext/retry-message rets (sentinel-id rets)))
          (strip-sentinel rets))
      rets)))

(defn apply-fn* [{:keys [onyx.core/batch onyx.core/decompressed] :as event} batch-f single-f]
  (if (seq decompressed)
    (let [batch-result (batch-f event)
          rets
          (merge event
                 (reduce
                  (fn [rets [raw thawed]]
                    (let [new-segments (single-f event thawed batch-result)
                          new-ack-vals (map (fn [x] (acker/gen-ack-value)) new-segments)
                          tagged (apply acker/prefuse-vals new-ack-vals)
                          results (map (fn [segment ack-val]
                                         {:id (:id raw)
                                          :acker-id (:acker-id raw)
                                          :completion-id (:completion-id raw)
                                          :message segment
                                          :ack-val ack-val})
                                       new-segments new-ack-vals)]
                      (-> rets
                          (update-in [:onyx.core/results] concat results)
                          (assoc-in [:onyx.core/children raw] tagged))))
                  {:onyx.core/results [] :onyx.core/children {}}
                  (map vector batch decompressed)))]
      (ack-messages rets)
      (merge rets (output-paths rets)))
    (merge event {:onyx.core/results []})))

(defn apply-fn [event]
  (if (:onyx/batch? (:onyx.core/task-map event))
    (apply-fn*
     event
     (fn [{:keys [onyx.core/decompressed] :as event}]
       (let [new-segments (p-ext/apply-fn event (map :message decompressed))]
         (if (sequential? new-segments) new-segments (vector new-segments))))
     (fn [event thawed batch-result]
       batch-result))
    (apply-fn*
     event
     (fn [_] nil)
     (fn [event thawed batch-result]
       (let [new-segments (p-ext/apply-fn event (:message thawed))]
         (if (sequential? new-segments) new-segments (vector new-segments)))))))

(defn compress-batch [event]
  (merge event (p-ext/compress-batch event)))

(defn add-completion-id [event m]
  (assoc m :completion-id (:onyx.core/id event)))

(defn sentinel-found? [event]
  (seq (filter (partial = :done) (map :message (:onyx.core/decompressed event)))))

