[blackopsrepl]

Refs #76

Draft. The commit series below is implemented on this branch. It stays a draft until the quiet-host regression gate and the maintainer benchmark (see Verification gates) are recorded.

Goal

Evaluate local-search candidates on several cores within one solve, for static Rust models and dynamic host-language models alike, without changing the search algorithm:

• one producer generates candidates in canonical order on the main score director;
• persistent evaluator replicas — each an owned working solution with its own initialized incremental score director — run the existing trial transaction (preview, apply, score, undo, restore) in parallel with the main director, which is otherwise idle while a window is evaluated. N evaluation threads need N − 1 replicas;
• one coordinator consumes the results in candidate order through the existing acceptor and forager, then performs the authoritative commit;
• the selected move is applied to every replica by borrow, then moves by ownership into the commit. No move is ever cloned.

Rayon is the only executor: one rayon::scope per evaluation window, on the solve thread's current pool.

Non-goals

• Portfolio / multi-trajectory search — a separate feature that can reuse the replica type introduced here.
• Parallel candidate generation. Generation stays serial and canonical.
• Variable neighborhood descent, construction, exhaustive search, and partitioned child solves. They keep evaluating sequentially.
• Streaming (non-windowed) execution. It remains a later comparison against this windowed baseline.

Design

Evaluator replicas

Director::fork_replica(&self) -> Option<Self> builds an independent replica: a clone of the working solution plus a constraint set built for that clone and initialized against it. ScoreDirector can fork when it carries a constraint-set constructor, set with with_replica_constraints(fn(&S) -> C).

Every solve entry point takes that constructor in place of a ready-made constraint set. It builds the main director's constraint set and every replica's with it, so main and replica sets are built the same way and every model can fork:

• the generated solve passes |_| define_constraints(), wrapping the model's zero-argument constraints function;
• solverforge_bridge::try_run_dynamic_solver_with_config_parts takes the same fn(&S) -> C. Bindings pass a function that rebuilds their host constraint set from the solution it receives; in solverforge-py that is PyDynamicConstraintSet::from_solution.

When more than one evaluator is configured, a director that cannot fork is a wiring error and panics. There is no sequential fallback.

Replica initialization clones the working solution: once per replica at local-search phase start, and again whenever something other than the loop's own commits writes the working solution (construction before the phase, problem changes, the partitioned-search merge).

• The copy is unavoidable. Trials write variable values into the director's working solution, so concurrent trials need disjoint solutions, and a solve owns exactly one.
• Replaying the outside writers on each replica instead would run user code such as ProblemChange::apply once per replica.
• The main director evaluates alongside the replicas, so N evaluation threads need N − 1 copies.

This is the same class of clone as partitioned search's partition copies. It is cold — never per candidate, never per step.

Keeping replicas in sync

SolverScope gains a monotonic working_solution_generation. It increments on every committed mutation and on every wholesale replacement, and it never resets; the existing solution_revision does reset on replacement. The evaluator pool records the generation it is synchronized to:

• At each step start, a generation mismatch rebuilds the replicas. This covers construction between phases, problem changes, solution replacement, and any other committed writer.
• After the loop's own commit, each replica applies the selected move through &MoveCandidateRef (do_move borrows). The replica's score must equal the committed step score; a mismatch panics with a divergence message, which the runtime reports through the existing Failed lifecycle. The pool then records the new generation.

Dynamic models

Dynamic models run through the same compiled runtime and the same window loop. Parallel evaluation asks four things of a binding:

• A constructor. Described above: the binding rebuilds its constraint set from whichever solution it receives.
• Isolated clones. A replica is a clone of the working solution, so a clone must not share mutable host objects with the original. solverforge-py already guarantees this: callback views project entity and fact rows per clone.
• Thread-independent scoring. Replica trials run on Rayon worker threads. scoped_dynamic_score_family is thread-local, and only the solve thread enters it. A constraint set that builds DynamicScore values must therefore enter its declared family inside its own ConstraintSet methods; otherwise worker-computed scores do not match the solve thread's.
• Concurrency-safe callbacks. Host constraint callbacks run on several workers at once. solverforge-py already requires callbacks to be safe to run concurrently on free-threaded CPython 3.14. GIL builds stay correct but serialize evaluation.

Host candidate providers run during production on the coordinator, exactly as today. A host exception raised inside a trial leaves rayon::scope with its original panic payload, so host tracebacks survive.

The step

Each step alternates between producing, evaluating and consuming a window of candidates:

1. Produce up to capacity candidates serially on the main director, in canonical order. All existing control polls, trace pulls and generation telemetry run here exactly as today. Cartesian previews mutate and restore only the main director.
2. Dispatch check: the existing "before evaluation" interruption check, applied once before the window is evaluated.
3. Evaluate every slot. Sequentially, the window holds one candidate and the trial runs inline on the main director. In parallel, the coordinator borrows each slot's MoveCandidateRef from the cursor (shared borrows only). The main director and every replica pull slot indexes from one AtomicUsize and run the same trial kernel on their own director; the main director's share runs inline on the coordinator thread.
4. Consume results in slot order through the unchanged accounting, acceptor, forager and release logic. Before consuming slot i > 0, the coordinator re-applies the forager quit check and the interruption check that the sequential loop would have made at that point. Any slot not consumed is discarded as speculative work.

capacity is 1 when no replicas exist. Otherwise it is min(PARALLEL_EVALUATION_WINDOW, remaining in-phase move/score-calculation budget), with PARALLEL_EVALUATION_WINDOW = 32. The cursor never needs to be Sync: the coordinator borrows the window's candidates, and workers receive only that shared slice.

Determinism and accounting contract

• move_thread_count resolving to 1 runs the current sequential loop: same candidates, same checks in the same order, same telemetry counts.
• With two or more evaluators, the window size is fixed at 32, independent of the evaluator count. For a fixed configuration and fixed work, the trajectory is therefore identical for every evaluator count ≥ 2.
• For selectors whose cursors keep no stream state across steps — the default scalar change/swap families, for example — the parallel trajectory equals the sequential one.
  • Ruin selectors keep their RNG streams across steps, and provider leaves activate lazily. Candidates pulled ahead of the decision point advance that state, so those models follow a different, equally reproducible trajectory once parallel evaluation is on.
• Logical counters (moves_evaluated, score_calculations, step, move-count and score-calculation limits) count only consumed candidates. Fixed-work limits therefore trigger at the same point as in sequential evaluation.
• Physical work is reported, not hidden:
  • moves_generated counts every pull.
  • The new SolverTelemetry::moves_speculative counts candidates that were evaluated ahead of the decision point and then discarded.
  • The candidate trace records discarded pulls with the new terminal disposition CandidateTraceDisposition::DiscardedSpeculative.
• Work budgets are reserved before dispatch: the window never exceeds the remaining in-phase move or score-calculation budget. Wall-clock limits are checked during production and consumption; a window in flight completes first.
• None of this depends on the model kind. It holds for dynamic models whose bindings meet the requirements above.

Configuration

SolverConfig::move_thread_count is already part of the configuration surface. Today only partitioned search reads it.

move_thread_count	Acceptor-forager local search
None	sequential
Auto (default)	sequential in this PR — see decision D1
Count(0), Count(1)	sequential
Count(n ≥ 2)	min(n, rayon::current_num_threads()) evaluators: the main director plus that many minus one replicas

TOML: move_thread_count = { count = 4 }. Python configs pass the same key through: {"move_thread_count": {"count": 4}}.

Rules for implementers

• Read AGENTS.md and load the session KV: nisaba --quiet docs/agents/drift-guard.nisaba scan drift-guard/. Mirror this checklist into your task ledger before the first code change.
• No unsafe, no unsafe impl Send/Sync, no Arc/Rc, no new dyn, no std::thread::spawn or std::thread::sleep, no .clone() of moves, candidates or scoring state. The only new clone is replica initialization in ScoreDirector::fork_replica_impl.
• Phantom type parameters use PhantomData<fn() -> T>. mod.rs files contain declarations and re-exports only. New files stay under 500 lines. Several files touched here are already over that size (run.rs, scope_core.rs, scope_progress.rs, stats/solver.rs, stats/candidate_trace.rs): add only the lines listed for them, and put new logic in new files.
• Run Cargo commands directly — never through a pipe — and read the real exit status. Per commit: cargo check -p <crate> --all-targets --message-format short, the listed tests, cargo clippy -p <crate> --all-targets -- -D warnings, and cargo fmt --all -- --check.
• When the compiler contradicts this plan, stop and report it on this PR instead of working around it. Examples: MoveCandidateRef turns out not to be Sync, or a struct literal needs a field the plan does not mention.
• Do not modify solverforge-bench, do not bump versions, and do not hand-edit CHANGELOG.md. solverforge-py changes belong to their own pull request (see Downstream).
• One conventional commit per section below, with the message given. Do not merge sections.

---

Commit plan

C1 — feat(scoring): fork evaluator replicas from ScoreDirector

• crates/solverforge-scoring/src/director/traits.rs. Add a defaulted method to Director<S>, after reset. The where Self: Sized clause keeps the trait object-safe for the existing &dyn Director<S> boundaries.

  /// Builds an independent evaluator replica: an owned copy of the working
  /// solution with freshly initialized incremental constraint state.
  /// Returns `None` when this director cannot construct one.
  fn fork_replica(&self) -> Option<Self>
  where
      Self: Sized,
  {
      None
  }

• crates/solverforge-scoring/src/director/score_director/incremental.rs:
  • Add the field replica_constraints: Option<fn(&S) -> C> to ScoreDirector, after entity_counter, and initialize it to None in with_descriptor. Every other constructor delegates to with_descriptor; confirm that.
  • In the impl<S, C> ScoreDirector<S, C> block that defines with_descriptor, add:

    /// Records the constructor that builds each evaluator replica's constraint set.
    pub fn with_replica_constraints(mut self, constraints: fn(&S) -> C) -> Self {
        self.replica_constraints = Some(constraints);
        self
    }
    
    pub(crate) fn fork_replica_impl(&self) -> Option<Self> {
        let constraints = self.replica_constraints?;
        let solution = self.working_solution.clone();
        let replica_constraints = constraints(&solution);
        let mut replica = Self::with_descriptor(
            solution,
            replica_constraints,
            self.solution_descriptor.clone(),
            self.entity_counter,
        );
        replica.replica_constraints = Some(constraints);
        replica.calculate_score_impl();
        Some(replica)
    }

  • The manual Debug impl lives in adapters.rs; show the new field there as replica_constraints.is_some().
• crates/solverforge-scoring/src/director/score_director/adapters.rs. In impl<S, C> Director<S> for ScoreDirector<S, C>, add fn fork_replica(&self) -> Option<Self> { self.fork_replica_impl() }.
• crates/solverforge-scoring/src/director/tests/score_director.rs (add at most 90 lines). Add a helper fn unassigned_constraints(_: &TestSolution) -> (impl IncrementalConstraint<TestSolution, SoftScore>,) { (make_unassigned_constraint(),) } and three tests:
  • fork_replica_requires_replica_constraints: a director built without with_replica_constraints returns None.
  • fork_replica_matches_score_and_stays_independent:
    1. Build ScoreDirector::new(solution.clone(), unassigned_constraints(&solution)).with_replica_constraints(unassigned_constraints) and call calculate_score().
    2. Fork it and assert the replica's get_score() equals the main director's.
    3. Change the replica with before_variable_changed, a direct write, then after_variable_changed.
    4. Assert the replica's score changed while the main director's score and working solution did not.
  • fork_of_uninitialized_director_initializes_the_replica: fork before any calculate_score(), then assert the replica is initialized and its score is correct.
• crates/solverforge-scoring/WIREFRAME.md. Document Director::fork_replica in the trait method list and with_replica_constraints in the ScoreDirector public methods (around line 507).
• Verify: cargo test -p solverforge-scoring fork_replica, cargo test -p solverforge-scoring, then clippy and fmt.

C2 — feat(solver): build constraint sets from a constructor at every solve entry

• crates/solverforge-solver/src/run.rs:
  • In try_run_solver_with_config_and_search, try_run_solver_with_config_and_search_request and try_run_solver_with_candidate_trace_request, change the parameter constraints: C to constraints: fn(&S) -> C and forward it along the chain.
  • In try_run_solver_with_candidate_trace_request, keep log_scale(&solution); first and replace the director construction with:

    let main_constraints = constraints(&solution);
    let director = ScoreDirector::with_descriptor(
        solution,
        main_constraints,
        descriptor.clone(),
        entity_count_by_descriptor,
    )
    .with_replica_constraints(constraints);

• crates/solverforge-macros/src/planning_solution/runtime/solve.rs. In solve_expr, replace the argument #constraints_fn(), with (|_: &Self| #constraints_fn()) as fn(&Self) -> _,.
• crates/solverforge-bridge/src/runner.rs. In try_run_dynamic_solver_with_config_parts, change constraints: C to constraints: fn(&S) -> C and forward it. Add to its doc comment: "constraints builds a constraint set for a given solution. The runtime calls it for the working solution and for every evaluator replica, so move_thread_count evaluates dynamic models in parallel."
• crates/solverforge-bridge/src/score.rs. Extend the doc comment of scoped_dynamic_score_family: "The family is per thread. The solve thread enters it for the whole solve, but parallel evaluation scores evaluator replicas on Rayon worker threads, so a constraint set that builds DynamicScore values must also enter its declared family inside its own ConstraintSet methods."
• Test callers:
  • crates/solverforge-solver/src/run_tests.rs (around line 168): add fn test_constraints(_: &TestSolution) -> ScoreFromSolutionConstraints { ScoreFromSolutionConstraints } and pass test_constraints.
  • crates/solverforge-solver/src/manager/mod_tests_integration/lifecycle_solutions.rs (around line 565): add fn zero_variable_constraints(_: &ZeroVariableLifecycleSolution) {} and pass zero_variable_constraints in place of ().
• crates/solverforge-macros/src/planning_solution_tests.rs. Extend the existing expansion test that checks for try_run_solver_with_config_and_search so it also asserts the constructor cast. Print the expansion once to get quote!'s exact spacing (for example fn (& Self) -> _).
• Docs:
  • Search every crates/*/WIREFRAME.md for try_run_solver_with_config_and_search and try_run_dynamic_solver_with_config_parts, and update each signature to the constructor parameter.
  • crates/solverforge-macros/WIREFRAME.md: the solve_internal bullet (near line 257) says the constraints function is passed as the constraint-set constructor.
  • crates/solverforge-bridge/WIREFRAME.md: the runner's constructor parameter, and the per-thread rule for scoped_dynamic_score_family.
• Verify: cargo check --workspace --all-targets --message-format short, cargo test -p solverforge-macros, cargo test -p solverforge --tests (the facade fixtures exercise the generated code), cargo test -p solverforge-bridge, cargo test -p solverforge-solver.

C3 — feat(solver): track working-solution generations

Landed inside C7: the counter has no reader until the evaluator pool exists, and the lint gate rejects dead code.

• crates/solverforge-solver/src/scope/solver/scope_core.rs:
  • Add the field working_solution_generation: u64 to SolverScope, after solution_revision. Initialize it to 0 in every constructor; the compiler lists the struct literals.
  • Add self.working_solution_generation += 1; as the first statement of replace_working_solution_and_reinitialize.
  • Add the accessor pub(crate) fn working_solution_generation(&self) -> u64.
• crates/solverforge-solver/src/scope/solver/scope_progress.rs. In committed_mutation, add self.working_solution_generation += 1; directly after self.advance_solution_revision();.
• crates/solverforge-solver/src/scope/tests.rs (add at most 40 lines). Add working_solution_generation_is_monotonic_across_commits_and_replacement:
  1. Call mutate(|_| ()) twice and assert the generation advanced by 2.
  2. Call replace_working_solution_and_reinitialize(...) and assert it advanced by 1 more.
  3. Assert the generation is still strictly greater than every earlier value, even though solution_revision went back to 1.
• Verify: cargo test -p solverforge-solver scope::.

C4 — refactor(solver): split the local-search trial kernel from its accounting

Pure refactor; every existing test must pass unchanged.

• crates/solverforge-solver/src/phase/localsearch/evaluation.rs:
  • Add the result type:

    pub(crate) enum TrialOutcome<Sc> {
        NotDoable,
        Scored(Sc),
    }

  • Add run_trial. Its body is the director-only part of today's evaluate_candidate, in the same order: the pinned/doable check returning NotDoable, snapshot_score_state, do_move_with_scratch, calculate_score, undo_move_with_scratch, restore_score_state, then Scored(score). It must not touch any statistics.

    #[inline]
    pub(crate) fn run_trial<S, D, M>(
        mov: &MoveCandidateRef<'_, S, M>,
        director: &mut D,
        journal_pool: &mut Vec<MoveCandidateUndo<M::Undo>>,
    ) -> TrialOutcome<S::Score>

  • Add pub(crate) fn account_trial(mov, step_scope, reference_score, selector_index, elapsed: Duration, outcome: TrialOutcome<S::Score>) -> CandidateEvaluation<S::Score>. Its body is all of today's bookkeeping, in today's order:
    • NotDoable: evaluated-move record, unscored label, not-doable counters.
    • Scored: record_score_calculation, the hard-delta counters, then the hard-improvement and score-improvement rejections.
  • evaluate_candidate keeps its signature and becomes account_trial(mov, step_scope, reference_score, selector_index, evaluation_started.elapsed(), run_trial(mov, step_scope.score_director_mut(), journal_pool)). Bind the outcome first so elapsed() is taken after the trial.
  • Change record_evaluated_move(step_scope, selector_index, elapsed: Duration) to take a Duration. Update its callers: phase/localsearch/phase/candidates.rs (around line 220) and phase/localsearch/vnd/phase.rs (around line 341) both pass evaluation_started.elapsed().
• Verify: cargo test -p solverforge-solver localsearch, then the full cargo test -p solverforge-solver.

C5 — refactor(solver): extract local-search candidate generation and consumption

Pure refactor of crates/solverforge-solver/src/phase/localsearch/phase/candidates.rs; behaviour is byte-identical.

• Create crates/solverforge-solver/src/phase/localsearch/phase/decision.rs and declare mod decision; next to mod candidates; in phase/localsearch/phase.rs. It contains:
  • The slot type:

    pub(super) struct WindowSlot {
        pub(super) id: CandidateId,
        pub(super) selector_index: Option<usize>,
        pub(super) move_label: &'static str,
        pub(super) trace_token: Option<CandidateTracePullToken>,
    }

  • pub(super) fn record_generated_candidate(cursor: &C, candidate_id, generation_started: Instant, step_scope, state: &mut CandidateLoopState) -> WindowSlot. This is today's block from the selector-index lookup through the progress poll (lines 102–137): selector index, move and selector labels, record_candidate_pull, generated_moves += 1, generated-move telemetry with the generation elapsed time, and the progress poll.
  • pub(super) fn consume_candidate(cursor: &mut C, step_scope, slot: &WindowSlot, outcome: TrialOutcome<S::Score>, elapsed: Duration, last_step_score, acceptor, forager, requires_move_signatures, state: &mut CandidateLoopState). This is today's block from state.evaluated_moves += 1 to the end of the loop body (lines 156–303), with evaluate_candidate replaced by account_trial(&mov, …, elapsed, outcome) and the post-acceptor record_evaluated_move given elapsed. Re-borrow mov from cursor.candidate(slot.id), and keep its last use before any cursor.release_candidate call, as today.
• In candidates.rs, the loop body becomes:
  1. the unchanged control poll;
  2. the pull;
  3. record_generated_candidate;
  4. the unchanged before-evaluation interruption check;
  5. run_trial timed with Instant::now();
  6. consume_candidate.
• Verify: cargo test -p solverforge-solver localsearch; every existing foraging, streaming and selector test passes with no expectation changed.

C6 — feat(solver): report speculative evaluation in telemetry and candidate traces

• crates/solverforge-solver/src/stats/solver.rs. Add a moves_speculative: u64 field to SolverStats, declared like moves_evaluated, with pub fn record_move_speculative(&mut self), and copy it in the telemetry snapshot (around line 279).
• crates/solverforge-solver/src/stats/telemetry.rs. Add pub moves_speculative: u64 to SolverTelemetry, documented as "Candidates evaluated ahead of the sequential decision point by parallel evaluation and then discarded. Zero for sequential evaluation." Initialize it to 0 in new_const.
• crates/solverforge-solver/src/scope/solver/scope_progress.rs. Add pub(crate) fn record_move_speculative(&mut self) next to record_score_calculation. crates/solverforge-solver/src/scope/phase.rs forwards it through PhaseScope, mirroring record_score_calculation around line 486.
• crates/solverforge-solver/src/stats/candidate_trace.rs. Add a DiscardedSpeculative variant documented as "Parallel evaluation pulled this candidate ahead of the sequential decision point; the step ended before the candidate was consumed." Give it code 10 in code() and list it in is_terminal().
• Add the new field to the struct literals in manager/mod_tests_integration/resume_support.rs and manager/mod_tests_integration/runtime_helpers.rs.
• crates/solverforge-solver/src/stats/tests.rs. Add speculative_moves_reach_the_telemetry_snapshot.
• crates/solverforge-solver/WIREFRAME.md. Add the SolverTelemetry field and the CandidateTraceDisposition variant.
• Verify: cargo test -p solverforge-solver stats::, cargo check --workspace --all-targets (the console crate must still build).

C7 — feat(solver): parallel evaluation windows for acceptor-forager local search

New module crates/solverforge-solver/src/phase/localsearch/parallel/:

• mod.rs — declarations and re-exports only:

  mod budget;
  mod pool;
  mod window;
  #[cfg(test)]
  mod tests;
  
  pub(crate) use budget::remaining_evaluation_budget;
  pub(crate) use pool::{EvaluatorPool, PARALLEL_EVALUATION_WINDOW};
  pub(crate) use window::evaluate_window;

• budget.rs:

  pub(crate) fn remaining_evaluation_budget<S, D, BestCb>(
      solver_scope: &SolverScope<'_, S, D, BestCb>,
  ) -> Option<u64>

  It returns the minimum of inphase_move_count_limit - stats().moves_evaluated and inphase_score_calc_count_limit - stats().score_calculations, both with saturating_sub, and None when neither limit is set. Both limit fields are already pub.
• pool.rs:

  /// Candidates per parallel window. Fixed so the trajectory does not depend on
  /// the number of evaluator replicas.
  pub(crate) const PARALLEL_EVALUATION_WINDOW: usize = 32;
  
  pub(crate) struct EvaluatorPool<S, D, M> {
      requested: usize,
      replicas: Vec<EvaluatorReplica<S, D, M>>,
      main_outputs: Vec<(usize, TrialOutcome<S::Score>, Duration)>,
      synced_generation: Option<u64>,
  }
  
  pub(super) struct EvaluatorReplica<S, D, M> {
      pub(super) director: D,
      pub(super) journal_pool: Vec<MoveCandidateUndo<M::Undo>>,
      pub(super) outputs: Vec<(usize, TrialOutcome<S::Score>, Duration)>,
  }

  Bounds: S: PlanningSolution, D: Director<S>, M: Move<S>. If the compiler reports an unused type parameter, add _phantom: PhantomData<fn() -> (S, M)>. Methods:
  • new(requested: usize) -> Self: stores requested.max(1) — the number of evaluators, main director included — with no replicas.
  • is_parallel(&self) -> bool: true when replicas exist.
  • window_capacity(&self, remaining: Option<u64>) -> usize: 1 when not parallel; otherwise PARALLEL_EVALUATION_WINDOW.min(remaining as usize).max(1), with a missing budget meaning unbounded.
  • sync(&mut self, solver_scope: &SolverScope<'_, S, D, BestCb>):
    • Return immediately when requested <= 1 or synced_generation == Some(solver_scope.working_solution_generation()).
    • Otherwise clear the replicas and fork requested - 1 of them with solver_scope.score_director().fork_replica().expect("parallel evaluation requires a director built with a constraint-set constructor"), then set synced_generation. The main director is the remaining evaluator.
  • advance(&mut self, selected: &MoveCandidateRef<'_, S, M>, expected: S::Score): when parallel, self.replicas.par_iter_mut().for_each(|replica| { let _undo = selected.do_move(&mut replica.director); let score = replica.director.calculate_score(); assert_eq!(score, expected, "evaluator replica diverged from the committed step"); }).
  • mark_synced(&mut self, generation: u64): when parallel, set synced_generation = Some(generation).
  • pub(super) fn parts_mut(&mut self) -> (&mut Vec<(usize, TrialOutcome<S::Score>, Duration)>, &mut [EvaluatorReplica<S, D, M>]): the main director's output buffer and the replicas, borrowed disjointly.
  • #[cfg(test)] pub(super) fn replica_count(&self) -> usize.
• window.rs. The coordinator borrows the window's candidates from the cursor and passes the slice in, so this function needs no cursor type and the cursor never has to be Sync. The main director takes its share inline on the coordinator thread; the replicas take theirs in spawned tasks.

  pub(crate) fn evaluate_window<S, D, M>(
      candidates: &[MoveCandidateRef<'_, S, M>],
      main_director: &mut D,
      main_journal_pool: &mut Vec<MoveCandidateUndo<M::Undo>>,
      evaluators: &mut EvaluatorPool<S, D, M>,
      results: &mut Vec<Option<(TrialOutcome<S::Score>, Duration)>>,
  ) where
      S: PlanningSolution,
      D: Director<S> + Send,
      M: Move<S>,
  {
      let next = AtomicUsize::new(0);
      let (main_outputs, replicas) = evaluators.parts_mut();
      main_outputs.clear();
      rayon::scope(|scope| {
          for replica in replicas.iter_mut() {
              let next = &next;
              replica.outputs.clear();
              scope.spawn(move |_| {
                  drain_window(
                      candidates,
                      next,
                      &mut replica.director,
                      &mut replica.journal_pool,
                      &mut replica.outputs,
                  )
              });
          }
          drain_window(candidates, &next, main_director, main_journal_pool, main_outputs);
      });
      results.clear();
      results.resize_with(candidates.len(), || None);
      for (index, outcome, elapsed) in main_outputs.drain(..) {
          results[index] = Some((outcome, elapsed));
      }
      for replica in replicas.iter_mut() {
          for (index, outcome, elapsed) in replica.outputs.drain(..) {
              results[index] = Some((outcome, elapsed));
          }
      }
  }
  
  fn drain_window<S, D, M>(
      candidates: &[MoveCandidateRef<'_, S, M>],
      next: &AtomicUsize,
      director: &mut D,
      journal_pool: &mut Vec<MoveCandidateUndo<M::Undo>>,
      outputs: &mut Vec<(usize, TrialOutcome<S::Score>, Duration)>,
  ) where
      S: PlanningSolution,
      D: Director<S>,
      M: Move<S>,
  {
      loop {
          let index = next.fetch_add(1, Ordering::Relaxed);
          let Some(candidate) = candidates.get(index) else { break };
          let started = Instant::now();
          let outcome = run_trial(candidate, director, journal_pool);
          outputs.push((index, outcome, started.elapsed()));
      }
  }

  The main director is in its committed state here: production restores every preview before a pull returns, which the sequential loop already relies on when it runs each trial on the main director.
  A worker panic propagates out of rayon::scope with its original payload, exactly like a sequential trial panic.

Modified files:

• phase/localsearch/mod.rs. Add mod parallel;.

• phase/localsearch/phase.rs:

  • solve_local_search_with_resources gains the parameter evaluator_count: usize after step_limit. It creates let mut evaluators = EvaluatorPool::<S, D, M>::new(evaluator_count); before the loop and passes &mut evaluators to execute_step.
  • LocalSearchPhase::solve passes 1.

• phase/localsearch/phase/step.rs. execute_step takes evaluators: &mut EvaluatorPool<S, D, M>:

  • Directly after StepScope::new: evaluators.sync(step_scope.phase_scope().solver_scope());.
  • Pass evaluators to evaluate_candidates.
  • In the commit block, after the tabu signature is computed and before step_scope.apply_committed_change(...): evaluators.advance(&selected_move, selected_score);.
  • Directly after apply_committed_change returns: evaluators.mark_synced(step_scope.phase_scope().solver_scope().working_solution_generation());.

• phase/localsearch/phase/decision.rs. Add the discard helpers:

  • interrupt_window_at(cursor, step_scope, window: &mut [WindowSlot], index, evaluated: bool):
    • Slot index: record the InterruptedBeforeEvaluation disposition, exactly as the sequential loop does, and do not release it.
    • Every later slot: record DiscardedSpeculative and release it.
    • When evaluated is true, call record_move_speculative() for slot index and every later slot.
  • discard_window_from(cursor, step_scope, window, from, evaluated: bool): record DiscardedSpeculative for slots from.., release them, and count them as speculative when evaluated is true.

• phase/localsearch/phase/candidates.rs. evaluate_candidates takes evaluators: &mut EvaluatorPool<S, D, M>, and its loop becomes:

  let mut journal_pool = Vec::new();
  let mut control_polls = 0usize;
  let mut control_latched = false;
  let mut window: Vec<WindowSlot> = Vec::new();
  let mut results = Vec::new();
  while !forager.is_quit_early() {
      if state.interrupted { break; }
  
      // 1. Production: serial, canonical order, main director.
      let capacity = evaluators
          .window_capacity(remaining_evaluation_budget(step_scope.phase_scope().solver_scope()));
      window.clear();
      let mut exhausted = false;
      while window.len() < capacity {
          control_polls += 1;
          if !control_latched && control_polls.is_multiple_of(CONTROL_POLL_INTERVAL) {
              control_latched = should_interrupt_before_candidate(step_scope)
                  || step_scope.phase_scope_mut().solver_scope_mut().should_terminate();
          }
          if control_latched { state.interrupted = true; break; }
          let generation_started = Instant::now();
          let Some(candidate_id) =
              cursor.next_candidate_with_resources(step_scope.score_director_mut(), resources)
          else { exhausted = true; break; };
          window.push(record_generated_candidate(&*cursor, candidate_id, generation_started, step_scope, &mut state));
      }
      if window.is_empty() { break; }
  
      // 2. Dispatch check: the sequential before-evaluation check for the first slot.
      if state.interrupted
          || should_interrupt_before_evaluation(step_scope)
          || step_scope.phase_scope_mut().solver_scope_mut().should_terminate()
      {
          interrupt_window_at(cursor, step_scope, &mut window, 0, false);
          state.interrupted = true;
          break;
      }
  
      // 3. Evaluation.
      if evaluators.is_parallel() {
          // Shared borrows only; dropped at the end of this block, before consumption.
          let candidates: Vec<_> = window
              .iter()
              .map(|slot| cursor.candidate(slot.id).expect("window candidate must remain borrowable"))
              .collect();
          evaluate_window(&candidates, step_scope.score_director_mut(), &mut journal_pool, evaluators, &mut results);
      } else {
          let mov = cursor.candidate(window[0].id).expect("discovered candidate id must remain borrowable");
          let started = Instant::now();
          let outcome = run_trial(&mov, step_scope.score_director_mut(), &mut journal_pool);
          results.clear();
          results.push(Some((outcome, started.elapsed())));
      }
  
      // 4. Consumption in candidate order.
      for index in 0..window.len() {
          if index > 0 {
              if forager.is_quit_early() {
                  discard_window_from(cursor, step_scope, &mut window, index, true);
                  break;
              }
              if should_interrupt_before_evaluation(step_scope)
                  || step_scope.phase_scope_mut().solver_scope_mut().should_terminate()
              {
                  interrupt_window_at(cursor, step_scope, &mut window, index, true);
                  state.interrupted = true;
                  break;
              }
          }
          let (outcome, elapsed) = results[index].take().expect("every window slot has a trial result");
          consume_candidate(cursor, step_scope, &window[index], outcome, elapsed, last_step_score,
              acceptor, forager, requires_move_signatures, &mut state);
      }
      if state.interrupted || exhausted { break; }
  }

  With capacity == 1 this performs the same checks, pulls, records and decisions in the same order as today. That property is what keeps every existing local-search test valid, so do not reorder anything to make the code shorter.

• runtime/compiler/executor/runner/local_search.rs. Pass 1 for now; C8 wires the configuration.

• parallel/tests.rs. Build a director with real incremental scoring over crate::test_utils::NQueensSolution:

  • Constraint: an solverforge_scoring::IncrementalUniConstraint over queens, built with solverforge_scoring::stream::collection_extract::{source, ChangeSource} and ChangeSource::Descriptor(0). Follow the pattern in solverforge-scoring/src/director/tests/score_director.rs, and use a non-constant weight such as (row * 3 + column) % 5. Wrap it in fn constraints(_: &NQueensSolution) -> (impl IncrementalConstraint<NQueensSolution, SoftScore>,).
  • Director: ScoreDirector::with_descriptor(solution.clone(), constraints(&solution), create_nqueens_descriptor(), |s, _| s.queens.len()).with_replica_constraints(constraints).
  • Scope: SolverScope::new(director).with_seed(7) followed by start_solving().
  • Source: SelectorCursorSource::new(ChangeMoveSelector::simple(get_queen_row, set_queen_row, 0, 0, "row", values)).

  Tests:

  • pool_with_one_evaluator_never_forks: build the director with ScoreDirector::simple (no constructor), create the pool with requested = 1, then sync; assert !is_parallel() and window_capacity(None) == 1.
  • pool_requires_a_forkable_director: the same director with requested = 2; #[should_panic(expected = "constraint-set constructor")] on sync.
  • pool_rebuilds_replicas_when_the_generation_changes: with requested = 4, sync and assert replica_count() == 3; commit a change through solver_scope.mutate(...), sync again, then assert the replica scores equal the main director's score.
  • sequential_and_parallel_reach_identical_trajectories: run HillClimbingAcceptor with AcceptedCountForager::new(4, false) and step limit 25 for evaluator counts 1, 2 and 4. Assert equal queen rows, moves_evaluated, moves_applied and step counts.
  • speculative_evaluations_are_reported_and_not_counted: run AcceptedCountForager::new(1, false) with evaluator counts 1 and 2. Assert equal moves_evaluated, a nonzero moves_speculative for count 2, and zero for count 1.
  • in_phase_move_limit_reserves_the_window: set solver_scope.inphase_move_count_limit = Some(5) and run BestScoreForager with evaluator count 2. Assert moves_evaluated == 5 and moves_speculative == 0.
  • replicas_follow_solution_replacement: run a phase, call replace_working_solution_and_reinitialize with a different solution, then run a second phase with evaluator count 2. The run must complete without a divergence panic and match the same sequence run with evaluator count 1.

• AGENTS.md:

  • Under Move Ownership → Clone is acceptable only for, replace item 2 with: "Parallel search creating worker-private solution copies: partitioned search's partition copies, and parallel local-search evaluation's evaluator replicas — one per evaluator beyond the main director, made at phase start and again after the working solution is written outside the parallel loop's own commits."
  • Under Threading, add: "Parallel local-search evaluation runs one rayon::scope per evaluation window on the solve thread's current pool; do not add dedicated pools or blocking waits inside evaluation workers."

• crates/solverforge-solver/WIREFRAME.md:

  • Add the new files to the file map: phase/localsearch/parallel/{mod,budget,pool,window,tests}.rs and phase/localsearch/phase/decision.rs.
  • Add a Parallel evaluation paragraph to the Local Search section that restates the determinism and accounting contract above.

• Verify: cargo test -p solverforge-solver localsearch, the full cargo test -p solverforge-solver, clippy and fmt.

C8 — feat(solver): drive acceptor-forager evaluation from move_thread_count

• phase/localsearch/parallel/config.rs (new; declare it in parallel/mod.rs and re-export it):

  pub(crate) fn resolve_evaluator_count(count: &MoveThreadCount) -> usize {
      match count {
          MoveThreadCount::Auto | MoveThreadCount::None => 1,
          MoveThreadCount::Count(count) => (*count).clamp(1, rayon::current_num_threads().max(1)),
      }
  }

• runtime/compiler/executor/runner/local_search.rs. Add evaluator_count: usize to RuntimeLocalSearch::AcceptorForager, destructure it in solve, and pass it to solve_local_search_with_resources.
• runtime/compiler/executor/runner/lowering.rs. Set evaluator_count: resolve_evaluator_count(&solver_config.move_thread_count) at both RuntimeLocalSearch::AcceptorForager construction sites: the default-runtime branch near line 82 and lower_local_search near line 163.
• parallel/tests.rs. Add evaluator_count_resolution: None, Auto, Count(0) and Count(1) resolve to 1, and Count(2) resolves to min(2, rayon::current_num_threads()).
• crates/solverforge-config/WIREFRAME.md. Under MoveThreadCount (around line 713) and the move_thread_count row (around line 63), document the consumers and the configuration table above.
• Verify: cargo test -p solverforge-solver runtime::, cargo test -p solverforge-solver, cargo test -p solverforge-config.

C9 — test(solverforge): parallel local-search parity through the generated runtime

• New fixture under crates/solverforge/tests/parallel_local_search/domain/, modelled on scalar_runtime_publication/domain/:
  • mod.rs: solverforge::planning_model! with root = "crates/solverforge/tests/parallel_local_search/domain" and modules plan, resource, task.
  • resource.rs: the same shape as the existing Resource.
  • task.rs: #[planning_id] pub id: String, pub weight: i64, and #[planning_variable(value_range_provider = "resources", allows_unassigned = true)] pub resource_idx: Option<usize>.
  • plan.rs: #[planning_solution(constraints = "define_constraints", config = "parallel_config")] with resources, tasks, score: Option<HardSoftScore>, and a plain field pub move_threads: u32.
    • Constraints: a hard 1 for each unassigned task, plus a soft penalty (task.weight * (resource_idx as i64 + 3)) % 17 for each assigned task.
    • parallel_config(solution, _base) parses the TOML below with SolverConfig::from_toml_str, sets config.move_thread_count to MoveThreadCount::Auto when move_threads == 0 and to MoveThreadCount::Count(n) otherwise, and returns the config.

    random_seed = 7
    
    [[phases]]
    type = "construction_heuristic"
    construction_heuristic_type = "first_fit"
    
    [[phases]]
    type = "local_search"
    
    [phases.termination]
    step_count_limit = 150

• crates/solverforge/tests/parallel_local_search.rs. Use 6 resources and 40 tasks with deterministic weights (for example (i * 7 + 3) % 11). Solve through SolverManager as in scalar_runtime_publication.rs.
  • fixed_work_trajectory_is_identical_across_thread_counts: for move_threads 1, 2 and 4, collect the completed solution and metadata.telemetry. Assert identical resource_idx vectors, scores, step_count, moves_evaluated and moves_applied. The default plain-scalar selectors keep no stream state across steps, so all three runs must match.
  • sequential_resolution_reports_no_speculation: with move_threads 0 (Auto) and 1, assert moves_speculative == 0.
  • Do not assert moves_speculative > 0 here: on a single-core runner Count(n) clamps to 1. The unit tests in C7 cover speculation independently of the core count.
• This fixture is a static model. Dynamic-model parity is pinned in the solverforge-py pull request.
• Verify: cargo test -p solverforge --test parallel_local_search, cargo test -p solverforge --tests.

C10 — docs: describe parallel local-search evaluation

• README.md. In the existing configuration or performance guidance, add two or three lines: move_thread_count = { count = 4 } evaluates local-search candidates on up to four cores, for Rust models and for Python models through solverforge-py; results are reproducible at fixed work. If the README has no such section, skip this file.
• docs/architecture.html. Check whether assets/solverforge-architecture.svg depicts the local-search step. If it does, add the evaluator replicas to the SVG and regenerate the page with node scripts/solverforge-architecture.js. Do not edit the HTML by hand. If the step is not depicted, make no change.

---

Verification gates

Run each gate once per landed series and record its verdict on this PR.

• Workspace: make build, make examples, make test, make lint, make pre-release. All must pass with zero ignored tests.
• Neighborhood regression gate: BASELINE=<merge-base of this branch with main> BASELINE_CURSOR_API=1 make bench-zero-regression.
  • It guards the selector cursor families, which this series must leave untouched.
  • The median regression budget is 5%, and allocation counts must match exactly.
  • Do not run Cargo builds or tests on the same host while the gate measures.
  • Record the verdict from the run's report.json.
  • Verdict: no regression (series head bdcdb644, baseline 1f224d06). All 12 cases match semantically and allocation counts match exactly.
    • With default codegen, list_swap measured +7% cycles at identical instruction and branch counts, reproduced across runs. Topdown counters put the difference in front-end bandwidth: the case's loop is inlined into the gate binary's main, whose start moved from offset 48 to 32 within a 64-byte block because unrelated code changed size. An unused field added to ScoreDirector on the baseline alone reproduced it (+5%).
    • Rebuilt with functions and loops aligned to 64 bytes on both sides (RUSTFLAGS="-C llvm-args=-align-all-functions=6 -C llvm-args=-align-loops=64"), every case's wall-time and cycle median is within ±2% of baseline (list_swap +0.10% wall / +0.33% cycles; worst construction_full +1.06% / +1.95%; k_opt −3.8% wall). The only failing criterion is cache-misses on list_swap and nearby_swap (+11.5%, +10.2% on small absolute counts).
    • Investigating this removed two avoidable copies: replicas now borrow the solve's descriptor instead of copying it, and evaluated candidates lend their tabu signature instead of cloning it.
• Benchmark (maintainer-run): solution quality at equal wall time and evaluation throughput with move_thread_count 1 against Count(4) on the scalar, list and CVRP suites using the existing benchmark infrastructure.
  • That run is the performance verdict; implementers do not build a separate measurement harness.
  • Exporting the generation/evaluation time split from the benchmark adapters is the prerequisite for any scaling claim (D3).

Decisions needed before merge

• D1 — Auto. This PR resolves Auto to sequential for local search, so default behaviour and memory use do not change. Turning it on (for example min(available cores − 1, 4)) is a one-line change in resolve_evaluator_count, made once the benchmark verdict is in.
• D2 — Window size. PARALLEL_EVALUATION_WINDOW = 32 is an internal constant. It sets both speculative waste on early-quitting foragers and per-window overhead. Make it configurable only if the benchmark shows a shape-dependent optimum.
• D3 — Benchmark telemetry. Export generation_time and evaluation_time from the benchmark adapters. Serial generation caps the speedup wherever candidate construction dominates; grouped-scalar compound moves are the likely case.
• D4 — Issue lifecycle. Keep Explore Rayon-based parallel move evaluation without duplicate search algorithms #76 open for the streaming-vs-windowed comparison, or close it when this lands.

Downstream

solverforge-py

The Python-binding work lives in SolverForge/solverforge-py#1 and lands after the release that ships this series.

Portfolio

The cooperative portfolio suggested in #76 can build on Director::fork_replica and EvaluatorPool as a separate scheduler, in its own issue.

[blackopsrepl]

Benchmark: 1 vs 4 evaluators

Setup. The series head is bdcdb644. The three SolverForge adapters from solverforge-bench 0.1.5 were built against this branch, with their stock solver.toml files. The only change is that move_thread_count is set from an environment variable. The benchmark's own runner ran both arms on the same four logical CPUs (taskset -c 6-9) of an i5-12400. There was one run per cell, with the configured seeds. Cost is the benchmark's objective; lower is better. "Gen share" is the fraction of the single-evaluator solve spent generating candidates.

Employee scheduling (INRC-II)

Instance	Budget	Cost 1 → 4	Δ cost	Evaluated 1 → 4	Speculative	Gen share
n060w8 H0	10 s	53,790 → 50,110	−6.8%	25,129 → 30,536 (+22%)	785	66%
n060w8 H0	60 s	19,940 → 16,270	−18.4%	221,885 → 345,554 (+56%)	11,783	32%
n060w8 H2	10 s	52,505 → 49,420	−5.9%	22,513 → 28,885 (+28%)	644	69%
n060w8 H2	60 s	20,030 → 16,045	−19.9%	244,655 → 391,838 (+60%)	13,467	29%
n120w8 H0	10 s	216,295 → 216,295	0.0%	3,836 → 3,836	0	96%
n120w8 H0	60 s	154,445 → 154,570	+0.1%	23,569 → 23,501	0	96%
n120w8 H1	10 s	217,865 → 217,865	0.0%	3,762 → 3,759	0	96%
n120w8 H1	60 s	164,290 → 166,635	+1.4%	20,113 → 19,497	0	96%

Job shop (Taillard)

Instance	Budget	Makespan 1 → 4	Evaluated 1 → 4	Speculative	Gen share
ta18	10 s	1741 → 1741	255,014 → 389,263 (+53%)	1,818	33%
ta18	60 s	1741 → 1741	1,302,830 → 2,113,673 (+62%)	12,704	30%
ta50	10 s	2330 → 2330	68,292 → 107,017 (+57%)	847	35%
ta50	60 s	2330 → 2314	444,288 → 717,501 (+61%)	3,163	39%

CVRP (X)

Instance	Budget	Cost 1 → 4	Evaluated 1 → 4	Speculative	Gen share
X-n101-k25	10 s	28,578 → 28,578	112,839 → 106,935 (−5%)	70,773	2%
X-n101-k25	60 s	28,578 → 28,578	549,818 → 515,242 (−6%)	426,850	2%
X-n106-k14	10 s	27,154 → 27,154	60,728 → 58,878 (−3%)	23,807	<1%
X-n106-k14	60 s	27,154 → 27,154	219,446 → 210,273 (−4%)	144,764	<1%

What this shows

• Where evaluation dominates and steps are long, four evaluators convert directly into quality. Employee scheduling n060 evaluates 56–60% more candidates at 60 s and finishes 18–20% cheaper. Job shop evaluates 53–62% more on both instances. ta18 sits on its plateau at either count; ta50 improves at 60 s. Speculative work stays under 4% of evaluations.
• Serial generation caps the speedup, as expected. On employee scheduling n120, 96% of the single-evaluator solve is spent building candidates, about 2.4 ms each, against roughly 0.1 ms to score one. Parallel evaluation has nothing to win there. That instance needs cheaper candidate generation, not more evaluators.
• A fixed 32-candidate window loses on short steps. The CVRP configuration stops each step after four accepted moves, roughly 23 evaluations per step. Most of every window is therefore discarded: speculative work is 40–45% of all trials, and consumed evaluations fall 3–6%. Generation share is 2% or less, so the work is all evaluation, but it is wasted past the decision point.

Consequences for the open decisions

• D1 (Auto): keep it sequential. It helps or is neutral on two suites, but it costs throughput on CVRP until D2 is resolved.
• D2 (window size): CVRP shows the window has to track the step's expected evaluation count rather than stay fixed at 32. One option is to size the next window from the previous step's evaluations, clamped between the evaluator count and 32. That keeps the trajectory reproducible at a fixed evaluator count, but it gives up the identical trajectory across counts of two or more.
• D3 (generation/evaluation split): answered. Generation dominates on employee scheduling n120 and caps scaling there. Elsewhere, at 60 s, evaluation accounts for 61–98% of the solve.
• Not run yet: a Python model on free-threaded CPython, which depends on the solverforge-py pull request.