SPDX-License-Identifier: AGPL-3.0-or-later

Commercial license available

© Concepts 1996–2026 Miroslav Šotek. All rights reserved.

© Code 2020–2026 Miroslav Šotek. All rights reserved.

ORCID: 0009-0009-3560-0851

Contact: www.anulum.li | protoscience@anulum.li

SCPN Phase Orchestrator — Autotune reward API reference

Autotune Reward Evaluation

The reward evaluator is the first safe slice of the reinforcement-learning autotune track. It scores candidate knob policies from replay or simulation metrics without applying control actions directly.

The default reward is coherence improvement minus target deficit, low-coherence risk, actuation energy, unsafe rollout flags, regime churn, positive Lyapunov growth, negative STL robustness, and explicit safety cost. The output is an audit-ready record that can be used by later PPO/SAC or hybrid physics-RL learners.

from scpn_phase_orchestrator.autotune import (
    KnobPolicyCandidate,
    RewardObservation,
    evaluate_knob_policy,
)

candidate = KnobPolicyCandidate(K=0.2, alpha=0.0, zeta=0.05, Psi=0.1)
observation = RewardObservation(
    coherence=0.82,
    previous_coherence=0.74,
    lyapunov_exponent=-0.015,
    stl_robustness=0.08,
    safety_cost=0.01,
)
report = evaluate_knob_policy(candidate, observation)

assert report.to_audit_record()["reward"] == report.reward

Replay-trained or simulation-trained searches can rank multiple candidates without applying any control action:

from scpn_phase_orchestrator.autotune import rank_replay_candidates

ranked = rank_replay_candidates(
    (
        (KnobPolicyCandidate(K=0.15), RewardObservation(coherence=0.72)),
        (KnobPolicyCandidate(K=0.30), RewardObservation(coherence=0.65)),
    ),
    top_k=1,
)

best_report = ranked[0].to_audit_record()

Offline search can generate deterministic coordinate candidates around a seed policy before replay scoring. The candidate surface includes the universal knobs, per-channel weights, and cross-channel coupling gains; the generator is still side-effect free and only emits replay candidates.

from scpn_phase_orchestrator.autotune import (
    OfflinePolicySearchConfig,
    generate_offline_policy_candidates,
)

candidates = generate_offline_policy_candidates(
    KnobPolicyCandidate(
        K=0.2,
        zeta=0.05,
        channel_weights=(1.0, 0.8),
        cross_channel_gains=(0.3, 0.5),
    ),
    OfflinePolicySearchConfig(
        K_step=0.05,
        zeta_step=0.02,
        channel_weight_step=0.1,
        cross_channel_gain_step=0.1,
        max_abs_knob=1.0,
    ),
)

Proposal records apply simple acceptance gates and remain review artefacts:

from scpn_phase_orchestrator.autotune import (
    PolicyProposalConfig,
    SafetyConstraintConfig,
    propose_replay_policy,
)

proposal = propose_replay_policy(
    (
        (
            candidate,
            RewardObservation(
                coherence=0.82,
                lyapunov_exponent=-0.015,
                stl_robustness=0.08,
                safety_cost=0.01,
            ),
        ),
    ),
    proposal_config=PolicyProposalConfig(
        min_coherence=0.75,
        safety_constraints=SafetyConstraintConfig(
            max_lyapunov_exponent=0.0,
            min_stl_robustness=0.0,
            max_safety_cost=0.05,
            require_lyapunov=True,
            require_stl=True,
            require_safety_cost=True,
        ),
    ),
)

audit_record = proposal.to_audit_record()

The safety-constraint gate is intentionally conservative. If a proposal config requires Lyapunov or STL evidence and a replay observation omits it, the candidate is rejected even when its coherence reward is high. This keeps safe-RL integration reviewable: the learner may optimise, but the acceptance record must still carry explicit stability and temporal-logic evidence.

For the higher-level replay-only search wrapper that generates candidates, evaluates them through a caller-supplied replay adapter, and returns a proposal record, see Autotune Replay Policy Search.

Operational overview

This module is the scoring seam between raw simulation outcomes and policy action. Rewards are structured to preserve risk awareness while still allowing optimization experiments.

When used in a staged autonomy lane, teams typically:

• score candidates with replay evidence first,
• enforce explicit safety gates,
• only then promote a candidate into a proposal record.

That order is important because it separates numeric optimisation from safety admissibility. A candidate can look strong on raw coherence and still fail safe policy constraints.

Safe RL readiness

The shape and fields in evaluate_knob_policy are aligned with later RL loop integration:

• Observations carry stability, coherence, and safety fields together.
• Proposal records contain audit-ready traces and gating decisions.
• Rejection reasons are represented in the proposal output, not only in external logs.

This makes the reward module usable as the first hard requirement layer for PPO/SAC or other search learners without changing governance rules later.

Why this function is separated from control execution

evaluate_knob_policy is intentionally scoped to scoring and evidence generation. It computes reward and audit fields so a learner can rank candidates, but it does not decide control actions by itself.

When integrating with RL stacks, keep the same sequence:

• replay or simulation produces observations,
• reward scoring evaluates candidates with coherence, stability, and safety terms,
• proposal gating enforces the hard safety thresholds,
• deployment code consumes only accepted proposals that preserve evidence boundaries.

This keeps model-driven optimisation inside a constrained review envelope rather than directly connecting policy gradients to hardware or external actuators.

reward

Auditable reward scoring for candidate autotune policies.

Classes

KnobPolicyCandidate dataclass

KnobPolicyCandidate(
    K: float | FloatArray = 0.0,
    alpha: float | FloatArray = 0.0,
    zeta: float | FloatArray = 0.0,
    Psi: float | FloatArray = 0.0,
    channel_weights: tuple[float, ...] = (),
    cross_channel_gains: tuple[float, ...] = (),
)

Candidate phase-control knobs proposed by autotune tooling.

RewardObservation dataclass

RewardObservation(
    coherence: float,
    previous_coherence: float | None = None,
    unsafe: bool = False,
    regime_changed: bool = False,
    lyapunov_exponent: float | None = None,
    stl_robustness: float | None = None,
    safety_cost: float = 0.0,
)

Observed rollout metrics used to score one policy candidate.

Methods:

__post_init__

__post_init__() -> None

Validate observation probabilities, flags, and optional safety evidence.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def __post_init__(self) -> None:
    """Validate observation probabilities, flags, and optional safety evidence."""
    _require_probability(self.coherence, "coherence")
    if self.previous_coherence is not None:
        _require_probability(self.previous_coherence, "previous_coherence")
    _require_bool(self.unsafe, "unsafe")
    _require_bool(self.regime_changed, "regime_changed")
    object.__setattr__(
        self,
        "lyapunov_exponent",
        _optional_finite_real(self.lyapunov_exponent, "lyapunov_exponent"),
    )
    object.__setattr__(
        self,
        "stl_robustness",
        _optional_finite_real(self.stl_robustness, "stl_robustness"),
    )
    object.__setattr__(
        self,
        "safety_cost",
        non_negative_real(self.safety_cost, name="safety_cost"),
    )

OfflinePolicySearchConfig dataclass

OfflinePolicySearchConfig(
    K_step: float = 0.05,
    alpha_step: float = 0.05,
    zeta_step: float = 0.05,
    Psi_step: float = 0.05,
    channel_weight_step: float = 0.05,
    cross_channel_gain_step: float = 0.05,
    include_baseline: bool = True,
    max_abs_knob: float | None = None,
)

Deterministic candidate-generation settings for replay searches.

Methods:

__post_init__

__post_init__() -> None

Validate coordinate-search step sizes and clipping bounds.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def __post_init__(self) -> None:
    """Validate coordinate-search step sizes and clipping bounds."""
    _require_bool(self.include_baseline, "include_baseline")
    for label, value in [
        ("K_step", self.K_step),
        ("alpha_step", self.alpha_step),
        ("zeta_step", self.zeta_step),
        ("Psi_step", self.Psi_step),
        ("channel_weight_step", self.channel_weight_step),
        ("cross_channel_gain_step", self.cross_channel_gain_step),
    ]:
        non_negative_real(value, name=label)
    if self.max_abs_knob is not None:
        non_negative_real(self.max_abs_knob, name="max_abs_knob")
        if self.max_abs_knob == 0.0:
            raise ValueError("max_abs_knob must be positive when provided")

RewardConfig dataclass

RewardConfig(
    target_coherence: float = 1.0,
    bad_coherence_threshold: float = 0.35,
    coherence_weight: float = 1.0,
    bad_coherence_penalty: float = 2.0,
    actuation_penalty: float = 0.01,
    churn_penalty: float = 0.1,
    unsafe_penalty: float = 10.0,
    lyapunov_penalty: float = 1.0,
    stl_penalty: float = 1.0,
    safety_cost_penalty: float = 1.0,
    component_order: tuple[str, ...] = (
        "coherence_gain",
        "target_tracking",
        "bad_coherence",
        "actuation",
        "regime_churn",
        "unsafe",
        "lyapunov_stability",
        "stl_robustness",
        "safety_cost",
    ),
)

Weights for auditable coherence-minus-risk autotune reward.

Methods:

__post_init__

__post_init__() -> None

Validate reward weights and component-order policy.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def __post_init__(self) -> None:
    """Validate reward weights and component-order policy."""
    _require_probability(self.target_coherence, "target_coherence")
    _require_probability(self.bad_coherence_threshold, "bad_coherence_threshold")
    for label, value in [
        ("coherence_weight", self.coherence_weight),
        ("bad_coherence_penalty", self.bad_coherence_penalty),
        ("actuation_penalty", self.actuation_penalty),
        ("churn_penalty", self.churn_penalty),
        ("unsafe_penalty", self.unsafe_penalty),
        ("lyapunov_penalty", self.lyapunov_penalty),
        ("stl_penalty", self.stl_penalty),
        ("safety_cost_penalty", self.safety_cost_penalty),
    ]:
        non_negative_real(value, name=label)
    _validate_component_order(self.component_order)

SafetyConstraintConfig dataclass

SafetyConstraintConfig(
    max_lyapunov_exponent: float | None = None,
    min_stl_robustness: float | None = None,
    max_safety_cost: float | None = None,
    require_lyapunov: bool = False,
    require_stl: bool = False,
    require_safety_cost: bool = False,
)

Lyapunov/STL gates for review-only safe-RL proposals.

Methods:

__post_init__

__post_init__() -> None

Validate safety evidence bounds and require-evidence flags.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def __post_init__(self) -> None:
    """Validate safety evidence bounds and require-evidence flags."""
    object.__setattr__(
        self,
        "max_lyapunov_exponent",
        _optional_finite_real(
            self.max_lyapunov_exponent,
            "max_lyapunov_exponent",
        ),
    )
    object.__setattr__(
        self,
        "min_stl_robustness",
        _optional_finite_real(self.min_stl_robustness, "min_stl_robustness"),
    )
    object.__setattr__(
        self,
        "max_safety_cost",
        _optional_non_negative_finite(self.max_safety_cost, "max_safety_cost"),
    )
    _require_bool(self.require_lyapunov, "require_lyapunov")
    _require_bool(self.require_stl, "require_stl")
    _require_bool(self.require_safety_cost, "require_safety_cost")
    if self.require_lyapunov and self.max_lyapunov_exponent is None:
        raise ValueError(
            "require_lyapunov needs max_lyapunov_exponent evidence bound"
        )
    if self.require_stl and self.min_stl_robustness is None:
        raise ValueError("require_stl needs min_stl_robustness evidence bound")
    if self.require_safety_cost and self.max_safety_cost is None:
        raise ValueError("require_safety_cost needs max_safety_cost evidence bound")

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-serialisable safety-gate configuration.

Returns

dict[str, object] A JSON-serialisable safety-gate configuration.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-serialisable safety-gate configuration.

    Returns
    -------
    dict[str, object]
        A JSON-serialisable safety-gate configuration.
    """
    return {
        "max_lyapunov_exponent": self.max_lyapunov_exponent,
        "min_stl_robustness": self.min_stl_robustness,
        "max_safety_cost": self.max_safety_cost,
        "require_lyapunov": self.require_lyapunov,
        "require_stl": self.require_stl,
        "require_safety_cost": self.require_safety_cost,
    }

PolicyProposalConfig dataclass

PolicyProposalConfig(
    min_reward: float = -np.inf,
    min_coherence: float = 0.0,
    max_alternatives: int = 3,
    require_safe: bool = True,
    safety_constraints: SafetyConstraintConfig = SafetyConstraintConfig(),
)

Acceptance gates for replay-trained policy proposals.

Methods:

__post_init__

__post_init__() -> None

Validate policy proposal gates and embedded safety constraints.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def __post_init__(self) -> None:
    """Validate policy proposal gates and embedded safety constraints."""
    if isinstance(self.min_reward, (bool, np.bool_)) or not isinstance(
        self.min_reward, Real
    ):
        raise ValueError("min_reward must be finite or -inf")
    if not np.isfinite(float(self.min_reward)) and self.min_reward != -np.inf:
        raise ValueError("min_reward must be finite or -inf")
    _require_probability(self.min_coherence, "min_coherence")
    if isinstance(self.max_alternatives, (bool, np.bool_)) or not isinstance(
        self.max_alternatives, Integral
    ):
        raise TypeError("max_alternatives must be a non-negative integer")
    if int(self.max_alternatives) < 0:
        raise ValueError("max_alternatives must be non-negative")
    object.__setattr__(self, "max_alternatives", int(self.max_alternatives))
    _require_bool(self.require_safe, "require_safe")
    if not isinstance(self.safety_constraints, SafetyConstraintConfig):
        raise TypeError("safety_constraints must be SafetyConstraintConfig")

AutotuneRewardReport dataclass

AutotuneRewardReport(
    reward: float,
    components: dict[str, float],
    candidate: KnobPolicyCandidate,
    observation: RewardObservation,
    config: RewardConfig,
)

Reward result suitable for policy search and audit logs.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a serialisable reward record.

Returns

dict[str, object] A serialisable reward record.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable reward record.

    Returns
    -------
    dict[str, object]
        A serialisable reward record.
    """
    return {
        "reward": self.reward,
        "components": dict(self.components),
        "candidate": {
            "K": _serialise_array(self.candidate.K),
            "alpha": _serialise_array(self.candidate.alpha),
            "zeta": _serialise_array(self.candidate.zeta),
            "Psi": _serialise_array(self.candidate.Psi),
            "channel_weights": list(self.candidate.channel_weights),
            "cross_channel_gains": list(self.candidate.cross_channel_gains),
        },
        "observation": {
            "coherence": self.observation.coherence,
            "previous_coherence": self.observation.previous_coherence,
            "unsafe": self.observation.unsafe,
            "regime_changed": self.observation.regime_changed,
            "lyapunov_exponent": self.observation.lyapunov_exponent,
            "stl_robustness": self.observation.stl_robustness,
            "safety_cost": self.observation.safety_cost,
        },
        "config": {
            "target_coherence": self.config.target_coherence,
            "bad_coherence_threshold": self.config.bad_coherence_threshold,
            "coherence_weight": self.config.coherence_weight,
            "bad_coherence_penalty": self.config.bad_coherence_penalty,
            "actuation_penalty": self.config.actuation_penalty,
            "churn_penalty": self.config.churn_penalty,
            "unsafe_penalty": self.config.unsafe_penalty,
            "lyapunov_penalty": self.config.lyapunov_penalty,
            "stl_penalty": self.config.stl_penalty,
            "safety_cost_penalty": self.config.safety_cost_penalty,
            "component_order": list(self.config.component_order),
        },
    }

AutotunePolicyProposal dataclass

AutotunePolicyProposal(
    accepted: bool,
    selected: AutotuneRewardReport | None,
    alternatives: tuple[AutotuneRewardReport, ...],
    reasons: tuple[str, ...],
    config: PolicyProposalConfig,
)

Replay-trained policy proposal record for human or CI review.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a serialisable policy proposal record.

Returns

dict[str, object] A serialisable policy proposal record.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def to_audit_record(self) -> dict[str, object]:
    """Return a serialisable policy proposal record.

    Returns
    -------
    dict[str, object]
        A serialisable policy proposal record.
    """
    return {
        "accepted": self.accepted,
        "selected": (
            self.selected.to_audit_record() if self.selected is not None else None
        ),
        "alternatives": [
            alternative.to_audit_record() for alternative in self.alternatives
        ],
        "reasons": list(self.reasons),
        "config": {
            "min_reward": (
                None
                if self.config.min_reward == -np.inf
                else self.config.min_reward
            ),
            "min_coherence": self.config.min_coherence,
            "max_alternatives": self.config.max_alternatives,
            "require_safe": self.config.require_safe,
            "safety_constraints": self.config.safety_constraints.to_audit_record(),
        },
    }

Functions:

evaluate_knob_policy

evaluate_knob_policy(
    candidate: KnobPolicyCandidate,
    observation: RewardObservation,
    config: RewardConfig | None = None,
) -> AutotuneRewardReport

Score a candidate policy from coherence and safety metrics.

The reward is intentionally model-free and side-effect free. Training systems can use it to rank replay candidates before any future policy learner is allowed to propose production control actions.

Parameters

candidate : KnobPolicyCandidate The candidate configuration. observation : RewardObservation The observation record. config : RewardConfig | None The configuration object.

Returns

AutotuneRewardReport A candidate policy from coherence and safety metrics.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def evaluate_knob_policy(
    candidate: KnobPolicyCandidate,
    observation: RewardObservation,
    config: RewardConfig | None = None,
) -> AutotuneRewardReport:
    """Score a candidate policy from coherence and safety metrics.

    The reward is intentionally model-free and side-effect free. Training
    systems can use it to rank replay candidates before any future policy
    learner is allowed to propose production control actions.

    Parameters
    ----------
    candidate : KnobPolicyCandidate
        The candidate configuration.
    observation : RewardObservation
        The observation record.
    config : RewardConfig | None
        The configuration object.

    Returns
    -------
    AutotuneRewardReport
        A candidate policy from coherence and safety metrics.
    """
    active_config = config or RewardConfig()
    _validate_candidate(candidate)
    previous = (
        active_config.bad_coherence_threshold
        if observation.previous_coherence is None
        else observation.previous_coherence
    )
    coherence_gain = active_config.coherence_weight * (observation.coherence - previous)
    target_tracking = -active_config.coherence_weight * max(
        0.0, active_config.target_coherence - observation.coherence
    )
    bad_deficit = max(
        0.0, active_config.bad_coherence_threshold - observation.coherence
    )
    bad_coherence = -active_config.bad_coherence_penalty * bad_deficit
    actuation = -active_config.actuation_penalty * _actuation_energy(candidate)
    regime_churn = -active_config.churn_penalty if observation.regime_changed else 0.0
    unsafe = -active_config.unsafe_penalty if observation.unsafe else 0.0
    lyapunov_stability = 0.0
    if observation.lyapunov_exponent is not None:
        lyapunov_stability = -active_config.lyapunov_penalty * max(
            0.0,
            observation.lyapunov_exponent,
        )
    stl_robustness = 0.0
    if observation.stl_robustness is not None:
        stl_robustness = active_config.stl_penalty * min(
            0.0,
            observation.stl_robustness,
        )
    safety_cost = -active_config.safety_cost_penalty * observation.safety_cost

    components = {
        "coherence_gain": coherence_gain,
        "target_tracking": target_tracking,
        "bad_coherence": bad_coherence,
        "actuation": actuation,
        "regime_churn": regime_churn,
        "unsafe": unsafe,
        "lyapunov_stability": lyapunov_stability,
        "stl_robustness": stl_robustness,
        "safety_cost": safety_cost,
    }
    reward = float(sum(components[name] for name in active_config.component_order))
    return AutotuneRewardReport(
        reward=reward,
        components=components,
        candidate=candidate,
        observation=observation,
        config=active_config,
    )

rank_replay_candidates

rank_replay_candidates(
    replay_candidates: Sequence[
        tuple[KnobPolicyCandidate, RewardObservation]
    ],
    config: RewardConfig | None = None,
    *,
    top_k: int | None = None,
    require_safe: bool = True,
) -> tuple[AutotuneRewardReport, ...]

Rank replay-evaluated policy candidates by reward.

This helper is the non-actuating bridge between reward scoring and future policy learners. It consumes replay or simulation observations, filters unsafe rollouts by default, and returns audit-ready reports sorted from highest to lowest reward.

Parameters

replay_candidates : Sequence[tuple[KnobPolicyCandidate, RewardObservation]] Replay candidate proposals. config : RewardConfig | None The configuration object. top_k : int | None Number of top items to retain. require_safe : bool Whether to require publication-safe output.

Returns

tuple[AutotuneRewardReport, ...] Rank replay-evaluated policy candidates by reward.

Raises

ValueError If the inputs are invalid or inconsistent. TypeError If an argument has the wrong type.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def rank_replay_candidates(
    replay_candidates: Sequence[tuple[KnobPolicyCandidate, RewardObservation]],
    config: RewardConfig | None = None,
    *,
    top_k: int | None = None,
    require_safe: bool = True,
) -> tuple[AutotuneRewardReport, ...]:
    """Rank replay-evaluated policy candidates by reward.

    This helper is the non-actuating bridge between reward scoring and future
    policy learners. It consumes replay or simulation observations, filters
    unsafe rollouts by default, and returns audit-ready reports sorted from
    highest to lowest reward.

    Parameters
    ----------
    replay_candidates : Sequence[tuple[KnobPolicyCandidate, RewardObservation]]
        Replay candidate proposals.
    config : RewardConfig | None
        The configuration object.
    top_k : int | None
        Number of top items to retain.
    require_safe : bool
        Whether to require publication-safe output.

    Returns
    -------
    tuple[AutotuneRewardReport, ...]
        Rank replay-evaluated policy candidates by reward.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    TypeError
        If an argument has the wrong type.
    """
    if not replay_candidates:
        raise ValueError("replay candidate ranking requires at least one candidate")
    if top_k is not None:
        if isinstance(top_k, (bool, np.bool_)) or not isinstance(top_k, Integral):
            raise TypeError("top_k must be a positive integer when provided")
        top_k = int(top_k)
    if top_k is not None and top_k < 1:
        raise ValueError("top_k must be positive when provided")
    _require_bool(require_safe, "require_safe")

    reports = [
        evaluate_knob_policy(candidate, observation, config)
        for candidate, observation in replay_candidates
        if not require_safe or not observation.unsafe
    ]
    if not reports:
        raise ValueError("no safe replay candidates remain after filtering")

    ranked = sorted(
        reports,
        key=lambda report: (
            report.reward,
            report.components["coherence_gain"],
            report.components["actuation"],
        ),
        reverse=True,
    )
    if top_k is None:
        return tuple(ranked)
    return tuple(ranked[:top_k])

propose_replay_policy

propose_replay_policy(
    replay_candidates: Sequence[
        tuple[KnobPolicyCandidate, RewardObservation]
    ],
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> AutotunePolicyProposal

Build a reviewable policy proposal from replay-ranked candidates.

The proposal is an audit artefact only. A caller still has to pass it through domain-specific policy review before any candidate is deployed.

Parameters

replay_candidates : Sequence[tuple[KnobPolicyCandidate, RewardObservation]] Replay candidate proposals. reward_config : RewardConfig | None The reward configuration. proposal_config : PolicyProposalConfig | None The proposal configuration.

Returns

AutotunePolicyProposal A reviewable policy proposal from replay-ranked candidates.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def propose_replay_policy(
    replay_candidates: Sequence[tuple[KnobPolicyCandidate, RewardObservation]],
    reward_config: RewardConfig | None = None,
    proposal_config: PolicyProposalConfig | None = None,
) -> AutotunePolicyProposal:
    """Build a reviewable policy proposal from replay-ranked candidates.

    The proposal is an audit artefact only. A caller still has to pass it
    through domain-specific policy review before any candidate is deployed.

    Parameters
    ----------
    replay_candidates : Sequence[tuple[KnobPolicyCandidate, RewardObservation]]
        Replay candidate proposals.
    reward_config : RewardConfig | None
        The reward configuration.
    proposal_config : PolicyProposalConfig | None
        The proposal configuration.

    Returns
    -------
    AutotunePolicyProposal
        A reviewable policy proposal from replay-ranked candidates.
    """
    active_config = proposal_config or PolicyProposalConfig()
    ranked = rank_replay_candidates(
        replay_candidates,
        reward_config,
        require_safe=active_config.require_safe,
    )
    eligible = tuple(
        report
        for report in ranked
        if not _safety_constraint_reasons(
            report.observation,
            active_config.safety_constraints,
        )
    )
    if not eligible:
        return AutotunePolicyProposal(
            accepted=False,
            selected=None,
            alternatives=tuple(ranked[: active_config.max_alternatives]),
            reasons=("no replay candidate satisfies Lyapunov/STL safety constraints",),
            config=active_config,
        )
    selected = eligible[0]
    reasons: list[str] = []
    if selected.reward < active_config.min_reward:
        reasons.append(
            f"selected reward {selected.reward:.6g} below minimum "
            f"{active_config.min_reward:.6g}"
        )
    if selected.observation.coherence < active_config.min_coherence:
        reasons.append(
            f"selected coherence {selected.observation.coherence:.6g} below "
            f"minimum {active_config.min_coherence:.6g}"
        )
    if selected.observation.unsafe:
        reasons.append("selected rollout is marked unsafe")
    reasons.extend(
        _safety_constraint_reasons(
            selected.observation,
            active_config.safety_constraints,
        )
    )

    accepted = not reasons
    alternatives = tuple(eligible[1 : 1 + active_config.max_alternatives])
    return AutotunePolicyProposal(
        accepted=accepted,
        selected=selected if accepted else None,
        alternatives=alternatives,
        reasons=tuple(reasons),
        config=active_config,
    )

generate_offline_policy_candidates

generate_offline_policy_candidates(
    seed: KnobPolicyCandidate,
    config: OfflinePolicySearchConfig | None = None,
) -> tuple[KnobPolicyCandidate, ...]

Generate deterministic replay-search candidates around a seed policy.

The generator performs a bounded coordinate search over the universal knobs, channel weights, and cross-channel coupling gains. It does not inspect plant state and it does not apply actions; callers must evaluate the returned candidates through replay or simulation before ranking them.

Parameters

seed : KnobPolicyCandidate Seed for the deterministic RNG. config : OfflinePolicySearchConfig | None The configuration object.

Returns

tuple[KnobPolicyCandidate, ...] Deterministic replay-search candidates around a seed policy.

Source code in src/scpn_phase_orchestrator/autotune/reward.py

def generate_offline_policy_candidates(
    seed: KnobPolicyCandidate,
    config: OfflinePolicySearchConfig | None = None,
) -> tuple[KnobPolicyCandidate, ...]:
    """Generate deterministic replay-search candidates around a seed policy.

    The generator performs a bounded coordinate search over the universal knobs,
    channel weights, and cross-channel coupling gains. It does not inspect plant
    state and it does not apply actions; callers must evaluate the returned
    candidates through replay or simulation before ranking them.

    Parameters
    ----------
    seed : KnobPolicyCandidate
        Seed for the deterministic RNG.
    config : OfflinePolicySearchConfig | None
        The configuration object.

    Returns
    -------
    tuple[KnobPolicyCandidate, ...]
        Deterministic replay-search candidates around a seed policy.
    """
    active_config = config or OfflinePolicySearchConfig()
    _validate_candidate(seed)

    candidates: list[KnobPolicyCandidate] = []
    if active_config.include_baseline:
        candidates.append(seed)

    for knob, step in [
        ("K", active_config.K_step),
        ("alpha", active_config.alpha_step),
        ("zeta", active_config.zeta_step),
        ("Psi", active_config.Psi_step),
    ]:
        if step > 0.0:
            candidates.extend(
                _mutate_knob(seed, knob, delta, active_config.max_abs_knob)
                for delta in (-step, step)
            )

    if active_config.channel_weight_step > 0.0:
        for index in range(len(seed.channel_weights)):
            for delta in (
                -active_config.channel_weight_step,
                active_config.channel_weight_step,
            ):
                candidates.append(
                    _mutate_channel_weight(
                        seed,
                        index,
                        delta,
                        active_config.max_abs_knob,
                    )
                )

    if active_config.cross_channel_gain_step > 0.0:
        for index in range(len(seed.cross_channel_gains)):
            for delta in (
                -active_config.cross_channel_gain_step,
                active_config.cross_channel_gain_step,
            ):
                candidates.append(
                    _mutate_cross_channel_gain(
                        seed,
                        index,
                        delta,
                        active_config.max_abs_knob,
                    )
                )

    return _deduplicate_candidates(candidates)