feat: parallel local-search evaluation for Python models

src/constraints/compiled_dispatch_tests.rs

@@ -94,7 +94,12 @@ fn compiled_precedence_never_decodes_generic_rows_or_plan_fields() {
         let baseline = set.initialize_all(&solution);
         assert_eq!(baseline, DynamicScore::hard_soft(0, -5));
         assert!(set.cached_rows.is_none());
-        assert_eq!(set.on_retract_all(&solution, 0, 0), DynamicScore::zero());
+        // The declared family holds even though this thread never entered a
+        // solve scope.
+        assert_eq!(
+            set.on_retract_all(&solution, 0, 0),
+            DynamicScore::zero_for_family(ScoreFamily::HardSoft)
+        );
         // A cached delta must not parse even precedence-specific immutable fields.
         let plan = set.compiled.plans()[0]
             .python_plan()

src/constraints/state.rs

@@ -15,7 +15,10 @@ use super::compiled::{
 use super::evaluate;
 use super::list_precedence::{self, ListPrecedenceDeltaMode, ListPrecedenceStateCache};
 use crate::error::{panic_with_py_err, py_err};
-use crate::score::{dynamic_score_from_native, DynamicScore};
+use crate::score::{
+    dynamic_score_from_native, scoped_dynamic_score_family, score_family_from_name, DynamicScore,
+    ScoreFamily,
+};
 use crate::state::PyDynamicSolution;
 use crate::value::DynamicValue;
 
@@ -333,14 +336,19 @@ pub struct PyDynamicConstraintSet {
     list_unassigned_count_cache: ListUnassignedCountCache,
     list_precedence_cache: ListPrecedenceStateCache,
     retracted_entities: RetractedEntitySet,
+    /// Declared score family. Every `ConstraintSet` entry point enters it, so
+    /// scores match on any thread, including Rayon workers scoring evaluator
+    /// replicas that never entered the solve thread's scope.
+    family: ScoreFamily,
 }
 
 unsafe impl Send for PyDynamicConstraintSet {}
 unsafe impl Sync for PyDynamicConstraintSet {}
 
 impl PyDynamicConstraintSet {
-    pub fn new(constraints: Py<PyAny>) -> Self {
+    pub fn new(constraints: Py<PyAny>, family: ScoreFamily) -> Self {
         Self {
+            family,
             compiled: CompiledConstraintSet::python(constraints),
             cached_rows: None,
             join_key_cache: JoinKeyCache::new(),
@@ -354,8 +362,13 @@ impl PyDynamicConstraintSet {
     }
 
     pub fn from_solution(py: Python<'_>, solution: &PyDynamicSolution) -> PyResult<Self> {
+        let family = score_family_from_name(&solution.schema().score_family)?;
+        let compiled = scoped_dynamic_score_family(family, || {
+            CompiledConstraintSet::from_solution(py, solution)
+        })?;
         Ok(Self {
-            compiled: CompiledConstraintSet::from_solution(py, solution)?,
+            family,
+            compiled,
             cached_rows: None,
             join_key_cache: JoinKeyCache::new(),
             join_index_cache: JoinIndexCache::new(),
@@ -490,8 +503,10 @@ impl PyDynamicConstraintSet {
 
 impl ConstraintSet<PyDynamicSolution, DynamicScore> for PyDynamicConstraintSet {
     fn evaluate_all(&self, solution: &PyDynamicSolution) -> DynamicScore {
-        Python::attach(|py| evaluate_compiled_constraints(py, solution, &self.compiled))
-            .unwrap_or_else(panic_with_py_err)
+        scoped_dynamic_score_family(self.family, || {
+            Python::attach(|py| evaluate_compiled_constraints(py, solution, &self.compiled))
+                .unwrap_or_else(panic_with_py_err)
+        })
     }
 
     fn constraint_count(&self) -> usize {
@@ -517,8 +532,11 @@ impl ConstraintSet<PyDynamicSolution, DynamicScore> for PyDynamicConstraintSet {
     }
 
     fn initialize_all(&mut self, solution: &PyDynamicSolution) -> DynamicScore {
-        self.initialize_incremental_state(solution)
-            .unwrap_or_else(panic_with_py_err)
+        let family = self.family;
+        scoped_dynamic_score_family(family, || {
+            self.initialize_incremental_state(solution)
+                .unwrap_or_else(panic_with_py_err)
+        })
     }
 
     fn on_insert_all(
@@ -527,21 +545,24 @@ impl ConstraintSet<PyDynamicSolution, DynamicScore> for PyDynamicConstraintSet {
         entity_index: usize,
         descriptor_index: usize,
     ) -> DynamicScore {
-        let delta = Python::attach(|py| {
-            self.evaluate_delta(
-                py,
-                solution,
-                DeltaTarget {
-                    entity_index,
-                    descriptor_index,
-                    mode: DeltaMode::Insert,
-                },
-            )
+        let family = self.family;
+        scoped_dynamic_score_family(family, || {
+            let delta = Python::attach(|py| {
+                self.evaluate_delta(
+                    py,
+                    solution,
+                    DeltaTarget {
+                        entity_index,
+                        descriptor_index,
+                        mode: DeltaMode::Insert,
+                    },
+                )
+            })
+            .unwrap_or_else(panic_with_py_err);
+            self.retracted_entities
+                .remove(&(descriptor_index, entity_index));
+            delta
         })
-        .unwrap_or_else(panic_with_py_err);
-        self.retracted_entities
-            .remove(&(descriptor_index, entity_index));
-        delta
     }
 
     fn on_retract_all(
@@ -550,21 +571,24 @@ impl ConstraintSet<PyDynamicSolution, DynamicScore> for PyDynamicConstraintSet {
         entity_index: usize,
         descriptor_index: usize,
     ) -> DynamicScore {
-        let delta = Python::attach(|py| {
-            self.evaluate_delta(
-                py,
-                solution,
-                DeltaTarget {
-                    entity_index,
-                    descriptor_index,
-                    mode: DeltaMode::Retract,
-                },
-            )
+        let family = self.family;
+        scoped_dynamic_score_family(family, || {
+            let delta = Python::attach(|py| {
+                self.evaluate_delta(
+                    py,
+                    solution,
+                    DeltaTarget {
+                        entity_index,
+                        descriptor_index,
+                        mode: DeltaMode::Retract,
+                    },
+                )
+            })
+            .unwrap_or_else(panic_with_py_err);
+            self.retracted_entities
+                .insert((descriptor_index, entity_index));
+            delta
         })
-        .unwrap_or_else(panic_with_py_err);
-        self.retracted_entities
-            .insert((descriptor_index, entity_index));
-        delta
     }
 
     fn reset_all(&mut self) {

src/manager/candidate_trace.rs

@@ -384,5 +384,6 @@ fn disposition_name(disposition: CandidateTraceDisposition) -> &'static str {
         CandidateTraceDisposition::ForagerIgnored => "forager_ignored",
         CandidateTraceDisposition::Selected => "selected",
         CandidateTraceDisposition::Applied => "applied",
+        CandidateTraceDisposition::DiscardedSpeculative => "discarded_speculative",
     }
 }

src/manager/events.rs

@@ -132,6 +132,7 @@ fn telemetry_to_python(py: Python<'_>, telemetry: &SolverTelemetry) -> PyResult<
     dict.set_item("step_count", telemetry.step_count)?;
     dict.set_item("moves_generated", telemetry.moves_generated)?;
     dict.set_item("moves_evaluated", telemetry.moves_evaluated)?;
+    dict.set_item("moves_speculative", telemetry.moves_speculative)?;
     dict.set_item("moves_accepted", telemetry.moves_accepted)?;
     dict.set_item("moves_applied", telemetry.moves_applied)?;
     dict.set_item("moves_score_improving", telemetry.moves_score_improving)?;

src/solver/api.rs

@@ -12,6 +12,7 @@ use crate::error::{panic_to_py_err, py_err};
 use crate::runtime::dynamic_assignment_group::validate_assignment_construction_groups;
 use crate::schema::compiled::CompiledSchema;
 use crate::score::{scoped_dynamic_score_family, score_family_from_name, DynamicScorePythonExt};
+use crate::solver::solvable::dynamic_constraints;
 use crate::state::marshal::{export_solution, import_solution};
 use crate::state::PyDynamicSolution;
 
@@ -58,15 +59,14 @@ pub fn solve(
     let py_solution = solution.bind(py);
     let mut imported = import_solution(py_solution, Arc::clone(&plan))?;
     let descriptor = plan.descriptor().clone();
-    let constraints = PyDynamicConstraintSet::from_solution(py, &imported)?;
     validate_assignment_construction_groups(&solver_config, parsed)?;
     imported.solver_config = solver_config.clone();
     let score_family = score_family_from_name(&parsed.score_family)?;
     let mut solved = catch_unwind(AssertUnwindSafe(|| {
         scoped_dynamic_score_family(score_family, || {
             try_run_dynamic_solver_with_config_parts(
                 imported,
-                constraints,
+                dynamic_constraints,
                 descriptor,
                 dynamic_entity_count,
                 SolverRuntime::detached(),

src/solver/solvable.rs

@@ -21,15 +21,12 @@ impl Solvable for PyDynamicSolution {
         let descriptor = runtime_plan.descriptor().clone();
         let model = runtime_plan.model().clone();
         let default_time_limit_secs = runtime_plan.default_time_limit_secs();
-        let constraints = pyo3::Python::attach(|py| {
-            PyDynamicConstraintSet::from_solution(py, &self).unwrap_or_else(panic_with_py_err)
-        });
         let score_family = score_family_from_name(&runtime_plan.schema().score_family)
             .unwrap_or_else(panic_with_py_err);
         let result = scoped_dynamic_score_family(score_family, || {
             try_run_dynamic_solver_with_config_parts(
                 self,
-                constraints,
+                dynamic_constraints,
                 descriptor,
                 dynamic_entity_count,
                 runtime,
@@ -44,6 +41,14 @@ impl Solvable for PyDynamicSolution {
     }
 }
 
+/// Builds the constraint set for one solution: the working solution and every
+/// evaluator replica use this same constructor.
+pub(crate) fn dynamic_constraints(solution: &PyDynamicSolution) -> PyDynamicConstraintSet {
+    pyo3::Python::attach(|py| {
+        PyDynamicConstraintSet::from_solution(py, solution).unwrap_or_else(panic_with_py_err)
+    })
+}
+
 fn dynamic_entity_count(solution: &PyDynamicSolution, descriptor_index: usize) -> usize {
     solution.entity_count(descriptor_index)
 }

tests/python/test_threading.py

@@ -1,10 +1,72 @@
 from concurrent.futures import ThreadPoolExecutor
 
 from tests.python.test_scalar_solving import Schedule
-from solverforge import Solver
+from solverforge import (
+    ConstraintFactory,
+    HardSoftScore,
+    Solver,
+    constraint_provider,
+    planning_entity,
+    planning_solution,
+    planning_variable,
+)
 
 
 def test_concurrent_python_solves_do_not_share_solution_state() -> None:
     with ThreadPoolExecutor(max_workers=2) as pool:
         results = list(pool.map(lambda _: Solver.solve(Schedule()), range(2)))
     assert results[0] is not results[1]
+
+
+@planning_entity
+class ParallelVisit:
+    nurse = planning_variable(value_range_provider="nurses", allows_unassigned=True)
+
+    def __init__(self, preferred: int) -> None:
+        self.preferred = preferred
+        self.nurse = None
+
+
+@constraint_provider
+def parallel_constraints(factory: ConstraintFactory):
+    return [
+        factory.for_each(ParallelVisit)
+        .filter(lambda visit: visit.nurse is None)
+        .penalize(HardSoftScore.ONE_HARD)
+        .named("unassigned"),
+        factory.for_each(ParallelVisit)
+        .filter(lambda visit: visit.nurse is not None and visit.nurse != visit.preferred)
+        .penalize(HardSoftScore.ONE_SOFT)
+        .named("not preferred"),
+    ]
+
+
+@planning_solution(score=HardSoftScore, constraints=parallel_constraints)
+class ParallelPlan:
+    visits: list[ParallelVisit]
+
+    def __init__(self) -> None:
+        self.visits = [ParallelVisit((index * 7 + 3) % 5) for index in range(30)]
+        self.nurses = [0, 1, 2, 3, 4]
+        self.score = None
+
+
+def _solve_with_threads(count: int) -> ParallelPlan:
+    return Solver.solve(
+        ParallelPlan(),
+        {
+            "random_seed": 7,
+            "move_thread_count": {"count": count},
+            "phases": [
+                {"type": "construction_heuristic"},
+                {"type": "local_search", "termination": {"step_count_limit": 60}},
+            ],
+        },
+    )
+
+
+def test_parallel_move_evaluation_matches_sequential_at_fixed_work() -> None:
+    plans = [_solve_with_threads(count) for count in (1, 2, 4)]
+    assignments = [[visit.nurse for visit in plan.visits] for plan in plans]
+    assert assignments[0] == assignments[1] == assignments[2]
+    assert plans[0].score == plans[1].score == plans[2].score

tests/python/test_tracebacks.py

@@ -183,3 +183,48 @@ def test_dynamic_provider_callback_traceback_surfaces(
         Solver.solve(plan(), config)
     frames = traceback.extract_tb(exc_info.value.__traceback__)
     assert any(frame.name == function_name for frame in frames)
+
+
+@planning_entity
+class LateBrokenTask:
+    worker = planning_variable(value_range_provider="workers", allows_unassigned=True)
+
+    def __init__(self) -> None:
+        self.worker = 0
+
+
+@constraint_provider
+def late_broken_constraints(factory: ConstraintFactory):
+    def explode_on_second_worker(task: LateBrokenTask) -> bool:
+        if task.worker == 1:
+            raise ValueError("parallel callback failed")
+        return False
+
+    return [
+        factory.for_each(LateBrokenTask)
+        .filter(explode_on_second_worker)
+        .penalize(1)
+        .named("late broken")
+    ]
+
+
+@planning_solution(constraints=late_broken_constraints)
+class LateBrokenPlan:
+    tasks: list[LateBrokenTask]
+
+    def __init__(self) -> None:
+        self.tasks = [LateBrokenTask(), LateBrokenTask()]
+        self.workers = [0, 1]
+        self.score = None
+
+
+def test_callback_traceback_surfaces_from_parallel_evaluation() -> None:
+    config = {
+        "random_seed": 7,
+        "move_thread_count": {"count": 2},
+        "phases": [{"type": "local_search", "termination": {"step_count_limit": 1}}],
+    }
+    with pytest.raises(ValueError, match="parallel callback failed") as exc_info:
+        Solver.solve(LateBrokenPlan(), config)
+    frames = traceback.extract_tb(exc_info.value.__traceback__)
+    assert any(frame.name == "explode_on_second_worker" for frame in frames)

tests/rust/constraint_set.rs

@@ -2,7 +2,7 @@ use pyo3::Python;
 use solverforge_py::constraints::PyDynamicConstraintSet;
 use solverforge_py::schema::runtime_plan::CompiledRuntimePlan;
 use solverforge_py::schema::DynamicSchema;
-use solverforge_py::score::DynamicScore;
+use solverforge_py::score::{DynamicScore, ScoreFamily};
 use solverforge_py::state::callback_view::PythonCallbackView;
 use solverforge_py::state::entity_table::DynamicState;
 use solverforge_py::state::PyDynamicSolution;
@@ -14,7 +14,7 @@ fn dynamic_constraint_set_reports_constraint_count() {
     crate::initialize_python();
     Python::attach(|py| {
         let constraints = pyo3::types::PyList::empty(py).unbind().into_any();
-        let set = PyDynamicConstraintSet::new(constraints);
+        let set = PyDynamicConstraintSet::new(constraints, ScoreFamily::HardSoft);
         assert_eq!(set.constraint_count(), 0);
         let schema = Arc::new(DynamicSchema {
             solution_type: "Plan".to_string(),
@@ -40,6 +40,11 @@ fn dynamic_constraint_set_reports_constraint_count() {
             solverforge_config::SolverConfig::default(),
             0,
         );
-        assert_eq!(set.evaluate_all(&solution), DynamicScore::ZERO);
+        // The test thread never entered a solve's family scope, so this pins
+        // that the constraint set scores with its own declared family.
+        assert_eq!(
+            set.evaluate_all(&solution),
+            DynamicScore::zero_for_family(ScoreFamily::HardSoft)
+        );
     });
 }