                    :group-id g
                    :args [:add-peer p]}]))))
 
+(defn reset-peer-path [segment]
+  (assoc segment :peer-path []))
+
+(defn normalize-state [state]
+  (set (map (juxt key 
+                  (comp set 
+                        (partial map reset-peer-path) 
+                        val)) 
+            state)))
+
 (defn job-state-properties [expected-state job-id state]
   ;(println "EXPECTED" expected-state)
   ;(println "STATE" state)
   ;let [state-values (reduce into [] state)]
   ;(prop-is (= (count (set state)) (count state-values)) "not enough state values")
-  (let [v1 (set (map (juxt key (comp set val)) 
-                        expected-state))
-        v2 (set (map (juxt key (comp set val))
-                        state))]
+  (let [v1 (normalize-state expected-state)
+        v2 (normalize-state state)]
     (prop-is (= v1 v2) 
              "job-state-properties: trigger state is incorrect"
              (vec (take 2 (clojure.data/diff v1 v2))))))

;;;;;;;;;
;; Job code

(defn add-path [job-id task+slot {:keys [n] :as segment}]
  (update segment :path conj [job-id task+slot]))

(def add-paths-calls
  {:lifecycle/before-task-start 
   (fn [event lifecycle]
     (if (= :input (:onyx/type (:task-map event))) 
       ;; include slot combinations with input
       {:params [(:job-id event) [(:task event) (:slot-id event)]]}
       {:params [(:job-id event) [(:task event)]]}))})

(defn add-paths-lifecycles [job]
  (update job 

  (mapcat (fn [path] 
            (map (fn [seg] 
                   (reduce (fn [segment task] 
                             (add-path [] job-id task segment)) 
                           seg
                           path)) 
                 inputs))
