     (async/pipe step-result-updates-ch only-status-updates)
     only-status-updates))
 
+(defn- clear-channel [ch]
+  (loop []
+    (let [[v _] (async/alts!! [ch] :default :finished)]
+      (when (not= :finished v)
+        (recur)))))
+
 (defn- init-state []
   (let [is-killed (atom false)
         ctx (some-ctx-with :is-killed is-killed)

    (swap! state #(assoc % :step-future future))
    state))

(defn start-wait-for-git-step [state & {:keys [ref ms-between-polls] :or {ms-between-polls 100
                                                                          ref              "refs/heads/master"}}]
  (execute-pipeline-step state (fn [_ ctx]
                                 (wait-for-git ctx (get-in @state [:git :remote]) :ref ref :ms-between-polls ms-between-polls))))


(defn start-clone-step [state ref cwd]


(defn wait-for-step-waiting [state]
  (let [step-status-ch (:step-status-channel @state)
        result         (read-channel-or-time-out
                         (async/go
                           (loop []
                             (let [status (async/<! step-status-ch)]
                               (if (not= :waiting status)
                                 (recur)
                                 nil)))))]
    (when (instance? Exception result)
      (throw result)))
  state)

(defn execute-pipeline-step [state step]
  (clear-channel (:step-status-channel @state))
  (let [future (future
                 (let [execute-step-result (lambdacd-core/execute-step {} (:ctx @state)
                                                                       step)]
