   (merge
    event
    (when (and children (not (:onyx/side-effects-only? (:onyx.core/task-map event))))
-     (doseq [raw-segment (keys children)]
+     (doseq [[raw-segment tagged] children]
        (when (:ack-val raw-segment)
          (let [link (operation/peer-link event (:acker-id raw-segment) :acker-peer-site)]
            (extensions/internal-ack-message

            results (map (partial build-next-segment thawed) segments)
            tagged (acker/prefuse-vals (map :ack-val results))]
        (-> rets
            (update-in [:onyx.core/results] concat results)
            (assoc-in [:onyx.core/children thawed] tagged))))
    {:onyx.core/results []
     :onyx.core/children {}}
    decompressed)))

(defn apply-fn-batch [{:keys [onyx.core/batch onyx.core/decompressed] :as event}]

            link
            (:id raw-segment)
            (:completion-id raw-segment)
            (fuse-ack-vals (:onyx.core/task-map event) (:ack-val raw-segment) tagged))))))))

(defn join-output-paths [all to-add downstream]
  (cond (= to-add :all) (into #{} downstream)
