feat(scoring): expose group_by on Tri/Quad/Penta joins

README.md

@@ -170,10 +170,13 @@ ConstraintFactory::<Schedule, HardSoftScore>::new()
 ```
 
 Collectors retain mapped values by ownership. The same collector protocol covers
-unary rows and joined pairs: direct cross-join grouping passes `(&A, &B)` to the
-collector mapper, while both direct cross-join grouped streams and projected
-grouped streams can continue into `complement(...)` for supply/demand-style
-scoring. `collect_vec` exposes a
+unary rows and joined tuples. `Tri`, `Quad`, and `Penta` cross streams and the
+same-source arity streams expose `.group_by(key, collector)`: the key receives
+three, four, or five entity arguments, and the collector mapper receives the
+corresponding borrowed tuple. Grouped scoring weights receive `(&key, &result)`.
+Upstream filters remain binding during full and incremental scoring. Direct
+cross-join grouped pairs and projected grouped streams can continue into
+`complement(...)` for supply/demand-style scoring. `collect_vec` exposes a
 `CollectedVec<T>` result view, so grouped payloads such as `String` labels do
 not need `Copy`, `Clone`, or `PartialEq` just to be collected.
 

crates/solverforge-macros/tests/ui/pass/solverforge_constraints_tri_grouped.rs

@@ -0,0 +1,28 @@
+use solverforge::prelude::*;
+use solverforge::stream::relational::{operator::Pair, Leaf};
+
+struct Model { a: Vec<u32>, b: Vec<u32>, c: Vec<u32> }
+fn a(m: &Model) -> &[u32] { &m.a }
+fn b(m: &Model) -> &[u32] { &m.b }
+fn c(m: &Model) -> &[u32] { &m.c }
+fn pair_key(row: &Pair<Leaf<'_, u32>, Leaf<'_, u32>>) -> u32 { *row.left.entity }
+fn leaf_key(row: &Leaf<'_, u32>) -> u32 { *row.entity }
+
+#[solverforge_constraints]
+fn constraints() -> impl ConstraintSet<Model, SoftScore> {
+    let factory = ConstraintFactory::<Model, SoftScore>::new();
+    (
+        factory.for_each(a as fn(&Model) -> &[u32])
+            .join((b as fn(&Model) -> &[u32], joiner::equal_bi(|a: &u32| *a, |b: &u32| *b)))
+            .join((c as fn(&Model) -> &[u32], joiner::equal_on(pair_key, leaf_key)))
+            .group_by(|a: &u32, _: &u32, _: &u32| *a, count())
+            .penalize(|_: &u32, n: &usize| SoftScore::of(*n as i64))
+            .named("tri grouped count"),
+    )
+}
+fn main() {
+    let mut constraints = constraints();
+    let model = Model { a: vec![1, 1], b: vec![1], c: vec![1] };
+    assert_eq!(constraints.evaluate_all(&model), SoftScore::of(-2));
+    assert_eq!(constraints.initialize_all(&model), SoftScore::of(-2));
+}

crates/solverforge-scoring/WIREFRAME.md

@@ -5,6 +5,26 @@ Zero-erasure incremental constraint scoring infrastructure for SolverForge.
 **Location:** `crates/solverforge-scoring/`
 **Workspace Release:** `0.20.4`
 
+## Joined grouping fluent surface
+
+- `stream::cross::{Tri, Quad, Penta}` and
+  `stream::{TriConstraintStream, QuadConstraintStream, PentaConstraintStream}`
+  expose `group_by(key, collector)`. Key closures take three/four/five entity
+  references; collector mappers receive the corresponding borrowed tuple.
+- `stream::JoinedGroupedStream<S, O, F, C, K, V, R, Acc, Sc>` exposes
+  `penalize(weight)` and `reward(weight)` over `(&K, &R)`.
+  `stream::JoinedGroupedBuilder<..., W, Sc>::named(name)` produces an
+  incremental `OperatorTerminal`. Collector results are borrowed, not cloned;
+  `collect_vec` retains move-only payloads and exposes `CollectedVec<T>`.
+- All these methods preserve upstream filtering and semantic source indexes.
+  Joined tuple grouping uses the existing `GroupNode` and `FilterNode`, not
+  arity-specific retention engines.
+- File map: `stream/group_by_{tri,quad,penta}.rs`,
+  `stream/joined_grouped_stream.rs`, and
+  `stream/relational/tuple_collector.rs`; same-source methods are generated in
+  `stream/arity_stream_macros/nary_stream/shared.rs`. The doc-hidden
+  `stream::relational::TupleCollector<C,V,R,Acc>` is the borrowed-row adapter.
+
 ## Typed relational operator surface
 
 These are module-level low-level exports, not facade-root exports or a completed

crates/solverforge-scoring/src/stream/arity_stream_macros/nary_stream/shared.rs

@@ -167,6 +167,44 @@ macro_rules! impl_nary_arity_stream_common {
                 }
             }
 
+            /// Groups each matching tuple, preserving upstream filters.
+            pub fn group_by<GK, GF, Co, V, R, Acc>(self, key: GF, collector: Co)
+                -> super::JoinedGroupedStream<S,
+                    crate::stream::relational::operator::FilterNode<
+                        crate::stream::relational::operator::SelfJoinNode<S, A,
+                            crate::stream::relational::operator::CollectionNode<S, E>, K, KE, $arity>,
+                        impl Fn(&S, &[crate::stream::relational::Leaf<'_, A>; $arity]) -> bool + Send + Sync>,
+                    impl Fn(&[crate::stream::relational::Leaf<'_, A>; $arity]) -> GK + Send + Sync,
+                    crate::stream::relational::TupleCollector<Co, V, R, Acc>, GK, V, R, Acc, Sc>
+            where
+                E: 'static,
+                KE: 'static,
+                F: 'static,
+                K: 'static,
+                GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static,
+                GF: Fn($(repeat_tokens!($entity => &A)),+) -> GK + Send + Sync + 'static,
+                Co: for<'i> super::collector::Collector<
+                    ($(repeat_tokens!($entity => &'i A)),+), Value = V, Result = R, Accumulator = Acc> + 'static,
+                V: Send + Sync + 'static,
+                R: Send + Sync + 'static,
+                Acc: super::collector::Accumulator<V, R> + 'static,
+                for<'i> crate::stream::relational::TupleCollector<Co, V, R, Acc>:
+                    super::collector::Collector<[crate::stream::relational::Leaf<'i, A>; $arity], Value = V, Result = R, Accumulator = Acc>,
+            {
+                let filter = self.filter;
+                let input = crate::stream::relational::operator::SelfJoinNode::new(
+                    crate::stream::relational::operator::CollectionNode::new(self.extractor, 0), self.key_extractor);
+                let filtered = crate::stream::relational::operator::FilterNode::new(input,
+                    move |s: &S, row: &[crate::stream::relational::Leaf<'_, A>; $arity]| {
+                        filter.test(s, $(row[repeat_nary_arity_position!($entity)].entity,)+
+                            $(row[repeat_nary_arity_position!($filter_idx)].index),+)
+                    });
+                super::JoinedGroupedStream::new(filtered,
+                    move |row: &[crate::stream::relational::Leaf<'_, A>; $arity]| {
+                        key($(row[repeat_nary_arity_position!($entity)].entity),+)
+                    }, crate::stream::relational::TupleCollector::new(collector))
+            }
+
             pub fn penalize<W>(
                 self,
                 weight: W,

crates/solverforge-scoring/src/stream/cross_tri_stream.rs

@@ -7,7 +7,7 @@ domain stays independent. One uniform `.join()` on the Bi stream builds
 it — no per-arity engine, no shared-key retention at any depth.
 */
 
-mod base;
+pub(crate) mod base;
 
 pub(crate) use base::assemble_tri;
 pub use base::Tri;

crates/solverforge-scoring/src/stream/cross_tri_stream/base.rs

@@ -129,8 +129,8 @@ where
     P1: CompileCondition<Plan: IndexedPlan>,
     P2: CompileCondition<Plan: IndexedPlan>,
 {
-    pub(super) tree: BuiltTri<S, EA, EB, EC, P1, P2>,
-    pub(super) filter: F,
+    pub(crate) tree: BuiltTri<S, EA, EB, EC, P1, P2>,
+    pub(crate) filter: F,
     pub(super) _phantom: PhantomData<(fn() -> S, fn() -> A, fn() -> B, fn() -> C, fn() -> Sc)>,
 }
 

crates/solverforge-scoring/src/stream/group_by_penta.rs

@@ -0,0 +1,98 @@
+use super::collector::{Accumulator, Collector};
+use super::filter::PentaFilter;
+use super::joiner::plan::{CompileCondition, IndexedPlan};
+use super::relational::operator::{FilterNode, Operator, Pair};
+use super::relational::{Leaf, TupleCollector};
+use super::JoinedGroupedStream;
+use solverforge_core::score::Score;
+
+type Row<'a, A, B, C, D, E> =
+    Pair<Pair<Pair<Pair<Leaf<'a, A>, Leaf<'a, B>>, Leaf<'a, C>>, Leaf<'a, D>>, Leaf<'a, E>>;
+
+impl<S, A, B, C, D, E, EA, EB, EC, ED, EE, P1, P2, P3, P4, F, Sc>
+    super::cross::Penta<S, A, B, C, D, E, EA, EB, EC, ED, EE, P1, P2, P3, P4, F, Sc>
+where
+    S: Send + Sync + 'static,
+    A: Clone + Send + Sync + 'static,
+    B: Clone + Send + Sync + 'static,
+    C: Clone + Send + Sync + 'static,
+    D: Clone + Send + Sync + 'static,
+    E: Clone + Send + Sync + 'static,
+    EA: 'static,
+    EB: 'static,
+    EC: 'static,
+    ED: 'static,
+    EE: 'static,
+    P1: CompileCondition<Plan: IndexedPlan> + 'static,
+    P2: CompileCondition<Plan: IndexedPlan> + 'static,
+    P3: CompileCondition<Plan: IndexedPlan> + 'static,
+    P4: CompileCondition<Plan: IndexedPlan> + 'static,
+    for<'a> super::cross_penta_stream::BuiltPenta<S, EA, EB, EC, ED, EE, P1, P2, P3, P4>:
+        Operator<S, View<'a> = Row<'a, A, B, C, D, E>> + Send + Sync,
+    F: PentaFilter<S, A, B, C, D, E> + 'static,
+    Sc: Score + 'static,
+{
+    /// Groups joined tuples by a key and an owned-value collector.
+    /// Upstream filters remain binding on full and incremental evaluation.
+    pub fn group_by<GK, GF, Co, V, R, Acc>(
+        self,
+        key: GF,
+        collector: Co,
+    ) -> JoinedGroupedStream<
+        S,
+        FilterNode<
+            super::cross_penta_stream::BuiltPenta<S, EA, EB, EC, ED, EE, P1, P2, P3, P4>,
+            impl Fn(&S, &Row<'_, A, B, C, D, E>) -> bool + Send + Sync,
+        >,
+        impl Fn(&Row<'_, A, B, C, D, E>) -> GK + Send + Sync,
+        TupleCollector<Co, V, R, Acc>,
+        GK,
+        V,
+        R,
+        Acc,
+        Sc,
+    >
+    where
+        GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static,
+        GF: Fn(&A, &B, &C, &D, &E) -> GK + Send + Sync + 'static,
+        Co: for<'i> Collector<
+                (&'i A, &'i B, &'i C, &'i D, &'i E),
+                Value = V,
+                Result = R,
+                Accumulator = Acc,
+            > + 'static,
+        V: Send + Sync + 'static,
+        R: Send + Sync + 'static,
+        Acc: Accumulator<V, R> + 'static,
+    {
+        let filter = self.filter;
+        let input = FilterNode::new(self.tree, move |s: &S, row: &Row<'_, A, B, C, D, E>| {
+            filter.test(
+                s,
+                row.left.left.left.left.entity,
+                row.left.left.left.right.entity,
+                row.left.left.right.entity,
+                row.left.right.entity,
+                row.right.entity,
+                row.left.left.left.left.index,
+                row.left.left.left.right.index,
+                row.left.left.right.index,
+                row.left.right.index,
+                row.right.index,
+            )
+        });
+        JoinedGroupedStream::new(
+            input,
+            move |row: &Row<'_, A, B, C, D, E>| {
+                key(
+                    row.left.left.left.left.entity,
+                    row.left.left.left.right.entity,
+                    row.left.left.right.entity,
+                    row.left.right.entity,
+                    row.right.entity,
+                )
+            },
+            TupleCollector::new(collector),
+        )
+    }
+}

crates/solverforge-scoring/src/stream/group_by_quad.rs

@@ -0,0 +1,91 @@
+use super::collector::{Accumulator, Collector};
+use super::filter::QuadFilter;
+use super::joiner::plan::{CompileCondition, IndexedPlan};
+use super::relational::operator::{FilterNode, Operator, Pair};
+use super::relational::{Leaf, TupleCollector};
+use super::JoinedGroupedStream;
+use solverforge_core::score::Score;
+
+type Row<'a, A, B, C, D> = Pair<Pair<Pair<Leaf<'a, A>, Leaf<'a, B>>, Leaf<'a, C>>, Leaf<'a, D>>;
+
+impl<S, A, B, C, D, EA, EB, EC, ED, P1, P2, P3, F, Sc>
+    super::cross::Quad<S, A, B, C, D, EA, EB, EC, ED, P1, P2, P3, F, Sc>
+where
+    S: Send + Sync + 'static,
+    A: Clone + Send + Sync + 'static,
+    B: Clone + Send + Sync + 'static,
+    C: Clone + Send + Sync + 'static,
+    D: Clone + Send + Sync + 'static,
+    EA: 'static,
+    EB: 'static,
+    EC: 'static,
+    ED: 'static,
+    P1: CompileCondition<Plan: IndexedPlan> + 'static,
+    P2: CompileCondition<Plan: IndexedPlan> + 'static,
+    P3: CompileCondition<Plan: IndexedPlan> + 'static,
+    for<'a> super::cross_quad_stream::BuiltQuad<S, EA, EB, EC, ED, P1, P2, P3>:
+        Operator<S, View<'a> = Row<'a, A, B, C, D>> + Send + Sync,
+    F: QuadFilter<S, A, B, C, D> + 'static,
+    Sc: Score + 'static,
+{
+    /// Groups joined tuples by a key and an owned-value collector.
+    /// Upstream filters remain binding on full and incremental evaluation.
+    pub fn group_by<GK, GF, Co, V, R, Acc>(
+        self,
+        key: GF,
+        collector: Co,
+    ) -> JoinedGroupedStream<
+        S,
+        FilterNode<
+            super::cross_quad_stream::BuiltQuad<S, EA, EB, EC, ED, P1, P2, P3>,
+            impl Fn(&S, &Row<'_, A, B, C, D>) -> bool + Send + Sync,
+        >,
+        impl Fn(&Row<'_, A, B, C, D>) -> GK + Send + Sync,
+        TupleCollector<Co, V, R, Acc>,
+        GK,
+        V,
+        R,
+        Acc,
+        Sc,
+    >
+    where
+        GK: Eq + std::hash::Hash + Clone + Send + Sync + 'static,
+        GF: Fn(&A, &B, &C, &D) -> GK + Send + Sync + 'static,
+        Co: for<'i> Collector<
+                (&'i A, &'i B, &'i C, &'i D),
+                Value = V,
+                Result = R,
+                Accumulator = Acc,
+            > + 'static,
+        V: Send + Sync + 'static,
+        R: Send + Sync + 'static,
+        Acc: Accumulator<V, R> + 'static,
+    {
+        let filter = self.filter;
+        let input = FilterNode::new(self.tree, move |s: &S, row: &Row<'_, A, B, C, D>| {
+            filter.test(
+                s,
+                row.left.left.left.entity,
+                row.left.left.right.entity,
+                row.left.right.entity,
+                row.right.entity,
+                row.left.left.left.index,
+                row.left.left.right.index,
+                row.left.right.index,
+                row.right.index,
+            )
+        });
+        JoinedGroupedStream::new(
+            input,
+            move |row: &Row<'_, A, B, C, D>| {
+                key(
+                    row.left.left.left.entity,
+                    row.left.left.right.entity,
+                    row.left.right.entity,
+                    row.right.entity,
+                )
+            },
+            TupleCollector::new(collector),
+        )
+    }
+}