[blackopsrepl]

Lets Python models use move_thread_count: local-search candidates are evaluated on several cores within one solve, with the same trajectory at fixed work. The engine side is SolverForge/solverforge#78.

Draft, blocked on the solverforge release that ships #78. These commits compile only against that release; CI stays red until the pins move (P0).

Verified before the split, against #78's crates through a local, uncommitted [patch.crates-io], on CPython 3.14.3 free-threaded with the GIL disabled: Rust tests pass and 330 Python tests pass, including the new 1/2/4-thread parity test and the worker-thread traceback test. test_examples_playwright.py crashed inside greenlet during a browser asset import that never runs the solver and was excluded from that run.

Python scaling is measured with the engine benchmark. Each dynamic constraint-set call attaches its worker thread to the interpreter; if Python scaling flattens early, profile thread attachment first.

Checklist

P0 — chore: adopt solverforge <release> (the version bump needs the owner's confirmation)

• Cargo.toml: bump the six =0.22.0 solverforge pins to the release.
• src/manager/candidate_trace.rs, disposition_name: add CandidateTraceDisposition::DiscardedSpeculative => "discarded_speculative".
• src/manager/events.rs: add dict.set_item("moves_speculative", telemetry.moves_speculative)?; next to moves_evaluated (around line 134), and the new field to the SolverTelemetry { ... } literal (around line 400).
• Search WIREFRAME.md, README.md and python/solverforge/_native.pyi for interrupted_before_evaluation and moves_evaluated, and add the new disposition and telemetry key wherever those lists appear.

P1 — feat: evaluate Python models on evaluator replicas

• src/solver/solvable.rs. Add

  fn dynamic_constraints(solution: &PyDynamicSolution) -> PyDynamicConstraintSet {
      pyo3::Python::attach(|py| {
          PyDynamicConstraintSet::from_solution(py, solution).unwrap_or_else(panic_with_py_err)
      })
  }

  delete the inline let constraints = ... construction, and pass dynamic_constraints as the second argument of try_run_dynamic_solver_with_config_parts. The surrounding scoped_dynamic_score_family call stays: it covers the solve thread.
• src/constraints/state.rs:
  • Add family: ScoreFamily to PyDynamicConstraintSet, importing scoped_dynamic_score_family, score_family_from_name and ScoreFamily from crate::score.
  • new(constraints, family: ScoreFamily) stores the family.
  • from_solution reads let family = score_family_from_name(&solution.schema().score_family)?; and compiles inside the scope: let compiled = scoped_dynamic_score_family(family, || CompiledConstraintSet::from_solution(py, solution))?;. Compiling then gives the same result on every thread.
  • In impl ConstraintSet<PyDynamicSolution, DynamicScore> for PyDynamicConstraintSet, wrap the bodies of evaluate_all, initialize_all, on_insert_all and on_retract_all in the scope. Copy the family into a local first so the closure can borrow self mutably:

    let family = self.family;
    scoped_dynamic_score_family(family, || {
        // existing body, unchanged
    })

• tests/rust/constraint_set.rs. In dynamic_constraint_set_reports_constraint_count, construct the set with PyDynamicConstraintSet::new(constraints, ScoreFamily::HardSoft) and expect DynamicScore::zero_for_family(ScoreFamily::HardSoft) from evaluate_all. The test thread never enters a solve scope, so this pins thread-independent scoring.
• README.md:
  • In the free-threaded paragraph, add that move_thread_count evaluates local-search candidates on several Rust worker threads, each scoring its own solution clone. Callbacks already have to be safe to run concurrently, so there is no new requirement. GIL builds stay correct but serialize evaluation.
  • In Current Support, add move_thread_count with the dict form {"move_thread_count": {"count": 4}}.
• WIREFRAME.md. In the config paragraph (around line 231), add that move_thread_count passes through to the upstream solver unchanged.

P2 — test: parallel evaluation parity for Python models

• tests/python/test_threading.py. Add the model and test below, extending the imports from solverforge as needed:

  @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

  The score dictionaries carry the family, so the score assertion also catches family drift on worker threads.
• tests/python/test_tracebacks.py. Add a model whose callback fails only during local search, mirroring BrokenPlan: two tasks pre-assigned to worker 0, workers [0, 1], and a filter explode_on_second_worker that raises ValueError("parallel callback failed") when task.worker == 1. Solve it with {"random_seed": 7, "move_thread_count": {"count": 2}, "phases": [{"type": "local_search", "termination": {"step_count_limit": 1}}]} and assert, as test_callback_traceback_surfaces does, that the ValueError surfaces with a frame named explode_on_second_worker.
• Verify: make test on CPython 3.14 free-threaded.