                                   (merge base-function-task partial-clojure-fn-task)))))
 
 (def TaskMap
-  (s/conditional #(= (:onyx/type %) :input) 
+  (s/conditional #(= (:onyx/type %) :input)
                  InputTaskSchema
                  #(= (:onyx/type %) :output)
                  OutputTaskSchema

   (s/optional-key :aggregation/super-aggregation-fn) Function})

(def WindowExtension
  (s/both Record
          {:window Window
           :id s/Keyword
           :task TaskName
           :type WindowType
           :aggregation (s/either s/Keyword [s/Keyword])
           (s/optional-key :init) (s/maybe s/Any)
           (s/optional-key :window-key) (s/maybe s/Any)
           (s/optional-key :min-value) (s/maybe SPosInt)
           (s/optional-key :range) (s/maybe Unit)
           (s/optional-key :slide) (s/maybe Unit)
           (s/optional-key :timeout-gap) (s/maybe Unit)
           (s/optional-key :session-key) (s/maybe s/Any)
           (s/optional-key :doc) (s/maybe s/Str)}))

(def TriggerRefinement
  NamespacedKeyword)

(def TriggerPeriod 
  (s/enum :milliseconds :seconds :minutes :hours :days
          :mullisecond :second :minute :hour :day))


   (s/optional-key :onyx/required-tags) [s/Keyword]
   UnsupportedTaskMapKey s/Any})

(def FluxPolicy
  (s/enum :continue :kill :recover))

(def FnPath
  (s/cond-pre NamespacedKeyword s/Keyword))

(def partial-grouping-task
  {(s/optional-key :onyx/group-by-key) s/Any
   (s/optional-key :onyx/group-by-fn) FnPath
   :onyx/flux-policy FluxPolicy})

(defn grouping-task? [task-map]
  (and (#{:function :output} (:onyx/type task-map))
       (or (not (nil? (:onyx/group-by-key task-map)))
           (not (nil? (:onyx/group-by-fn task-map))))))

(def partial-input-task
  {:onyx/plugin (s/cond-pre NamespacedKeyword s/Keyword)
   :onyx/medium s/Keyword
   :onyx/type (s/enum :input)
   (s/optional-key :onyx/fn) FnPath
   (s/optional-key :onyx/input-retry-timeout) PosInt
   (s/optional-key :onyx/pending-timeout) PosInt
   (s/optional-key :onyx/max-pending) PosInt})

(def partial-output-task
  {:onyx/plugin (s/cond-pre NamespacedKeyword s/Keyword)
   :onyx/medium s/Keyword
   :onyx/type (s/enum :output)
   (s/optional-key :onyx/fn) FnPath})

(def NonNamespacedKeyword
  (s/pred (fn [v]
            (and (keyword? v)
                 (not (namespace v))))
          'keyword-non-namespaced))

(def partial-java-plugin
  {:onyx/plugin NonNamespacedKeyword
   (s/optional-key :onyx/fn) FnPath})

(def partial-clojure-plugin
  {:onyx/plugin NamespacedKeyword
   (s/optional-key :onyx/fn) FnPath})

(def partial-fn-task
  {:onyx/fn (s/cond-pre NamespacedKeyword s/Keyword)
   (s/optional-key :onyx/plugin) (s/cond-pre NamespacedKeyword s/Keyword)})

(def partial-clojure-fn-task
  {:onyx/fn NamespacedKeyword})

(def partial-java-fn-task
  {:onyx/fn s/Keyword})

(defn java? [task-map]
  (= :java (:onyx/language task-map)))

(defn valid-min-peers-max-peers-n-peers? [entry]
  (case (:onyx/flux-policy entry)
    :continue
    (or (:onyx/n-peers entry)
        (:onyx/min-peers entry)
        (= (:onyx/max-peers entry) 1))
    :kill
    (or (:onyx/n-peers entry)
        (:onyx/min-peers entry)
        (= (:onyx/max-peers entry) 2))
    :recover
    (or (:onyx/n-peers entry)
        (and (:onyx/max-peers entry)
             (= (:onyx/max-peers entry)
                (:onyx/min-peers entry)))
        (= (:onyx/max-peers entry) 1))))

(def OutputTaskSchema
  (let [base-output-schema (merge base-task-map partial-output-task)
        base-output-grouping (merge base-output-schema partial-grouping-task)
        java-output-grouping (merge base-output-grouping partial-java-plugin)
        clojure-output-grouping (merge base-output-grouping partial-clojure-plugin)]
    (s/conditional grouping-task?
                   (s/conditional java?
                                  (s/constrained java-output-grouping
                                                 valid-min-peers-max-peers-n-peers?
                                                 'valid-flux-policy-min-max-n-peers)
                                  :else
                                  (s/constrained clojure-output-grouping
                                                 valid-min-peers-max-peers-n-peers?
                                                 'valid-flux-policy-min-max-n-peers))
                   :else
                   (s/conditional java?
                                  (merge base-output-schema partial-java-plugin)
                                  :else
                                  (merge base-output-schema partial-clojure-plugin)))))

(def InputTaskSchema
  (let [base-input-schema (merge base-task-map partial-input-task)]
    (s/conditional java?
                   (merge base-input-schema partial-java-plugin)
                   :else
                   base-input-schema)))

(def FunctionTaskSchema
  (let [base-function-task (merge base-task-map partial-fn-task)
        java-function-task (merge base-function-task partial-grouping-task partial-java-fn-task)
        clojure-function-task (merge base-function-task partial-grouping-task partial-clojure-fn-task)]
    (s/conditional grouping-task?
                   (s/conditional java?
                                  (s/constrained java-function-task
                                                 valid-min-peers-max-peers-n-peers?
                                                 'valid-flux-policy-min-max-n-peers)
                                  :else
                                  (s/constrained clojure-function-task
                                                 valid-min-peers-max-peers-n-peers?
                                                 'valid-flux-policy-min-max-n-peers))
                   :else
                   (s/conditional java?
                                  (merge base-function-task partial-java-fn-task)
