perf(csa): coalesce immutable proposal issuance

CHANGELOG.md

@@ -8,6 +8,19 @@ format. Stability guarantees for the public surface are documented in the
 
 ## [Unreleased]
 
+### Breaking
+
+- Replaced `CSAOptimizer.propose_candidate(state)` with `emit_proposal(state)`.
+  Overrides now return `(proposal, tracks_generation, planned_provenance, state)`:
+  the first item is an issued `Proposal`, not a raw candidate. The returned state
+  must already contain its allocated ID and pending registration, plus queue
+  advancement and generation tracking for generated children. Leave provenance
+  binding to `ask()`, which still registers it at the end of the batch.
+  No compatibility shim is retained: the old hook exposed a dequeued but unissued
+  state and required separate ID-allocation and registration transitions. The new
+  hook makes issuance one operation. The `ask()`/`tell()` and checkpoint formats
+  are unchanged.
+
 ### Fixed
 
 - Joblib's `loky` backend can now return local-search refinements with large

src/variopt/algorithms/population/csa/engine/__init__.py

@@ -4,7 +4,6 @@
     CSAAskPlan,
     CSAMaterializedGeneration,
     commit_materialized_generation,
-    dequeue_generation_candidate,
     materialize_generation,
     plan_next_ask,
 )
@@ -25,7 +24,6 @@
     "apply_tell",
     "begin_stage_transition",
     "commit_materialized_generation",
-    "dequeue_generation_candidate",
     "materialize_generation",
     "plan_next_ask",
 ]

src/variopt/algorithms/population/csa/engine/ask.py

@@ -10,6 +10,7 @@
 from .....distance import require_valid_distance
 from .....diversity import DiversityMetric
 from .....operators import VariationOperator
+from .....randomness import RandomStateSnapshot
 from .....spaces import SearchSpace
 from .....spaces.projections import compile_homogeneous_numeric_subspace
 from .....spaces.structured import require_space_candidate_value
@@ -640,49 +641,39 @@ def distance_between_entry_indices(left_index: int, right_index: int) -> float:
 def commit_materialized_generation(
     engine_state: CSAEngineState[CandidateT],
     materialized_generation: CSAMaterializedGeneration[CandidateT],
-) -> tuple[GeneratedCandidate[CandidateT], CSAEngineState[CandidateT]]:
-    """Commit one materialized child pool and dequeue its first candidate.
+    *,
+    random_state: RandomStateSnapshot,
+) -> CSAEngineState[CandidateT]:
+    """Commit a child pool and its RNG snapshot before issuing any proposals.
 
     Parameters
     ----------
     engine_state : CSAEngineState[CandidateT]
         Current CSA engine state.
     materialized_generation : CSAMaterializedGeneration[CandidateT]
         Materialized child pool ready for commit.
+    random_state : RandomStateSnapshot
+        RNG snapshot after materializing the child pool.
 
     Returns
     -------
-    tuple[GeneratedCandidate[CandidateT], CSAEngineState[CandidateT]]
-        First generated candidate together with the committed engine state.
+    CSAEngineState[CandidateT]
+        State with selection, generation queue, trace, and RNG committed.
+        Proposal IDs and pending registries are unchanged until issuance.
+
+    Raises
+    ------
+    RuntimeError
+        If another child pool is still active.
+    ValueError
+        If the materialized queue is empty.
     """
-    next_engine_state = replace(
+    return replace(
         engine_state,
+        random_state=random_state,
         selection_state=materialized_generation.selection_state,
         generation_state=engine_state.generation_state.begin(
             materialized_generation.generation_queue,
         ),
         trace_state=materialized_generation.trace_state,
     )
-    return dequeue_generation_candidate(next_engine_state)
-
-
-def dequeue_generation_candidate(
-    engine_state: CSAEngineState[CandidateT],
-) -> tuple[GeneratedCandidate[CandidateT], CSAEngineState[CandidateT]]:
-    """Purely dequeue one already-generated candidate from the active pool.
-
-    Parameters
-    ----------
-    engine_state : CSAEngineState[CandidateT]
-        Current CSA engine state with an active generation queue.
-
-    Returns
-    -------
-    tuple[GeneratedCandidate[CandidateT], CSAEngineState[CandidateT]]
-        Next generated candidate and the updated engine state.
-    """
-    candidate, next_generation_state = engine_state.generation_state.dequeue_candidate()
-    return candidate, replace(
-        engine_state,
-        generation_state=next_generation_state,
-    )

src/variopt/algorithms/population/csa/engine/state.py

@@ -24,6 +24,7 @@
 from ..banking.growth.policy import CSABankGrowthPolicy
 from ..generation.proposal.policy import CSAProposalPolicy
 from ..generation.proposal.state.aggregate import CSAProposalState
+from ..generation.proposal.state.attribution import PlannedProposalProvenance
 from ..generation.state import GenerationRuntimeState
 from ..progression.state import CSAProgressionState
 from ..scoring.acceptance import CSAAcceptancePolicy
@@ -363,23 +364,40 @@ def from_dict(
             proposal_index=proposal_index,
         )
 
-    def allocate_proposal_id(self, *, prefix: str = "csa-") -> tuple[str, Self]:
-        """Return one new proposal id together with the advanced engine state.
+    def issue_sampled_proposal(
+        self,
+        candidate: CandidateT,
+        *,
+        random_state: RandomStateSnapshot,
+    ) -> tuple[Proposal[CandidateT], Self]:
+        """Issue an initial-bank or refresh sample with its advanced RNG snapshot.
 
         Parameters
         ----------
-        prefix : str, default=\"csa-\"
-            Prefix used while formatting the proposal id.
+        candidate : CandidateT
+            Sample already validated by the optimizer's search space.
+        random_state : RandomStateSnapshot
+            RNG snapshot after sampling this candidate.
 
         Returns
         -------
-        tuple[str, Self]
-            Allocated proposal id and engine state with the counter advanced.
+        tuple[Proposal[CandidateT], Self]
+            Proposal and state with its ID allocated, pending entry registered,
+            and RNG snapshot replaced. Generation tracking is unchanged.
+
+        Raises
+        ------
+        ValueError
+            If the next proposal ID is already pending.
         """
-        proposal_id = f"{prefix}{self.proposal_index}"
-        return proposal_id, replace(
+        proposal = Proposal(
+            candidate=candidate, proposal_id=f"csa-{self.proposal_index}"
+        )
+        return proposal, replace(
             self,
             proposal_index=self.proposal_index + 1,
+            pending_proposals=self.pending_proposals.add(proposal),
+            random_state=random_state,
         )
 
     def replace_random_state(self, random_state: RandomStateSnapshot) -> Self:
@@ -397,40 +415,38 @@ def replace_random_state(self, random_state: RandomStateSnapshot) -> Self:
         """
         return replace(self, random_state=random_state)
 
-    def issue_proposal(
+    def issue_generation_proposal(
         self,
-        proposal: Proposal[CandidateT],
-        *,
-        tracks_generation: bool,
-    ) -> Self:
-        """Return an engine state that records one issued proposal.
-
-        Parameters
-        ----------
-        proposal : Proposal[CandidateT]
-            Proposal being issued to the evaluator boundary.
-        tracks_generation : bool
-            Whether the proposal should also be registered in generation
-            runtime state.
+    ) -> tuple[Proposal[CandidateT], PlannedProposalProvenance | None, Self]:
+        """Issue the next queued child without an intermediate untracked state.
 
         Returns
         -------
-        Self
-            Engine state with the proposal added to pending proposals and, when
-            requested, generation tracking.
-        """
-        next_generation_state = self.generation_state
-        if tracks_generation:
-            proposal_id = proposal.proposal_id
-            assert proposal_id is not None
-            next_generation_state = self.generation_state.register_proposal(proposal_id)
-
-        next_pending_proposals = self.pending_proposals.add(proposal)
+        tuple[Proposal[CandidateT], PlannedProposalProvenance | None, Self]
+            Proposal, planned provenance, and state with the queue advanced,
+            ID allocated, and both pending registries updated. The optimizer
+            binds provenance after assembling the whole ask batch.
 
-        return replace(
-            self,
-            pending_proposals=next_pending_proposals,
-            generation_state=next_generation_state,
+        Raises
+        ------
+        RuntimeError
+            If the generation queue is empty.
+        ValueError
+            If the next proposal ID is already pending.
+        """
+        proposal_id = f"csa-{self.proposal_index}"
+        generated, next_generation_state = self.generation_state.issue_next(proposal_id)
+        proposal = Proposal(candidate=generated.candidate, proposal_id=proposal_id)
+
+        return (
+            proposal,
+            generated.planned_attribution,
+            replace(
+                self,
+                proposal_index=self.proposal_index + 1,
+                pending_proposals=self.pending_proposals.add(proposal),
+                generation_state=next_generation_state,
+            ),
         )
 
     def consume_pending_proposals(self, proposal_ids: AbstractSet[str]) -> Self:

src/variopt/algorithms/population/csa/generation/state.py

@@ -195,23 +195,10 @@ def begin(self, queue: GenerationQueue[CandidateT]) -> Self:
 
         return type(self)(queue=queue)
 
-    def dequeue_candidate(self) -> tuple[GeneratedCandidate[CandidateT], Self]:
-        """Return the next queued candidate and the updated runtime.
-
-        Returns
-        -------
-        tuple[GeneratedCandidate[CandidateT], Self]
-            Head candidate and runtime state with the remaining queue.
-        """
-        candidate, next_queue = self.queue.dequeue()
-        return candidate, type(self)(
-            queue=next_queue,
-            pending_proposal_ids=self.pending_proposal_ids,
-            buffered_evaluations=self.buffered_evaluations,
-        )
-
-    def register_proposal(self, proposal_id: str) -> Self:
-        """Return a runtime that tracks an issued proposal from the queue.
+    def issue_next(
+        self, proposal_id: str
+    ) -> tuple[GeneratedCandidate[CandidateT], Self]:
+        """Dequeue one child and track its issued proposal in the same transition.
 
         Parameters
         ----------
@@ -220,11 +207,23 @@ def register_proposal(self, proposal_id: str) -> Self:
 
         Returns
         -------
-        Self
-            Runtime state with ``proposal_id`` added to the pending set.
+        tuple[GeneratedCandidate[CandidateT], Self]
+            Queued child and runtime with its queue advanced and proposal tracked.
+
+        Raises
+        ------
+        RuntimeError
+            If the queue is empty.
+        ValueError
+            If ``proposal_id`` is already pending in this generation.
         """
-        return type(self)(
-            queue=self.queue,
+        if proposal_id in self.pending_proposal_ids:
+            msg = "generation proposal id is already pending"
+            raise ValueError(msg)
+
+        candidate, next_queue = self.queue.dequeue()
+        return candidate, type(self)(
+            queue=next_queue,
             pending_proposal_ids=self.pending_proposal_ids | {proposal_id},
             buffered_evaluations=self.buffered_evaluations,
         )