fix(scoring)!: enforce Projection::MAX_EMITS

crates/solverforge-scoring/WIREFRAME.md

@@ -754,7 +754,7 @@ Dynamic closure weights are non-hard metadata by default, even when their score
 
 **`UniConstraintBuilder<S, A, E, F, W, Sc>`** — `named()` → `IncrementalUniConstraint`
 
-**`Projection<A>`** — Retained projection contract for single-source `.project(...)`. Implementations define `type Out`, `const MAX_EMITS: usize`, and `project(&self, input: &A, sink: &mut impl ProjectionSink<Self::Out>)`. Projection implementations emit bounded scoring rows into the sink; Vec-returning closures are not part of the API. `Out` does not need `Clone`.
+**`Projection<A>`** — Retained projection contract for single-source `.project(...)`. Implementations define `type Out`, `const MAX_EMITS: usize`, and `project(&self, input: &A, sink: &mut impl ProjectionSink<Self::Out>)`. Projection implementations emit bounded scoring rows into the sink: emitting more than `MAX_EMITS` rows for one input panics, naming the projection and both counts, and retained state sizes each entity's row list by `MAX_EMITS`; Vec-returning closures are not part of the API. `Out` does not need `Clone`.
 
 The fluent projected stream keeps its concrete source type in the return type:
 `SingleSource` for one projection, `FilteredSource` after a projected filter,

crates/solverforge-scoring/src/constraint/tests/projected.rs

@@ -1,3 +1,4 @@
+mod emit_bound;
 mod joined;
 mod localization;
 mod self_join;

crates/solverforge-scoring/src/constraint/tests/projected/emit_bound.rs

@@ -0,0 +1,75 @@
+use super::support::*;
+
+/// Emits one entry per unit of demand while declaring at most two.
+struct EntryPerDemand;
+
+impl Projection<Work> for EntryPerDemand {
+    type Out = Entry;
+    const MAX_EMITS: usize = 2;
+
+    fn project<Sink>(&self, work: &Work, out: &mut Sink)
+    where
+        Sink: ProjectionSink<Self::Out>,
+    {
+        for bucket in 0..work.demand as usize {
+            out.emit(Entry { bucket, delta: 1 });
+        }
+    }
+}
+
+fn entries_per_demand() -> impl IncrementalConstraint<Plan, SoftScore> {
+    ConstraintFactory::<Plan, SoftScore>::new()
+        .for_each(source(
+            work as fn(&Plan) -> &[Work],
+            ChangeSource::Descriptor(0),
+        ))
+        .project(EntryPerDemand)
+        .group_by(
+            |entry: &Entry| entry.bucket,
+            sum(|entry: &Entry| entry.delta),
+        )
+        .penalize(|_bucket: &usize, delta: &i64| SoftScore::of(*delta))
+        .named("entries")
+}
+
+fn plan(demand: i64) -> Plan {
+    Plan {
+        work: vec![Work {
+            bucket: 0,
+            demand,
+            enabled: true,
+        }],
+        capacity: Vec::new(),
+    }
+}
+
+#[test]
+fn a_projection_may_emit_up_to_its_bound() {
+    let mut constraint = entries_per_demand();
+    let mut plan = plan(1);
+    let mut total = constraint.initialize(&plan);
+    assert_eq!(total, SoftScore::of(-1));
+
+    total = total + constraint.on_retract(&plan, 0, 0);
+    plan.work[0].demand = 2;
+    total = total + constraint.on_insert(&plan, 0, 0);
+    assert_eq!(total, SoftScore::of(-2));
+    assert_eq!(total, constraint.evaluate(&plan));
+}
+
+#[test]
+#[should_panic(expected = "EntryPerDemand` emitted 3 rows for one input; its MAX_EMITS is 2")]
+fn emitting_past_the_bound_panics_on_initialization() {
+    entries_per_demand().initialize(&plan(3));
+}
+
+#[test]
+#[should_panic(expected = "EntryPerDemand` emitted 3 rows for one input; its MAX_EMITS is 2")]
+fn emitting_past_the_bound_panics_on_an_incremental_insert() {
+    let mut constraint = entries_per_demand();
+    let mut plan = plan(2);
+    constraint.initialize(&plan);
+    constraint.on_retract(&plan, 0, 0);
+    plan.work[0].demand = 3;
+    constraint.on_insert(&plan, 0, 0);
+}

crates/solverforge-scoring/src/stream/projected_stream/source.rs

@@ -25,8 +25,10 @@ pub trait ProjectionSink<Out> {
 pub trait Projection<A>: Send + Sync {
     /// The emitted row type; it need not be `Clone`.
     type Out: Send + Sync + 'static;
-    /// Declared maximum number of rows one `project` call emits. Sources combine
-    /// it (a merge adds both bounds); nothing reads it at runtime.
+    /// Maximum number of rows one `project` call emits. Emitting more panics,
+    /// naming the projection and both counts. Retained scoring state sizes
+    /// each entity's row list by it, and sources combine it (a merge adds both
+    /// bounds).
     const MAX_EMITS: usize;
 
     /// Emits the rows for `input` into `sink`.
@@ -35,20 +37,57 @@ pub trait Projection<A>: Send + Sync {
         Sink: ProjectionSink<Self::Out>;
 }
 
-struct VisitSink<V> {
+/// Runs `projection` on `input`, passing each row with its emit index to
+/// `visit`.
+///
+/// # Panics
+///
+/// When the projection emits more than `P::MAX_EMITS` rows.
+fn project_bounded<A, P, V>(projection: &P, input: &A, visit: V)
+where
+    P: Projection<A>,
+    V: FnMut(usize, P::Out),
+{
+    let mut sink = BoundedSink {
+        visit,
+        emitted: 0,
+        max_emits: P::MAX_EMITS,
+        projection: std::any::type_name::<P>,
+    };
+    projection.project(input, &mut sink);
+}
+
+struct BoundedSink<V> {
     visit: V,
+    emitted: usize,
+    max_emits: usize,
+    projection: fn() -> &'static str,
 }
 
-impl<Out, V> ProjectionSink<Out> for VisitSink<V>
+impl<Out, V> ProjectionSink<Out> for BoundedSink<V>
 where
-    V: FnMut(Out),
+    V: FnMut(usize, Out),
 {
     #[inline]
     fn emit(&mut self, output: Out) {
-        (self.visit)(output);
+        if self.emitted == self.max_emits {
+            emitted_past_bound((self.projection)(), self.max_emits);
+        }
+        let emit_index = self.emitted;
+        self.emitted += 1;
+        (self.visit)(emit_index, output);
     }
 }
 
+#[cold]
+#[inline(never)]
+fn emitted_past_bound(projection: &str, max_emits: usize) -> ! {
+    panic!(
+        "projection `{projection}` emitted {} rows for one input; its MAX_EMITS is {max_emits}",
+        max_emits + 1
+    );
+}
+
 #[doc(hidden)]
 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
 pub struct RowOwner {

crates/solverforge-scoring/src/stream/projected_stream/source/operator.rs

@@ -66,7 +66,12 @@ where
         }
         let handle = self.rows.insert(StoredRow { coordinate, value });
         self.by_coordinate.insert(coordinate, handle);
-        coordinate.for_each_owner(|owner| self.by_owner.entry(owner).or_default().push(handle));
+        coordinate.for_each_owner(|owner| {
+            self.by_owner
+                .entry(owner)
+                .or_insert_with(|| Vec::with_capacity(Src::MAX_EMITS))
+                .push(handle);
+        });
         Some(handle)
     }
 

crates/solverforge-scoring/src/stream/projected_stream/source/single.rs

@@ -3,7 +3,7 @@ use std::marker::PhantomData;
 use crate::stream::collection_extract::{ChangeSource, CollectionExtract};
 use crate::stream::filter::UniFilter;
 
-use super::{Projection, RowCoordinate, Source, VisitSink};
+use super::{project_bounded, Projection, RowCoordinate, Source};
 
 pub struct SingleSource<S, A, E, F, P, Out> {
     extractor: E,
@@ -58,15 +58,9 @@ where
             if !self.filter.test(solution, entity) {
                 continue;
             }
-            let mut emit_index = 0;
-            let mut sink = VisitSink {
-                visit: |output| {
-                    let coordinate = RowCoordinate::single(0, idx, emit_index);
-                    emit_index += 1;
-                    visit(coordinate, output);
-                },
-            };
-            self.projection.project(entity, &mut sink);
+            project_bounded(&self.projection, entity, |emit_index, output| {
+                visit(RowCoordinate::single(0, idx, emit_index), output);
+            });
         }
     }
 
@@ -90,15 +84,9 @@ where
         if !self.filter.test(solution, entity) {
             return;
         }
-        let mut emit_index = 0;
-        let mut sink = VisitSink {
-            visit: |output| {
-                let coordinate = RowCoordinate::single(0, entity_index, emit_index);
-                emit_index += 1;
-                visit(coordinate, output);
-            },
-        };
-        self.projection.project(entity, &mut sink);
+        project_bounded(&self.projection, entity, |emit_index, output| {
+            visit(RowCoordinate::single(0, entity_index, emit_index), output);
+        });
     }
 
     fn insert_entity_state(

crates/solverforge/WIREFRAME.md

@@ -373,7 +373,7 @@ ConstraintFactory::<Plan, HardSoftScore>::new()
 
 Generated existence ergonomics: `#[planning_solution]` generates inherent source methods such as `Plan::assignments()` with hidden descriptor/static metadata, plus the matching `PlanConstraintStreams` convenience trait. Localized incremental callbacks use entity indexes only for the owning planning-entity collection. Raw facade `for_each(...)` extractors do not carry localized source ownership. Flattened existence targets use `.flattened(...)` and `FlattenedCollectionTarget`.
 
-Projected scoring ergonomics: `ConstraintFactory::new().for_each(Plan::assignments()).project(entries)`, where `entries` is a value of a named `Projection<A>` type, creates bounded scoring rows from a named `Projection<A>` type without materializing facts or entities. Keyed cross joins can group joined pairs directly with `.group_by(|left, right| key, collector)` and complement the grouped result, or project them with `.project(|assignment, capacity| Row { ... })` to emit one scoring row per retained joined pair. Projected streams can be filtered, self-joined, merged, grouped, complemented after grouping, and weighted as retained scoring state. Single-source projection implementations emit through `ProjectionSink` and declare `MAX_EMITS`; joined-pair closures do not need a helper type. Projected output rows, projected self-join keys, and grouped collector values do not need `Clone`. Projected self-join ordering is coordinate-stable by `RowCoordinate`; directed projected self-joins use `.join(joiner::equal_bi(left_key, right_key))`; low-level pair-filter indexes are primary owner entity indexes, not sparse storage row IDs.
+Projected scoring ergonomics: `ConstraintFactory::new().for_each(Plan::assignments()).project(entries)`, where `entries` is a value of a named `Projection<A>` type, creates bounded scoring rows from a named `Projection<A>` type without materializing facts or entities. Keyed cross joins can group joined pairs directly with `.group_by(|left, right| key, collector)` and complement the grouped result, or project them with `.project(|assignment, capacity| Row { ... })` to emit one scoring row per retained joined pair. Projected streams can be filtered, self-joined, merged, grouped, complemented after grouping, and weighted as retained scoring state. Single-source projection implementations emit through `ProjectionSink` and declare `MAX_EMITS`, the most rows one input may emit (emitting more panics); joined-pair closures do not need a helper type. Projected output rows, projected self-join keys, and grouped collector values do not need `Clone`. Projected self-join ordering is coordinate-stable by `RowCoordinate`; directed projected self-joins use `.join(joiner::equal_bi(left_key, right_key))`; low-level pair-filter indexes are primary owner entity indexes, not sparse storage row IDs.
 
 Constraint authoring: `#[solverforge_constraints]` is the canonical attribute
 for constraint factory functions. It preserves fluent stream syntax and gives