 (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))))
 

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

(defn tag-each-message [event]
  (if (= (:onyx/type (:onyx.core/task-map event)) :input)
    (let [event (update-in event 
                           [:onyx.core/batch]
                           (fn [batch]
                             (map (comp (partial add-completion-id event)
                                        (partial add-acker-id event)
                                        add-ack-value)
                                  batch)))]
      (doseq [raw-segment (:onyx.core/batch event)]
        (extensions/internal-ack-message
         (:onyx.core/messenger event)
         event
         (operation/peer-link event (:acker-id raw-segment) :acker-peer-site)
         (:id raw-segment)
         (:completion-id raw-segment)
         (:ack-val raw-segment)))
      event)
    event))

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


                           :onyx.core/catalog catalog
                           :onyx.core/workflow (extensions/read-chunk log :workflow job-id)
                           :onyx.core/flow-conditions flow-conditions
                           :onyx.core/compiled-norm-fcs (compile-fc-norms flow-conditions (:name task))
                           :onyx.core/compiled-ex-fcs (compile-fc-exs flow-conditions (:name task))
                           :onyx.core/task-map catalog-entry
                           :onyx.core/serialized-task task
                           :onyx.core/params (resolve-calling-params catalog-entry  opts)
