 
 (def remote nil)
 (defmulti remote
-  "Create a new remote-execution task chained following the provided
-job graph `node` and implemented by function `f`, which should accept
-and return reducible collections of task tuples.  When producing a
-`:map`-stage node, may provide a combiner function `g`, will replace
-any existing job combiner function."
+  "Create a new remote-execution task chained following the provided job graph
+`node` and implemented by function `f`, which should accept and return reducible
+collections of task tuples.  When producing a `:map`-stage node, may provide a
+combiner function `g`, which will replace any existing job combiner function."
   {:arglists '([node f] [node f g])}
   stage)
 
 (defmethod remote :default
-  [node f]
+  [node & fs]
   (let [stage (:stage node)
-        msg (str "Cannot change remote-execution task from node of stage `"
-                 stage "`.")]
-    (throw (ex-info msg {:node node, :f f}))))
+        msg (str "Cannot chain remote-execution from stage `" stage "`.")]
+    (throw (ex-info msg {:node node}))))
+
+(defmethod remote ::vector
+  [nodes & fs]
+  (if-not (every? (comp #{:source} stage) nodes)
+    (throw (ex-info "Cannot merge non-`:source` nodes." {:nodes nodes}))
+    (assoc (source (mapv :source nodes))
+      :requires (into [] (mapcat :requires nodes)))))
 
 (defmethod remote :source
   ([node f] (assoc node :stage :map, :mapper (f-list f)))

  ([node f] (assoc node :mapper (cons f (:mapper node))))
  ([node f g] (assoc (remote node f) :combiner (f-list g))))

(def partition nil)
(defmulti partition
  "Create new reduce-partition task chained following the provided job
graph `node` and optionally implemented by function `f`, which should
follow the interface described in `parkour.mapreduce/partition!`.
Configures shuffle via `classes`, which should either be a vector of
the two map-output key & value classes, or a config step specifying
the shuffle."
  {:arglists '([node classes] [node classes f])}
  (comp type pr/arg0))

(defn shuffle
  "Base shuffle configuration; sets map output key & value types to


(def ^:private stage
  "The stage of a job graph node."
  (fn [node & args]
    (cond
     (vector? node) ::vector
     (map? node) (:stage node)
     :else (throw (ex-info "Invalid `node`" {:node node})))))

(let [id (atom 0)]
  (defn ^:private gen-id
