 
 (defn source
   "Return a fresh `:source`-stage job graph node consuming from the
-provided `dseqs`."
-  [& dseqs]
-  {:stage :source, :source-id (gen-id), :source (vec dseqs), :config []})
+provided `dseq`."
+  [dseq]
+  {:stage :source,
+   :source-id (gen-id),
+   :source (dseq/dseq dseq),
+   :config [],
+   })
 
 (defn config
   "Add arbitrary configuration steps to current job graph `node`."

  "Configure a job according to the state of a job graph node."
  [^Job job node]
  (doto job
    (source-config (:source node))
    (subnodes-config node (:subnodes node))
    (pgt/mapper! node)
    (pgt/combiner! node)


(defmethod job-fn :source
  [node conf jname]
  (constantly (:source node)))

(defmethod job-fn :default
  [node conf jname]
