                          :onyx.core/params (or (get (:fn-params opts) task) [])
                          :onyx.core/queue queue
                          :onyx.core/sync sync
-                         :onyx.core/blocked? (atom {:try-seal? false :seal? false :complete? false})}
+                         :onyx.core/pipeline-state (atom {:try-seal? false :seal? false :complete? false})}
 
           pipeline-data (assoc pipeline-data :onyx.core/queue (extensions/optimize-concurrently queue pipeline-data))
           pipeline-data (merge pipeline-data (l-ext/inject-lifecycle-resources* pipeline-data))]


(defn munge-seal-resource
  [{:keys [onyx.core/sync onyx.core/exhaust-node onyx.core/seal-node
           onyx.core/blocked?] :as event}]
  (if (:try-seal? @blocked?)
    (merge event {:onyx.core/sealed? false})
    (let [seal-response-ch (chan)]
      (extensions/on-change sync seal-node #(>!! seal-response-ch %))

          (let [path (:path response)
                result (extensions/read-node sync path)]
            (if (:completed? result)
              (do (swap! pipeline-state assoc :complete? true)
                  (merge event {:onyx.core/complete-success? true}))
              (merge event {:onyx.core/complete-success? true}))))))))

