I was wondering if there might be a simpler or more elegant way of doing this... every ms milliseconds, batch produces a vector of all of the values that the input flow produced during the time interval, if the input flow produced any values during that time. For example, if the interval is set to one minute and the input flow produces three items in the first minute, after one minute batch will produce a vector of those three items. After two minutes, batch will produce a vector of any items that came in during the second minute, if any items were produced; and so on. If the input flow terminates, batch immediately produces a vector of the remaining items (if there are any remaining items) without waiting for the end of the time interval, and terminates. I struggled a bit to handle termination. I ended up using a dfv to signal when the input flow had terminated:

(defn batch [ms >f]
  (let [terminated (m/dfv)
        !accum (atom [])]
    (letfn [(add-item [x] (swap! !accum conj x))
            (take-items [] (first (swap-vals! !accum (constantly []))))]
      (m/ap
       (m/amb=

        ;; Add items from the input flow to `!accum` as they come
        ;; in; assign `terminated` when the input flow terminates.
        (m/amb
         (do (add-item (m/?> >f))
             (m/amb))
         (do (terminated nil)
             (m/amb)))

        ;; Process until `terminated` is assigned.
        (if (m/?< (m/ap (m/amb true (do (m/? terminated) false))))
          ;; Not yet terminated; produce items every `ms`
          ;; milliseconds.  Will be canceled if terminated is assigned
          ;; during the sleep.
          (try
            (loop []
              (m/? (m/sleep ms))
              (let [items (take-items)]
                (if (= items [])
                  (recur)
                  (m/amb items (recur)))))
            (catch Cancelled _ (m/amb)))

          ;; The input flow has terminated; produce the final batch of items.
          (let [items (take-items)]
            (if (= items [])
              (m/amb)
              (m/amb items)))))))))