fix(evaluators): preserve refinement validation across loky transport

CHANGELOG.md

@@ -10,6 +10,10 @@ format. Stability guarantees for the public surface are documented in the
 
 ### Fixed
 
+- Joblib's `loky` backend can now return local-search refinements with large
+  integer or floating-point candidates without failing to deserialize them.
+  Restored successes retain alignment evidence without serializing candidate
+  equality callables; the coordinator still validates them before assimilation.
 - CSA generation tracing no longer fails when a perturbation schedule has no
   mutation family. Regular-only, initial-only, and combined regular/initial
   schedules can now be traced without changing their proposals or RNG state.

docs/guides/request-local-episodes.md

@@ -72,12 +72,44 @@ result, final_state = study.optimize(
 
 `batch_size=1` is supported but exposes no proposal-level parallelism.
 
-!!! warning "Process-backend limitation in 0.2.0"
+### Use process workers in the development version
 
-    Use the threading backend for built-in local-search episodes that can return
-    `CandidateRefinement`. In 0.2.0, loky cannot deserialize a refinement-bearing
-    request-local success. This limitation concerns episode result transport;
-    ordinary Joblib objective evaluation remains available with loky.
+The development version supports refinement-bearing episodes with `loky` in
+both `problem_transport="per_request"` and `"worker_session"` modes. For example,
+replace the threading study above with:
+
+```python
+study = Study(
+    problem=problem,
+    run_method=optimizer,
+    kernel=StructuredHillClimbKernel(max_steps=8),
+    evaluator=JoblibEvaluator(
+        n_jobs=4,
+        backend="loky",
+        problem_transport="worker_session",
+    ),
+)
+
+result, final_state = study.optimize(
+    max_evaluations=200,
+    batch_size=8,
+    execution_model=SYNC_BATCH_EXECUTION_MODEL,
+)
+```
+
+Omit `problem_transport` to use the default `"per_request"` mode. See
+[Reuse a Problem in Joblib Workers](reuse-a-problem-in-joblib-workers.md) for
+the lifetime and mutability constraints of `"worker_session"`.
+
+!!! warning "0.2.0 requires a workaround"
+
+    The fix above has not yet been released. In `0.2.0`, `loky` can fail to
+    deserialize episode results carrying `CandidateRefinement`, including
+    results with large integer or floating-point candidates. Use
+    `backend="threading"` for these episodes on `0.2.0`. Ordinary Joblib objective
+    evaluation with `loky` is unaffected.
+
+## Limit SciPy objective calls
 
 For SciPy local search, cap objective calls separately from optimizer
 iterations:

docs/reference/evaluator-contracts.md

@@ -57,11 +57,14 @@ execution uses the coordinator-owned kernel path.
 
 ## Supported request-local placement
 
+The table describes the development version. The `0.2.0` limitation is noted
+below.
+
 | Evaluator and execution model | Placement |
 | --- | --- |
 | `SequentialEvaluator`, `sequential` or `sync_batch` | Episode runs inline in the evaluator |
 | `JoblibEvaluator` with threading, `sequential` or `sync_batch` | Episode may run in a Joblib worker |
-| `JoblibEvaluator` with loky, `sequential` or `sync_batch` | 0.2.0 cannot return refinement-bearing episode successes |
+| `JoblibEvaluator` with loky, `sequential` or `sync_batch` | Episode may run in a Joblib process, including refinement-bearing results |
 | `MpiEvaluator`, `sync_batch` | Coordinator fallback |
 | `AsyncJoblibEvaluator`, `exact_async` | Unsupported; study-level exact async requires `DirectKernel` |
 | `stale_async` | Unsupported; the current study path requires `DirectKernel` |
@@ -70,9 +73,17 @@ The Joblib threading backend shares the exact `Problem` instance across
 workers, so the evaluation protocol must be thread-safe. The loky backend
 crosses a process serialization boundary. The problem, objective, kernel,
 candidate, proposal-local context, and returned payload must be serializable.
-In 0.2.0, loky result transport also fails when a request-local success carries
-`CandidateRefinement`; use threading for built-in episodes that may refine a
-candidate.
+
+Both `problem_transport="per_request"` and `"worker_session"` support
+refinement-bearing results in the development version.
+
+!!! warning "0.2.0 process-result limitation"
+
+    The refinement transport fix is not yet released. In `0.2.0`, `loky` can
+    fail to deserialize request-local successes carrying `CandidateRefinement`,
+    including results with large integer or floating-point candidates. Use
+    threading for these episodes on `0.2.0`; ordinary objective evaluation with
+    `loky` is unaffected.
 
 ## Hard-budget accounting
 
@@ -111,6 +122,12 @@ Moving an eligible episode to an evaluator preserves:
 - inner failure summaries and terminal status in `KernelDiagnostics`
 - original proposal identity
 
+After worker execution, the coordinator independently validates request and
+refinement alignment with the problem space's candidate-equality contract,
+then rebinds that predicate before materialization and assimilation. Process
+transport preserves prior alignment evidence but does not replace this check
+or serialize the equality callable with each success.
+
 An objective exception that is successfully recorded is data, not a backend
 failure. An episode may return a successful top-level attempt with failed inner
 trials summarized in diagnostics, or an `EvaluationFailure` when no trial
@@ -121,9 +138,7 @@ remain hard execution failures.
 
 Stochastic built-in episodes use a deterministic proposal-local
 `RandomStateSnapshot` derived by the run method. Worker scheduling and
-completion order do not select the random stream. The coordinator rebinds
-candidate-equality validation after worker execution so process transport does
-not weaken refinement checks.
+completion order do not select the random stream.
 
 Durable checkpointing currently serializes CSA run-method state only. It does
 not serialize reports, evaluator workers, live episode reservations, or async

src/variopt/artifacts/attempts.py

@@ -96,23 +96,23 @@ def candidate(self) -> object:
 
 
 @dataclass(frozen=True, slots=True)
-class _UnvalidatedRefinementCandidate:
+class UnvalidatedRefinementCandidate:
     """Sentinel for refinement pairs that have not been revalidated."""
 
 
-_UNVALIDATED_REFINEMENT_CANDIDATE = _UnvalidatedRefinementCandidate()
-ValidatedRefinementCandidate: TypeAlias = CandidateT | _UnvalidatedRefinementCandidate
+_UNVALIDATED_REFINEMENT_CANDIDATE = UnvalidatedRefinementCandidate()
+ValidatedRefinementCandidate: TypeAlias = CandidateT | UnvalidatedRefinementCandidate
 EvaluationSuccessPickleState: TypeAlias = tuple[
     EvaluationRequest[CandidateT],
     PayloadT_co,
     int,
     CandidateRefinement[CandidateT] | None,
     KernelDiagnostics | None,
     bool,
-    ValidatedRefinementCandidate[CandidateT],
-    ValidatedRefinementCandidate[CandidateT],
-    ValidatedRefinementCandidate[CandidateT],
-    ValidatedRefinementCandidate[CandidateT],
+    EvaluationRequest[CandidateT] | None,
+    CandidateRefinement[CandidateT] | None,
+    EvaluationRequest[CandidateT] | None,
+    CandidateRefinement[CandidateT] | None,
 ]
 
 
@@ -1029,17 +1029,26 @@ def with_kernel_diagnostics(
 
     def _pickle_state(self) -> EvaluationSuccessPickleState[CandidateT, PayloadT_co]:
         """Return pickle state without serializing candidate equality callables."""
+        refinement_is_prevalidated = self._refinement_alignment_is_prevalidated()
+        validated_payload_request: EvaluationRequest[CandidateT] | None = None
+        if _is_request_aligned_payload_in_candidate_domain(
+            self.payload, self.request.candidate
+        ) and self._payload_source_alignment_is_prevalidated(self.payload.request):
+            validated_payload_request = self.payload.request
+
+        # Pickle memoizes the owners, but not every scalar candidate occurrence.
+        # Carry evidence for these owners, not scalar identity across processes.
         return (
             self.request,
             self.payload,
             self.evaluation_count,
             self.refinement,
             self.kernel_diagnostics,
             self._candidate_equal_required,
-            self._validated_request_candidate,
-            self._validated_refined_candidate,
-            self._validated_payload_request_candidate,
-            self._validated_refinement_source_candidate,
+            self.request if refinement_is_prevalidated else None,
+            self.refinement if refinement_is_prevalidated else None,
+            validated_payload_request,
+            self.refinement if validated_payload_request is not None else None,
         )
 
     def _restore_pickle_state(
@@ -1054,40 +1063,50 @@ def _restore_pickle_state(
             refinement,
             kernel_diagnostics,
             candidate_equal_required,
-            validated_request_candidate,
-            validated_refined_candidate,
-            validated_payload_request_candidate,
-            validated_refinement_source_candidate,
+            validated_request,
+            validated_refinement,
+            validated_payload_request,
+            validated_source_refinement,
         ) = state
-        object.__setattr__(self, "__orig_class__", None)
-        object.__setattr__(self, "request", request)
-        object.__setattr__(self, "payload", payload)
-        object.__setattr__(self, "evaluation_count", evaluation_count)
-        object.__setattr__(self, "refinement", refinement)
-        object.__setattr__(self, "kernel_diagnostics", kernel_diagnostics)
-        object.__setattr__(self, "_candidate_equal", None)
-        object.__setattr__(self, "_candidate_equal_required", candidate_equal_required)
-        object.__setattr__(
-            self,
-            "_validated_request_candidate",
-            validated_request_candidate,
+        request_candidate: ValidatedRefinementCandidate[CandidateT] = (
+            _UNVALIDATED_REFINEMENT_CANDIDATE
         )
-        object.__setattr__(
-            self,
-            "_validated_refined_candidate",
-            validated_refined_candidate,
+        refined_candidate: ValidatedRefinementCandidate[CandidateT] = (
+            _UNVALIDATED_REFINEMENT_CANDIDATE
         )
-        object.__setattr__(
-            self,
-            "_validated_payload_request_candidate",
-            validated_payload_request_candidate,
+        payload_request_candidate: ValidatedRefinementCandidate[CandidateT] = (
+            _UNVALIDATED_REFINEMENT_CANDIDATE
         )
-        object.__setattr__(
-            self,
-            "_validated_refinement_source_candidate",
-            validated_refinement_source_candidate,
+        source_candidate: ValidatedRefinementCandidate[CandidateT] = (
+            _UNVALIDATED_REFINEMENT_CANDIDATE
+        )
+        if type(request) is EvaluationRequest and _is_candidate_refinement(refinement):
+            if validated_request is request and validated_refinement is refinement:
+                request_candidate = request.candidate
+                refined_candidate = refinement.refined_candidate
+
+            if (
+                _is_request_aligned_payload_in_candidate_domain(
+                    payload, request.candidate
+                )
+                and validated_payload_request is payload.request
+                and validated_source_refinement is refinement
+            ):
+                payload_request_candidate = payload.request.candidate
+                source_candidate = refinement.source_candidate
+
+        self.__init__(
+            request=request,
+            payload=payload,
+            evaluation_count=evaluation_count,
+            refinement=refinement,
+            kernel_diagnostics=kernel_diagnostics,
+            _candidate_equal_required=candidate_equal_required,
+            _validated_request_candidate=request_candidate,
+            _validated_refined_candidate=refined_candidate,
+            _validated_payload_request_candidate=payload_request_candidate,
+            _validated_refinement_source_candidate=source_candidate,
         )
-        self._validate(candidate_equal=None)
 
 
 def _success_from_scalar_observation(