         (async/close! ch)))
     ch))
 
+(defn- parallel-process-value [env tx ast
+                               key-watchers
+                               res waiting processing
+                               read mutate key-iterations tail]
+  (let [{:keys [query key type params]} ast
+        env   (cond-> (merge env {:ast           ast
+                                  :query         query
+                                  ::waiting      waiting
+                                  ::key-watchers key-watchers})
+                (nil? query) (dissoc :query)
+                (= '... query) (assoc :query tx))
+        value (case type
+                :call
+                (do
+                  (assert mutate "Parse mutation attempted but no :mutate function supplied")
+                  (let [{:keys [action]} (mutate env key params)]
+                    (if action
+                      (try
+                        (trace env {::pt/event ::call-mutation
+                                    :mutation  key})
+                        (action)
+                        (catch #?(:clj Throwable :cljs :default) e
+                          {::error e})))))
+
+                (:prop :join :union)
+                (do
+                  (assert read "Parse read attempted but no :read function supplied")
+                  (tracing env {::pt/event ::call-read :key key}
+                    (read env)))
+
+                nil)]
+    (cond
+      (chan? value)
+      (let [provides #{key}
+            stream   (go
+                       {::provides       provides
+                        ::merge-result?  true
+                        ::response-value {key (<! value)}})]
+        (trace env {::pt/event ::async-return
+                    :key       key})
+        [res
+         (into waiting provides)
+         (conj processing stream)
+         key-iterations
+         tail])
+
+      (::provides value)
+      (let [provides (::provides value)
+            stream   (::response-stream value)]
+        (trace env {::pt/event ::provided-return
+                    ::provides provides})
+        [res
+         (into waiting provides)
+         (conj processing stream)
+         key-iterations
+         tail])
+
+      :else
+      (do
+        (trace env {::pt/event ::value-return
+                    :key       key})
+        [(assoc res key value) waiting processing key-iterations tail]))))
+
+(defn- parallel-flush-watchers [env key-watchers provides]
+  (doseq [[pkey watchers] @key-watchers]
+    (when (contains? provides pkey)
+      (trace env {::pt/event      ::flush-watchers
+                  :key            pkey
+                  ::watcher-count (count watchers)})
+          (doseq [out watchers]
+            (async/put! out {::provides #{pkey}})
+            (async/close! out))
+      (swap! key-watchers dissoc pkey))))
+
 (defn parallel-parser [{:keys [read mutate]}]
   (fn self [{::keys [waiting key-watchers max-key-iterations]
              :or    {max-key-iterations 5}

                            []))

                        (let [next-children (remove (comp (set (keys res)) :key) (:children (query->ast (focus-subquery tx (vec provides)))))]
                          (pt/trace env {::pt/event ::reset-loop
                                         ::loop-keys (mapv :key next-children)})
                          (recur res
                            (into #{} (remove provides) waiting)

                    (recur res waiting processing key-iterations tail))

                  :else
                  (let [[res waiting processing key-iterations tail]
                        (parallel-process-value
                          env tx ast
                          key-watchers
                          res waiting processing
                          read mutate key-iterations tail)]
                    (recur res waiting processing key-iterations tail))))

              (if (seq processing)
                (let [[{::keys [response-value provides merge-result?] :as msg} p] (async/alts! (vec processing))]
                  (if msg

                    (do
                      (trace env {::pt/event       ::process-pending
                                  ::provides       provides
                                  ::response-value response-value
                                  ::merge-result?  (boolean merge-result?)})
                      (swap! (:com.wsscode.pathom.core/entity env) #(merge response-value %))

                      (parallel-flush-watchers env key-watchers provides)

                      (if merge-result?
                        (do
                          (pt/trace env {::pt/event ::merge-result ::response-value response-value})
