       ((:window/agg-init w) w)))
 
 (defn replay-windows-from-log
-  [{:keys [onyx.core/windows onyx.core/window-state onyx.core/state-log] :as event}]
-  (when (seq windows)
+  [{:keys [onyx.core/window-state onyx.core/state-log] :as event}]
+  (when (windowed-task? event)
     (swap! window-state 
            (fn [wstate] 
              (let [replayed-state (state-extensions/playback-log-entries state-log event wstate)]


(defn resolve-window-state [{:keys [onyx.core/peer-opts] :as pipeline}]
  (let [filter-impl (arg-or-default :onyx.peer/state-filter-impl peer-opts)] 
    (assoc pipeline :onyx.core/window-state (if (exactly-once-task? pipeline) 
                                              (do
                                                (info "Initialising window state " )
                                               
                                                (atom (->WindowState (state-extensions/initialize-filter filter-impl pipeline) 
                                                                   {})))))))

(defrecord TaskLifeCycle
    [id log messenger-buffer messenger job-id task-id replica peer-replica-view restart-ch


(defn add-completion-id [id m]
  (assoc m :completion-id id))

(defn windowed-task? [event]
  (or (not-empty (:onyx.core/windows event))
      (not-empty (:onyx.core/triggers event))))

(defn sentinel-found? [event]
  (seq (filter #(= :done (:message %))
