From 5b940f9111939fe9b55396706efff3f540bb9b90 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 20:41:47 +0200 Subject: [PATCH 1/3] feat(manager): report speculative evaluation from parallel local search Exposes the solver's new moves_speculative telemetry counter and the discarded_speculative candidate-trace outcome, so Python callers see the work parallel evaluation performed ahead of the decision point and then discarded. --- src/manager/candidate_trace.rs | 1 + src/manager/events.rs | 1 + 2 files changed, 2 insertions(+) diff --git a/src/manager/candidate_trace.rs b/src/manager/candidate_trace.rs index f6e80b4..eade17c 100644 --- a/src/manager/candidate_trace.rs +++ b/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", } } diff --git a/src/manager/events.rs b/src/manager/events.rs index 0b38198..eb8e7cf 100644 --- a/src/manager/events.rs +++ b/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)?; From a288e35ea9d310332dc83df369fb3f9cb80b1dc0 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 20:41:47 +0200 Subject: [PATCH 2/3] feat(solver): evaluate Python models on several cores Every solve entry now hands the runtime a constraint-set constructor instead of a ready-made set, so the solver builds one set for the working solution and one for each evaluator replica, and move_thread_count applies to Python models. The bridge's active score family is per thread and only the solve thread enters it. Replicas are scored on Rayon worker threads, so PyDynamicConstraintSet now records its declared family and enters it inside every ConstraintSet method; scores stay comparable whichever thread produced them. --- src/constraints/compiled_dispatch_tests.rs | 7 +- src/constraints/state.rs | 94 ++++++++++++++-------- src/solver/api.rs | 4 +- src/solver/solvable.rs | 13 ++- tests/rust/constraint_set.rs | 11 ++- 5 files changed, 84 insertions(+), 45 deletions(-) diff --git a/src/constraints/compiled_dispatch_tests.rs b/src/constraints/compiled_dispatch_tests.rs index f50d968..7e18513 100644 --- a/src/constraints/compiled_dispatch_tests.rs +++ b/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() diff --git a/src/constraints/state.rs b/src/constraints/state.rs index 635c429..177613c 100644 --- a/src/constraints/state.rs +++ b/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) -> Self { + pub fn new(constraints: Py, 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 { + 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 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 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 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 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) { diff --git a/src/solver/api.rs b/src/solver/api.rs index 55f43c6..eddbce4 100644 --- a/src/solver/api.rs +++ b/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,7 +59,6 @@ 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)?; @@ -66,7 +66,7 @@ pub fn solve( scoped_dynamic_score_family(score_family, || { try_run_dynamic_solver_with_config_parts( imported, - constraints, + dynamic_constraints, descriptor, dynamic_entity_count, SolverRuntime::detached(), diff --git a/src/solver/solvable.rs b/src/solver/solvable.rs index 3874c87..0f8c925 100644 --- a/src/solver/solvable.rs +++ b/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) } diff --git a/tests/rust/constraint_set.rs b/tests/rust/constraint_set.rs index 36820bf..06e0544 100644 --- a/tests/rust/constraint_set.rs +++ b/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) + ); }); } From 79cb23da2c49f28930068d9419e53a087bf08031 Mon Sep 17 00:00:00 2001 From: Vittorio Distefano Date: Thu, 22 Oct 2026 20:41:48 +0200 Subject: [PATCH 3/3] test: pin parallel evaluation parity and worker-thread tracebacks A fixed-work solve reaches the same assignment and score with move_thread_count 1, 2 and 4, and a constraint callback that raises on a worker thread surfaces its Python traceback. --- tests/python/test_threading.py | 64 ++++++++++++++++++++++++++++++++- tests/python/test_tracebacks.py | 45 +++++++++++++++++++++++ 2 files changed, 108 insertions(+), 1 deletion(-) diff --git a/tests/python/test_threading.py b/tests/python/test_threading.py index d5e22f4..af5a9ce 100644 --- a/tests/python/test_threading.py +++ b/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 diff --git a/tests/python/test_tracebacks.py b/tests/python/test_tracebacks.py index 355bad5..6e0f9ea 100644 --- a/tests/python/test_tracebacks.py +++ b/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)