                              many? (conj id))
                        xch (chan 1 (delta/nest ks'))]
                    (service/request compscribe-service
-                                    [(get direct-hooks ks) id]
+                                    [(get-in spec-map ks) id]
                                     xch)
                    (vswap! subs assoc ks' xch)
                    ;; xch will be closed by the subscribed-composition
-                   (mix-in target-mix xch)))
-        ch-in (chan)]
-    (service/request source-service [endpoint id] ch-in)
+                   (mix-in target-mix xch)))]
     (go-loop []
       (if-some [delta (<! ch-in)]
-        (let [[{:keys [subs-one subs-many unsubs]} delta]
-              (extract-hooks conformed-spec delta)
-
-              subs-by-pks (group-by-pks @subs)]
-          (doseq [[ks xch]
-                  (->> unsubs
-                       (mapcat (fn [ks]
-                                 (concat (->> (get subs-by-pks ks)
-                                              (map rest))
-                                         (some-> (find @subs ks)
-                                                 (vector)))))
-                       distinct)]
+        (let [[{:keys [subs-one subs-many unsubs]} delta :as dbg]
+              (extract-hooks spec-map delta)]
+          (doseq [[ks xch] (filter (fn [[ks v]]
+                                     (some (fn [uks]
+                                             (= uks (take (count uks) ks)))
+                                           unsubs))
+                                   @subs)]
             (service/cancel compscribe-service xch)
             (mix-out target-mix xch)
             (vswap! subs dissoc ks))

      service/IService
      (request [this spec target]
        (let [[spec id] spec
              [endpoint direct-hooks deep-hooks :as spec] (compile-spec spec)
              subs-target (on-close-source target
                                           #(service/cancel this target))]
          (if (and (empty? direct-hooks)
                   (empty? deep-hooks))
            ;; OPT: If there is nothing to compscribe, directly reach
            ;; through to the source-service:
            (do
              (service/request source-service [endpoint id] subs-target)
              (swap! subscriptions assoc target
                     #(service/cancel source-service subs-target)))
            (do
              (swap! subscriptions assoc target
                     (compscribe* subs-target
                                  source-service
                                  (service/cached this)
                                  spec
                                  id))))))
      (cancel [this target]
        (when-let [unsubs-fn (swap-out! subscriptions target)]
          (unsubs-fn))))))

(defn compscriber
  "Return a service for requesting composite streams of multiple

                                               pipe
                                               #?@(:clj [go go-loop])
                                               tap untap] :as a]
            [clojure.walk :refer [prewalk postwalk]])
  #?(:cljs (:require-macros [cljs.core.async.macros :refer [go-loop go]])))

(defprotocol ClosingMix
  "Implementation detail, subject to change."
