core: jolt-cru — partition-all/partition-by transducer (1-arg) arities

(partition-all n) and (partition-by f) now return stateful transducers, so they
compose with into/sequence/transduce/comp (previously errored "called with 1
argument, expected 2"). Both buffer a window, emit on boundary, flush a partial
trailing window in the completion arity, and honor `reduced` (early stop drops
the in-progress window, no extra trailing partition) — matching Clojure.

partition-all: td-partition-all in core.janet (Janet, alongside the other td-*),
core-partition-all made variadic. partition-by: transducer arity added to the
overlay defn in Clojure with volatiles (stays with its lazy collection arity).

Gate: conformance 253x3 (+4 transducer cases incl comp + reduced), lazy-infinite
44/44, fixpoint, self-host, specs+unit green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

jolt-core/clojure/core/10-seq.clj

@@ -23,14 +23,39 @@
       (recur (conj ret (first s)) (next s))
       (seq ret))))
 
-;; Lazy partition-by: groups consecutive elements by (f x), matching Clojure/CLJS.
-(defn partition-by [f coll]
-  (let [step (fn step [s]
-               (lazy-seq
-                 (let [s (seq s)]
-                   (when s
-                     (let [fst (first s)
-                           fv (f fst)
-                           run (cons fst (take-while (fn [x] (= fv (f x))) (rest s)))]
-                       (cons run (step (lazy-seq (drop (count run) s)))))))))]
-    (step coll)))
+;; partition-by: (partition-by f) is a stateful transducer (buffer a run, emit on
+;; key change, flush on completion — via volatiles, matching Clojure); (partition-by
+;; f coll) is the lazy collection arity.
+(defn partition-by
+  ([f]
+   (fn [rf]
+     (let [buf (volatile! [])
+           pv (volatile! nil)
+           started (volatile! false)]
+       (fn
+         ([] (rf))
+         ([result]
+          (let [b @buf
+                result (if (zero? (count b))
+                         result
+                         (do (vreset! buf []) (unreduced (rf result b))))]
+            (rf result)))
+         ([result input]
+          (let [val (f input)]
+            (if (or (not @started) (= val @pv))
+              (do (vreset! started true) (vreset! pv val) (vswap! buf conj input) result)
+              (let [b @buf]
+                (vreset! buf []) (vreset! pv val)
+                (let [ret (rf result b)]
+                  (when-not (reduced? ret) (vswap! buf conj input))
+                  ret)))))))))
+  ([f coll]
+   (let [step (fn step [s]
+                (lazy-seq
+                  (let [s (seq s)]
+                    (when s
+                      (let [fst (first s)
+                            fv (f fst)
+                            run (cons fst (take-while (fn [x] (= fv (f x))) (rest s)))]
+                        (cons run (step (lazy-seq (drop (count run) s)))))))))]
+     (step coll))))

src/jolt/core.janet

@@ -827,6 +827,30 @@
     (var i -1)
     (fn [& a] (case (length a) 0 (rf) 1 (rf (a 0)) (do (++ i) (rf (a 0) (f i (a 1))))))))
 
+# Stateful windowing transducers. The 1-arg (completion) arity flushes a partial
+# trailing window before delegating to rf's completion; matches Clojure.
+(defn td-partition-all [n]
+  (fn [rf]
+    (var buf @[])
+    (fn [& a]
+      (case (length a)
+        0 (rf)
+        1 (let [result (if (= 0 (length buf)) (a 0)
+                         (let [v (tuple/slice (tuple ;buf))]
+                           (set buf @[])
+                           (core-unreduced (rf (a 0) v))))]
+            (rf result))
+        (do
+          (array/push buf (a 1))
+          (if (= n (length buf))
+            (let [v (tuple/slice (tuple ;buf))]
+              (set buf @[])
+              (rf (a 0) v))
+            (a 0)))))))
+
+# partition-by's transducer arity lives with its (lazy) collection arity in the
+# overlay (10-seq tier), written in Clojure with volatiles.
+
 (defn- reduce-with-reduced
   "Reduce coll with reducing fn rf and seed init, honoring `reduced`. Steps lazy
   seqs one cell at a time so a reducing fn that returns `reduced` (e.g. the
@@ -1361,7 +1385,9 @@
   (if (> (length part) 0) (array/push result (tuple/slice (tuple ;part))))
   result)
 
-(defn core-partition-all [n coll]
+(defn core-partition-all [n & rest]
+ (if (= 0 (length rest)) (td-partition-all n)
+  (let [coll (in rest 0)]
   # Option A: always lazy.
   (defn pstep [c]
     (fn []
@@ -1373,7 +1399,7 @@
             (set cur (core-rest cur))
             (++ i))
           @[(tuple/slice (tuple ;part)) (pstep cur)]))))
-  (make-lazy-seq (pstep (lazy-from coll))))
+  (make-lazy-seq (pstep (lazy-from coll))))))
 
 
 (defn core-keep-indexed [f coll]

test/integration/conformance-test.janet

@@ -303,6 +303,10 @@
    ["transduce remove"  "[1 3 5]" "(into [] (remove even?) [1 2 3 4 5])"]
    ["transduce take-while" "[1 2]" "(into [] (take-while (fn [x] (< x 3))) [1 2 3 4 1])"]
    ["transduce map-indexed" "[[0 :a] [1 :b]]" "(into [] (map-indexed (fn [i x] [i x])) [:a :b])"]
+   ["partition-all xform"   "[[1 2] [3 4] [5]]" "(into [] (partition-all 2) [1 2 3 4 5])"]
+   ["partition-all xform comp" "[2 2 1]" "(into [] (comp (partition-all 2) (map count)) [1 2 3 4 5])"]
+   ["partition-by xform"    "[[1 1] [2 4] [5]]" "(into [] (partition-by odd?) [1 1 2 4 5])"]
+   ["partition-by xform reduced" "[[1 1] [2 4]]" "(into [] (comp (partition-by odd?) (take 2)) [1 1 2 4 5 5])"]
 
    ### ==== regex (capturing groups, backtracking, flags, lookahead) ====
    ["re-find groups"    "[\"12-34\" \"12\" \"34\"]" "(re-find #\"(\\d+)-(\\d+)\" \"x12-34y\")"]