Control Systems

SPO adds a supervision layer over coupled-oscillator dynamics that emits review-only control proposals; TVB, neurolib, Brian2, and NEST are simulate-and-observe libraries. SPO does not close a control loop on hardware.

Model-Predictive Controller (MPC)

Predicts R trajectory 10 steps ahead using the Ott-Antonsen mean-field reduction as a fast forward model. Acts BEFORE degradation, not after. Detects divergence and reverts to reactive control as fallback.

from scpn_phase_orchestrator.supervisor.predictive import PredictiveSupervisor

supervisor = PredictiveSupervisor(engine, horizon=10)
# supervisor.step() predicts future R, triggers actions preemptively

predictive

Predictive and free-energy supervisor diagnostics for bounded action proposals.

The module provides Ott-Antonsen horizon prediction, variational free-energy assessment, and hierarchy-level FEP assessments over validated phase/frequency state. Predictive supervisors emit conservative ControlAction proposals for degradation, critical forecasts, hard boundaries, or high surprise. They do not apply actuation or mutate caller-owned phase/coupling arrays.

Classes

Prediction dataclass

Prediction(
    R_predicted: list[float],
    will_degrade: bool,
    will_critical: bool,
    steps_to_degradation: int,
)

Forward model output: predicted R trajectory and degradation flags.

FEPPredictionAssessment dataclass

FEPPredictionAssessment(
    free_energy: float,
    complexity: float,
    mean_abs_error: float,
    precision_mean: float,
    precision_spread: float,
    observed_R: float,
    observed_psi: float,
    predicted_R: float,
    target_R: float,
    surprise: float,
)

One-step variational free-energy assessment for supervisor control.

Attributes

above_target property

above_target: bool

Return True when observed coherence exceeds the target.

Returns

bool Return True when observed coherence exceeds the target.

Methods:

to_audit_record

to_audit_record() -> dict[str, float]

Return a serialisable audit payload.

Returns

dict[str, float] Return a serialisable audit payload.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def to_audit_record(self) -> dict[str, float]:
    """Return a serialisable audit payload.

    Returns
    -------
    dict[str, float]
        Return a serialisable audit payload.
    """
    return {
        "free_energy": self.free_energy,
        "complexity": self.complexity,
        "mean_abs_error": self.mean_abs_error,
        "precision_mean": self.precision_mean,
        "precision_spread": self.precision_spread,
        "observed_R": self.observed_R,
        "observed_psi": self.observed_psi,
        "predicted_R": self.predicted_R,
        "target_R": self.target_R,
        "surprise": self.surprise,
    }

FEPHierarchyChildAssessment dataclass

FEPHierarchyChildAssessment(
    name: str,
    assessment: FEPPredictionAssessment,
    actions: tuple[ControlAction, ...],
)

Assessment for one child node in a hierarchical FEP supervisor.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe child hierarchy audit record.

Returns

dict[str, object] Return a JSON-safe child hierarchy audit record.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe child hierarchy audit record.

    Returns
    -------
    dict[str, object]
        Return a JSON-safe child hierarchy audit record.
    """
    return {
        "name": self.name,
        "assessment": self.assessment.to_audit_record(),
        "actions": [_action_record(action) for action in self.actions],
    }

FEPHierarchyAssessment dataclass

FEPHierarchyAssessment(
    hierarchy: str,
    children: tuple[FEPHierarchyChildAssessment, ...],
    parent_assessment: FEPPredictionAssessment,
    parent_actions: tuple[ControlAction, ...],
    child_R_values: tuple[float, ...],
    parent_phase_encoding: tuple[float, ...],
)

Audit-ready child-to-parent FEP hierarchy assessment.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe hierarchy assessment payload.

Returns

dict[str, object] Return a JSON-safe hierarchy assessment payload.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe hierarchy assessment payload.

    Returns
    -------
    dict[str, object]
        Return a JSON-safe hierarchy assessment payload.
    """
    return {
        "hierarchy": self.hierarchy,
        "children": [child.to_audit_record() for child in self.children],
        "parent": {
            "assessment": self.parent_assessment.to_audit_record(),
            "actions": [_action_record(action) for action in self.parent_actions],
        },
        "child_R_values": list(self.child_R_values),
        "parent_phase_encoding": list(self.parent_phase_encoding),
    }

PredictiveSupervisor

PredictiveSupervisor(
    n_oscillators: int,
    dt: float,
    horizon: int = 10,
    divergence_threshold: float = 0.3,
)

Model-predictive supervisor using Ott-Antonsen forward model.

Predicts R trajectory horizon steps ahead. Acts preemptively when predicted R crosses thresholds, instead of waiting for actual degradation. Falls back to reactive supervision if OA prediction diverges.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def __init__(
    self,
    n_oscillators: int,
    dt: float,
    horizon: int = 10,
    divergence_threshold: float = 0.3,
):
    self._n = _require_positive_int(n_oscillators, "n_oscillators")
    self._dt = _require_positive_real(dt, "dt")
    self._horizon = _require_positive_int(horizon, "horizon")
    self._divergence_threshold = non_negative_real(
        divergence_threshold,
        name="divergence_threshold",
    )

Methods:

predict

predict(
    phases: FloatArray,
    omegas: FloatArray,
    knm: FloatArray,
    alpha: FloatArray,
) -> Prediction

Predict R trajectory using OA reduction as fast forward model.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). omegas : FloatArray Natural frequencies in rad/s, shape (N,). knm : FloatArray Coupling matrix K_nm, shape (N, N). alpha : FloatArray Phase-lag matrix in radians, shape (N, N), or None for no lag.

Returns

Prediction The predicted R trajectory.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def predict(
    self,
    phases: FloatArray,
    omegas: FloatArray,
    knm: FloatArray,
    alpha: FloatArray,
) -> Prediction:
    """Predict R trajectory using OA reduction as fast forward model.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : FloatArray
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : FloatArray
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.

    Returns
    -------
    Prediction
        The predicted ``R`` trajectory.
    """
    phases, omegas, knm, alpha = _validate_predictive_inputs(
        phases,
        omegas,
        knm,
        alpha,
        self._n,
    )
    R_current, psi = compute_order_parameter(phases)

    # Fit Lorentzian to omegas for OA
    omega_0 = float(np.median(omegas))
    q75, q25 = np.percentile(omegas, [75, 25])
    delta = max((q75 - q25) / 2.0, 0.01)
    K_eff = float(np.mean(knm[knm > 0])) if np.any(knm > 0) else 0.0

    oa = OttAntonsenReduction(omega_0, delta, K_eff, dt=self._dt)
    z0 = complex(R_current * np.cos(psi), R_current * np.sin(psi))

    trajectory = [R_current]
    z = z0
    for _ in range(self._horizon):
        z = oa.step(z)
        trajectory.append(abs(z))

    # Check for divergence (OA prediction unreliable)
    if abs(trajectory[-1] - R_current) > self._divergence_threshold:
        trajectory = [R_current] * (self._horizon + 1)

    will_degrade = any(r < _R_DEGRADED for r in trajectory)
    will_critical = any(r < _R_CRITICAL for r in trajectory)
    steps_to_deg = self._horizon
    for i, r in enumerate(trajectory):
        if r < _R_DEGRADED:
            steps_to_deg = i
            break

    return Prediction(
        R_predicted=trajectory,
        will_degrade=will_degrade,
        will_critical=will_critical,
        steps_to_degradation=steps_to_deg,
    )

decide

decide(
    phases: FloatArray,
    omegas: FloatArray,
    knm: FloatArray,
    alpha: FloatArray,
    upde_state: UPDEState,
    boundary_state: BoundaryState,
) -> list[ControlAction]

Predictive control: act before degradation, not after.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). omegas : FloatArray Natural frequencies in rad/s, shape (N,). knm : FloatArray Coupling matrix K_nm, shape (N, N). alpha : FloatArray Phase-lag matrix in radians, shape (N, N), or None for no lag. upde_state : UPDEState The current UPDE state. boundary_state : BoundaryState The current boundary-observer state.

Returns

list[ControlAction] The predictive control actions for the current state.

Raises

ValueError If the state inputs are invalid.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def decide(
    self,
    phases: FloatArray,
    omegas: FloatArray,
    knm: FloatArray,
    alpha: FloatArray,
    upde_state: UPDEState,
    boundary_state: BoundaryState,
) -> list[ControlAction]:
    """Predictive control: act before degradation, not after.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : FloatArray
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : FloatArray
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    upde_state : UPDEState
        The current UPDE state.
    boundary_state : BoundaryState
        The current boundary-observer state.

    Returns
    -------
    list[ControlAction]
        The predictive control actions for the current state.

    Raises
    ------
    ValueError
        If the state inputs are invalid.
    """
    if not isinstance(upde_state, UPDEState):
        raise ValueError(f"upde_state must be a UPDEState, got {upde_state!r}")
    if not isinstance(boundary_state, BoundaryState):
        raise ValueError(
            f"boundary_state must be a BoundaryState, got {boundary_state!r}"
        )
    if boundary_state.hard_violations:
        return [
            ControlAction(
                knob="zeta",
                scope="global",
                value=0.1,
                ttl_s=5.0,
                justification="hard boundary violation",
            )
        ]

    pred = self.predict(phases, omegas, knm, alpha)

    if pred.will_critical:
        return [
            ControlAction(
                knob="K",
                scope="global",
                value=_K_BOOST * 2,
                ttl_s=10.0,
                justification=(
                    f"MPC: R predicted to hit CRITICAL "
                    f"in {pred.steps_to_degradation} steps"
                ),
            )
        ]

    if pred.will_degrade and pred.steps_to_degradation < self._horizon // 2:
        return [
            ControlAction(
                knob="K",
                scope="global",
                value=_K_BOOST,
                ttl_s=10.0,
                justification=(
                    f"MPC: R predicted to degrade "
                    f"in {pred.steps_to_degradation} steps"
                ),
            )
        ]

    return []

FEPPredictiveSupervisor

FEPPredictiveSupervisor(
    n_oscillators: int,
    dt: float,
    target_R: float = 0.8,
    free_energy_threshold: float = 1.0,
    error_threshold: float = 0.25,
    drive_gain: float = 0.1,
    learning_rate: float = 0.01,
    prior_precision: float = 1.0,
)

Free-energy predictive supervisor built on VariationalPredictor.

The class turns the existing FEP-Kuramoto variational predictor into a bounded supervisor mode. It does not claim a complete biological FEP model; it exposes an auditable one-step free-energy signal and maps high surprise into conservative zeta / Psi control actions.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def __init__(
    self,
    n_oscillators: int,
    dt: float,
    target_R: float = 0.8,
    free_energy_threshold: float = 1.0,
    error_threshold: float = 0.25,
    drive_gain: float = 0.1,
    learning_rate: float = 0.01,
    prior_precision: float = 1.0,
) -> None:
    n_oscillators = _require_positive_int(n_oscillators, "n_oscillators")
    dt = _require_positive_real(dt, "dt")
    _require_unit_interval(target_R, "target_R")
    non_negative_real(free_energy_threshold, name="free_energy_threshold")
    non_negative_real(error_threshold, name="error_threshold")
    non_negative_real(drive_gain, name="drive_gain")
    non_negative_real(learning_rate, name="learning_rate")
    non_negative_real(prior_precision, name="prior_precision")

    self._n = n_oscillators
    self._dt = dt
    self._target_R = target_R
    self._free_energy_threshold = free_energy_threshold
    self._error_threshold = error_threshold
    self._drive_gain = drive_gain
    self._predictor = VariationalPredictor(
        n_oscillators,
        prior_precision=prior_precision,
        learning_rate=learning_rate,
    )
    self._last_assessment: FEPPredictionAssessment | None = None

Attributes

target_R property

target_R: float

Target order parameter used by the free-energy controller.

Returns

float Target order parameter used by the free-energy controller.

last_assessment property

last_assessment: FEPPredictionAssessment | None

Most recent free-energy assessment, if assess has run.

Returns

FEPPredictionAssessment | None Most recent free-energy assessment, if assess has run.

Methods:

assess

assess(
    phases: FloatArray, omegas: FloatArray
) -> FEPPredictionAssessment

Update the variational predictor and return audit-ready metrics.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). omegas : FloatArray Natural frequencies in rad/s, shape (N,).

Returns

FEPPredictionAssessment The free-energy prediction assessment.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def assess(self, phases: FloatArray, omegas: FloatArray) -> FEPPredictionAssessment:
    """Update the variational predictor and return audit-ready metrics.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.

    Returns
    -------
    FEPPredictionAssessment
        The free-energy prediction assessment.
    """
    phases_arr, omegas_arr = _validate_phase_inputs(phases, omegas, self._n)
    variational = self._predictor.update(phases_arr, omegas_arr, self._dt)
    observed_R, observed_psi = compute_order_parameter(phases_arr)
    predicted_R, _ = compute_order_parameter(variational.predicted_phases)
    mean_abs_error = float(np.mean(np.abs(variational.error)))
    precision_mean = float(np.mean(variational.precision))
    precision_spread = float(
        np.max(variational.precision) - np.min(variational.precision)
    )
    surprise = float(abs(observed_R - self._target_R) + mean_abs_error)
    assessment = FEPPredictionAssessment(
        free_energy=float(variational.free_energy),
        complexity=float(variational.complexity),
        mean_abs_error=mean_abs_error,
        precision_mean=precision_mean,
        precision_spread=precision_spread,
        observed_R=observed_R,
        observed_psi=observed_psi,
        predicted_R=predicted_R,
        target_R=self._target_R,
        surprise=surprise,
    )
    self._last_assessment = assessment
    return assessment

decide

decide(
    phases: FloatArray,
    omegas: FloatArray,
    upde_state: UPDEState,
    boundary_state: BoundaryState,
) -> list[ControlAction]

Return FEP-MPC control actions for the current observation.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,). omegas : FloatArray Natural frequencies in rad/s, shape (N,). upde_state : UPDEState The current UPDE state. boundary_state : BoundaryState The current boundary-observer state.

Returns

list[ControlAction] The FEP-MPC control actions for the current observation.

Raises

ValueError If the state inputs are invalid.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def decide(
    self,
    phases: FloatArray,
    omegas: FloatArray,
    upde_state: UPDEState,
    boundary_state: BoundaryState,
) -> list[ControlAction]:
    """Return FEP-MPC control actions for the current observation.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    upde_state : UPDEState
        The current UPDE state.
    boundary_state : BoundaryState
        The current boundary-observer state.

    Returns
    -------
    list[ControlAction]
        The FEP-MPC control actions for the current observation.

    Raises
    ------
    ValueError
        If the state inputs are invalid.
    """
    if not isinstance(upde_state, UPDEState):
        raise ValueError(f"upde_state must be a UPDEState, got {upde_state!r}")
    if not isinstance(boundary_state, BoundaryState):
        raise ValueError(
            f"boundary_state must be a BoundaryState, got {boundary_state!r}"
        )
    if boundary_state.hard_violations:
        return [
            ControlAction(
                knob="zeta",
                scope="global",
                value=self._drive_gain,
                ttl_s=5.0,
                justification="FEP-MPC: hard boundary violation",
            )
        ]

    assessment = self.assess(phases, omegas)
    if not self._should_act(assessment, upde_state):
        return []

    psi_target = assessment.observed_psi
    if assessment.above_target:
        psi_target = (psi_target + np.pi) % TWO_PI

    return [
        ControlAction(
            knob="zeta",
            scope="global",
            value=self._drive_gain,
            ttl_s=5.0,
            justification=(
                "FEP-MPC: free energy "
                f"{assessment.free_energy:.4g}, surprise "
                f"{assessment.surprise:.4g}"
            ),
        ),
        ControlAction(
            knob="Psi",
            scope="global",
            value=float(psi_target),
            ttl_s=5.0,
            justification="FEP-MPC: precision-weighted phase target",
        ),
    ]

reset

reset() -> None

Reset the underlying variational predictor and cached assessment.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def reset(self) -> None:
    """Reset the underlying variational predictor and cached assessment."""
    self._predictor.reset()
    self._last_assessment = None

Functions:

assess_fep_hierarchy

assess_fep_hierarchy(
    children: Mapping[str, tuple[FloatArray, FloatArray]],
    *,
    dt: float,
    child_target_R: float = 0.8,
    parent_target_R: float = 0.8,
    parent_dt: float | None = None,
    free_energy_threshold: float = 0.0,
    child_drive_gain: float = 0.08,
    parent_drive_gain: float = 0.05,
    hierarchy: str = "child_regions_to_parent_fep_supervisor",
) -> FEPHierarchyAssessment

Assess child FEP supervisors and a parent over reduced child coherence.

Each child receives its own FEPPredictiveSupervisor. The parent encodes child coherence as phases via arccos(2R - 1) so the same FEP machinery can reason over cross-child coherence without accessing raw child signals.

Parameters

children : Mapping[str, tuple[FloatArray, FloatArray]] Child supervisor summaries. dt : float Integration step size. child_target_R : float Target order parameter for each child. parent_target_R : float Target order parameter for the parent. parent_dt : float | None Parent integration step size, or None. free_energy_threshold : float Free-energy threshold above which control acts. child_drive_gain : float Drive gain applied at the child level. parent_drive_gain : float Drive gain applied at the parent level. hierarchy : str Hierarchy label.

Returns

FEPHierarchyAssessment The hierarchical free-energy assessment.

Source code in src/scpn_phase_orchestrator/supervisor/predictive.py

def assess_fep_hierarchy(
    children: Mapping[str, tuple[FloatArray, FloatArray]],
    *,
    dt: float,
    child_target_R: float = 0.8,
    parent_target_R: float = 0.8,
    parent_dt: float | None = None,
    free_energy_threshold: float = 0.0,
    child_drive_gain: float = 0.08,
    parent_drive_gain: float = 0.05,
    hierarchy: str = "child_regions_to_parent_fep_supervisor",
) -> FEPHierarchyAssessment:
    """Assess child FEP supervisors and a parent over reduced child coherence.

    Each child receives its own ``FEPPredictiveSupervisor``. The parent encodes
    child coherence as phases via ``arccos(2R - 1)`` so the same FEP machinery
    can reason over cross-child coherence without accessing raw child signals.

    Parameters
    ----------
    children : Mapping[str, tuple[FloatArray, FloatArray]]
        Child supervisor summaries.
    dt : float
        Integration step size.
    child_target_R : float
        Target order parameter for each child.
    parent_target_R : float
        Target order parameter for the parent.
    parent_dt : float | None
        Parent integration step size, or ``None``.
    free_energy_threshold : float
        Free-energy threshold above which control acts.
    child_drive_gain : float
        Drive gain applied at the child level.
    parent_drive_gain : float
        Drive gain applied at the parent level.
    hierarchy : str
        Hierarchy label.

    Returns
    -------
    FEPHierarchyAssessment
        The hierarchical free-energy assessment.
    """
    _validate_hierarchy_inputs(
        children=children,
        dt=dt,
        parent_dt=parent_dt,
        child_target_R=child_target_R,
        parent_target_R=parent_target_R,
        free_energy_threshold=free_energy_threshold,
        child_drive_gain=child_drive_gain,
        parent_drive_gain=parent_drive_gain,
    )
    child_records: list[FEPHierarchyChildAssessment] = []
    child_rs: list[float] = []
    for name, (phases, omegas) in children.items():
        phases_arr, omegas_arr = _validate_child_observation(name, phases, omegas)
        supervisor = FEPPredictiveSupervisor(
            n_oscillators=phases_arr.size,
            dt=dt,
            target_R=child_target_R,
            free_energy_threshold=free_energy_threshold,
            drive_gain=child_drive_gain,
        )
        assessment = supervisor.assess(phases_arr, omegas_arr)
        actions = tuple(
            supervisor.decide(
                phases_arr,
                omegas_arr,
                _state_from_r(assessment.observed_R),
                BoundaryState(),
            )
        )
        child_records.append(
            FEPHierarchyChildAssessment(
                name=name,
                assessment=assessment,
                actions=actions,
            )
        )
        child_rs.append(assessment.observed_R)

    child_r_arr = np.asarray(child_rs, dtype=np.float64)
    parent_phases = _coherence_to_parent_phases(child_r_arr)
    parent_omegas = np.full(parent_phases.shape, 1.0, dtype=np.float64)
    parent = FEPPredictiveSupervisor(
        n_oscillators=parent_phases.size,
        dt=dt if parent_dt is None else parent_dt,
        target_R=parent_target_R,
        free_energy_threshold=free_energy_threshold,
        drive_gain=parent_drive_gain,
    )
    parent_assessment = parent.assess(parent_phases, parent_omegas)
    parent_actions = tuple(
        parent.decide(
            parent_phases,
            parent_omegas,
            _state_from_r(parent_assessment.observed_R),
            BoundaryState(),
        )
    )
    return FEPHierarchyAssessment(
        hierarchy=hierarchy,
        children=tuple(child_records),
        parent_assessment=parent_assessment,
        parent_actions=parent_actions,
        child_R_values=tuple(float(value) for value in child_r_arr),
        parent_phase_encoding=tuple(float(value) for value in parent_phases),
    )

Regime Manager

Finite state machine for synchronization regimes with hysteresis. States: NOMINAL, DEGRADED, CRITICAL. Transitions based on R thresholds with configurable hysteresis bands to prevent oscillation between states.

regimes

Regime classification with hysteresis, cooldown, and optional event emission.

RegimeManager classifies reduced UPDE and boundary state into nominal, degraded, critical, or recovery regimes, then applies cooldown and hysteresis rules before committing transitions. Transition history is bounded and optional events are posted through an injected in-process bus. The manager emits regime state only; policy modules decide any control proposals.

Classes

Regime

Bases: Enum

Operational regime of the SCPN supervisor.

RegimeManager

RegimeManager(
    hysteresis: float = 0.05,
    cooldown_steps: int = 10,
    event_bus: EventBus | None = None,
    hysteresis_hold_steps: int = 0,
)

Classify system state into regimes with hysteresis and cooldown.

Source code in src/scpn_phase_orchestrator/supervisor/regimes.py

def __init__(
    self,
    hysteresis: float = 0.05,
    cooldown_steps: int = 10,
    event_bus: EventBus | None = None,
    hysteresis_hold_steps: int = 0,
) -> None:
    self._hysteresis = _validate_nonnegative_float(
        hysteresis,
        name="hysteresis",
    )
    self._cooldown_steps = _validate_nonnegative_int(
        cooldown_steps,
        name="cooldown_steps",
    )
    if event_bus is not None and not isinstance(event_bus, EventBus):
        raise ValueError(f"event_bus must be an EventBus, got {event_bus!r}")
    self._current = Regime.NOMINAL
    self._step_counter = 0
    self._last_transition_step = -self._cooldown_steps
    self._event_bus = event_bus
    self._hysteresis_hold_steps = _validate_nonnegative_int(
        hysteresis_hold_steps,
        name="hysteresis_hold_steps",
    )
    self._downward_streak = 0
    self.transition_history: deque[tuple[int, Regime, Regime]] = deque(maxlen=100)

Attributes

current_regime property

current_regime: Regime

The regime established after the most recent transition.

Returns

Regime The regime established after the most recent transition.

Methods:

evaluate

evaluate(
    upde_state: UPDEState, boundary_state: BoundaryState
) -> Regime

Propose a regime based on current R values and boundary state.

Parameters

upde_state : UPDEState The current UPDE state. boundary_state : BoundaryState The current boundary-observer state.

Returns

Regime The regime proposed for the current state.

Source code in src/scpn_phase_orchestrator/supervisor/regimes.py

def evaluate(self, upde_state: UPDEState, boundary_state: BoundaryState) -> Regime:
    """Propose a regime based on current R values and boundary state.

    Parameters
    ----------
    upde_state : UPDEState
        The current UPDE state.
    boundary_state : BoundaryState
        The current boundary-observer state.

    Returns
    -------
    Regime
        The regime proposed for the current state.
    """
    if boundary_state.hard_violations:
        return Regime.CRITICAL

    avg_r = self._mean_r(upde_state)

    if avg_r < _R_CRITICAL:
        return Regime.CRITICAL

    is_recovering = self._current in (Regime.CRITICAL, Regime.RECOVERY)

    if avg_r < _R_DEGRADED:
        if is_recovering:
            return Regime.RECOVERY
        return Regime.DEGRADED

    if self._current == Regime.DEGRADED and avg_r < _R_DEGRADED + self._hysteresis:
        return Regime.DEGRADED
    if is_recovering and avg_r < _R_DEGRADED + self._hysteresis:
        return Regime.RECOVERY

    if self._current == Regime.CRITICAL:
        return Regime.RECOVERY

    return Regime.NOMINAL

transition

transition(proposed: Regime) -> Regime

Apply cooldown/hysteresis logic and commit the regime transition.

Parameters

proposed : Regime The proposed regime to transition into.

Returns

Regime The committed regime after cooldown/hysteresis.

Source code in src/scpn_phase_orchestrator/supervisor/regimes.py

def transition(self, proposed: Regime) -> Regime:
    """Apply cooldown/hysteresis logic and commit the regime transition.

    Parameters
    ----------
    proposed : Regime
        The proposed regime to transition into.

    Returns
    -------
    Regime
        The committed regime after cooldown/hysteresis.
    """
    proposed = _validate_regime(proposed)
    self._step_counter += 1

    if proposed == self._current:
        self._downward_streak = 0
        return self._current

    # Soft downward transitions (non-critical) require N consecutive steps
    is_downward = self._regime_rank(proposed) > self._regime_rank(self._current)
    if (
        is_downward
        and proposed != Regime.CRITICAL
        and self._hysteresis_hold_steps > 0
    ):
        self._downward_streak += 1
        if self._downward_streak < self._hysteresis_hold_steps:
            return self._current
    else:
        self._downward_streak = 0

    in_cooldown = (
        self._step_counter - self._last_transition_step
    ) < self._cooldown_steps
    if in_cooldown and proposed != Regime.CRITICAL:
        return self._current

    prev = self._current
    self._last_transition_step = self._step_counter
    self._current = proposed
    self._downward_streak = 0
    self.transition_history.append((self._step_counter, prev, proposed))
    self._emit_transition(prev, proposed)
    return proposed

force_transition

force_transition(regime: Regime) -> Regime

Bypass cooldown and hysteresis hold.

Parameters

regime : Regime The current control regime.

Returns

Regime The regime after a forced transition.

Source code in src/scpn_phase_orchestrator/supervisor/regimes.py

def force_transition(self, regime: Regime) -> Regime:
    """Bypass cooldown and hysteresis hold.

    Parameters
    ----------
    regime : Regime
        The current control regime.

    Returns
    -------
    Regime
        The regime after a forced transition.
    """
    regime = _validate_regime(regime)
    self._step_counter += 1
    prev = self._current
    if regime == prev:
        return prev
    self._last_transition_step = self._step_counter
    self._current = regime
    self._downward_streak = 0
    self.transition_history.append((self._step_counter, prev, regime))
    self._emit_transition(prev, regime)
    return regime

Petri Net State Machine

Formal Petri net FSM with guard conditions, token counts, and priority-based transition firing. Enables formal verification of safety properties (deadlock freedom, liveness).

from scpn_phase_orchestrator.supervisor.petri_net import PetriNet

net = PetriNet()
net.add_place("nominal", tokens=1)
net.add_place("critical", tokens=0)
net.add_transition("degrade", inputs=["nominal"], outputs=["critical"],
                    guard=lambda ctx: ctx["R"] < 0.3)

petri_net

Guarded Petri-net primitives for deterministic regime transition modeling.

The module defines validated places, weighted arcs, guards, transitions, markings, and a first-match-priority Petri net. Marking updates are local and non-negative, guard metrics must be finite, and net construction rejects arcs to unknown places. The engine performs no event emission or policy action mapping; adapter modules own those boundaries.

Classes

Place dataclass

Place(name: str)

Named place (state) in the Petri net.

Arc dataclass

Arc(place: str, weight: int = 1)

Weighted arc connecting a place to a transition.

Guard dataclass

Guard(metric: str, op: str, threshold: float)

Boolean guard condition on a named metric (e.g. 'stability_proxy > 0.6').

Methods:

evaluate

evaluate(ctx: Mapping[str, float]) -> bool

Return True if the guard condition is satisfied by ctx.

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def evaluate(self, ctx: Mapping[str, float]) -> bool:
    """Return True if the guard condition is satisfied by *ctx*."""
    val = ctx.get(self.metric)
    if val is None:
        return False
    val = _validate_finite_real(val, name=f"context metric {self.metric!r}")
    fn = _OPS.get(self.op)
    if fn is None:
        return False
    return bool(fn(val, self.threshold))

Transition dataclass

Transition(
    name: str,
    inputs: list[Arc],
    outputs: list[Arc],
    guard: Guard | None = None,
)

Petri net transition with input/output arcs and optional guard.

Marking dataclass

Marking(tokens: dict[str, int] = dict())

Token distribution across places in a Petri net.

Methods:

active_places

active_places() -> list[str]

Return names of places that hold at least one token.

Returns

list[str] Return names of places that hold at least one token.

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def active_places(self) -> list[str]:
    """Return names of places that hold at least one token.

    Returns
    -------
    list[str]
        Return names of places that hold at least one token.
    """
    return [p for p, n in self.tokens.items() if n > 0]

copy

copy() -> Marking

Return a shallow copy of this marking.

Returns

Marking Return a shallow copy of this marking.

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def copy(self) -> Marking:
    """Return a shallow copy of this marking.

    Returns
    -------
    Marking
        Return a shallow copy of this marking.
    """
    return Marking(tokens=dict(self.tokens))

PetriNet

PetriNet(
    places: list[Place], transitions: list[Transition]
)

Classical Petri net with guard-gated transitions.

step() fires at most one enabled transition per call (first-match priority).

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def __init__(
    self,
    places: list[Place],
    transitions: list[Transition],
) -> None:
    self._place_names = frozenset(p.name for p in places)
    self._transitions = transitions
    self._guard_metrics = frozenset(
        t.guard.metric for t in transitions if t.guard is not None
    )
    self._validate()

Attributes

place_names property

place_names: frozenset[str]

All place names registered in this net.

Returns

frozenset[str] All place names registered in this net.

transitions property

transitions: list[Transition]

All transitions in firing-priority order.

Returns

list[Transition] All transitions in firing-priority order.

guard_metrics property

guard_metrics: frozenset[str]

Whitelisted context metric names used by transition guards.

Returns

frozenset[str] Whitelisted context metric names used by transition guards.

Methods:

enabled

enabled(
    marking: Marking, ctx: Mapping[str, float]
) -> list[Transition]

Return all transitions whose input arcs and guards are satisfied.

Parameters

marking : Marking The Petri net marking (token distribution). ctx : Mapping[str, float] Context metric values keyed by guard-metric name.

Returns

list[Transition] The transitions whose input arcs and guards are satisfied.

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def enabled(self, marking: Marking, ctx: Mapping[str, float]) -> list[Transition]:
    """Return all transitions whose input arcs and guards are satisfied.

    Parameters
    ----------
    marking : Marking
        The Petri net marking (token distribution).
    ctx : Mapping[str, float]
        Context metric values keyed by guard-metric name.

    Returns
    -------
    list[Transition]
        The transitions whose input arcs and guards are satisfied.
    """
    ctx = self._validated_context(ctx)
    result = []
    for t in self._transitions:
        if t.guard is not None and not t.guard.evaluate(ctx):
            continue
        if all(marking[arc.place] >= arc.weight for arc in t.inputs):
            result.append(t)
    return result

fire

fire(marking: Marking, transition: Transition) -> Marking

Fire transition, consuming input tokens and producing output tokens.

Parameters

marking : Marking The Petri net marking (token distribution). transition : Transition The transition to fire.

Returns

Marking The marking after firing the transition.

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def fire(self, marking: Marking, transition: Transition) -> Marking:
    """Fire *transition*, consuming input tokens and producing output tokens.

    Parameters
    ----------
    marking : Marking
        The Petri net marking (token distribution).
    transition : Transition
        The transition to fire.

    Returns
    -------
    Marking
        The marking after firing the transition.
    """
    new = marking.copy()
    for arc in transition.inputs:
        new[arc.place] = new[arc.place] - arc.weight
    for arc in transition.outputs:
        new[arc.place] = new[arc.place] + arc.weight
    return new

step

step(
    marking: Marking, ctx: Mapping[str, float]
) -> tuple[Marking, Transition | None]

Fire the first enabled transition, return (new_marking, fired_transition).

Parameters

marking : Marking The Petri net marking (token distribution). ctx : Mapping[str, float] Context metric values keyed by guard-metric name.

Returns

tuple[Marking, Transition | None] The new marking and the fired transition (or None).

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def step(
    self, marking: Marking, ctx: Mapping[str, float]
) -> tuple[Marking, Transition | None]:
    """Fire the first enabled transition, return (new_marking, fired_transition).

    Parameters
    ----------
    marking : Marking
        The Petri net marking (token distribution).
    ctx : Mapping[str, float]
        Context metric values keyed by guard-metric name.

    Returns
    -------
    tuple[Marking, Transition | None]
        The new marking and the fired transition (or ``None``).
    """
    ctx = self._validated_context(ctx)
    for t in self._transitions:
        if t.guard is not None and not t.guard.evaluate(ctx):
            continue
        if all(marking[arc.place] >= arc.weight for arc in t.inputs):
            return self.fire(marking, t), t
    return marking, None

Functions:

parse_guard

parse_guard(text: str) -> Guard

Parse guard string like 'stability_proxy > 0.6'.

Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py

def parse_guard(text: str) -> Guard:
    """Parse guard string like 'stability_proxy > 0.6'."""
    parts = text.split()
    if len(parts) != 3:
        raise PolicyError(f"guard must be 'metric op threshold', got {text!r}")
    try:
        threshold = float(parts[2])
    except ValueError as exc:
        raise PolicyError(f"threshold must be finite, got {parts[2]!r}") from exc
    return Guard(metric=parts[0], op=parts[1], threshold=threshold)

Policy Engine

Rule-based policy evaluation for supervisor actions. Rules define conditions (R thresholds, boundary violations) and actions (coupling boost, frequency adjustment, external drive).

policy

Reactive supervisor policy that maps regimes and state into control proposals.

SupervisorPolicy derives a proposed regime from direct metrics or an optional Petri adapter, commits it through RegimeManager, and emits bounded ControlAction proposals for degraded, critical, or recovery states. Petri failures fall back to direct regime logic. The policy only proposes actions; it does not apply actuation or mutate coupling matrices.

Classes

SupervisorPolicyGains dataclass

SupervisorPolicyGains(
    k_bump: float = 0.05,
    zeta_bump: float = 0.1,
    k_reduce: float = -0.03,
    restore_fraction: float = 0.5,
)

Tunable regime-action gains for a deployment-specific supervisor.

SupervisorPolicy

SupervisorPolicy(
    regime_manager: RegimeManager,
    petri_adapter: PetriNetAdapter | None = None,
    gains: SupervisorPolicyGains | None = None,
    admission_gate: PolicyCBFAdmissionGate | None = None,
)

Decide control actions based on regime and system state.

When petri_adapter is provided, regime is derived from the Petri net marking instead of RegimeManager.evaluate().

Source code in src/scpn_phase_orchestrator/supervisor/policy.py

def __init__(
    self,
    regime_manager: RegimeManager,
    petri_adapter: PetriNetAdapter | None = None,
    gains: SupervisorPolicyGains | None = None,
    admission_gate: PolicyCBFAdmissionGate | None = None,
) -> None:
    self._regime_manager = regime_manager
    self._petri_adapter = petri_adapter
    self._gains = gains or SupervisorPolicyGains()
    self._admission_gate = admission_gate
    self._last_admission_records: tuple[PolicyCBFAdmissionRecord, ...] = ()

Attributes

last_admission_records property

last_admission_records: tuple[PolicyCBFAdmissionRecord, ...]

Return the CBF admission records from the latest decision.

Returns

tuple[PolicyCBFAdmissionRecord, ...] Deterministic audit records for actions matched by the optional CBF admission gate in the previous :meth:decide call.

Methods:

decide

decide(
    upde_state: UPDEState,
    boundary_state: BoundaryState,
    petri_ctx: dict[str, float] | None = None,
) -> list[ControlAction]

Evaluate regime and return control actions for the current state.

Parameters

upde_state : UPDEState The current UPDE state. boundary_state : BoundaryState The current boundary-observer state. petri_ctx : dict[str, float] | None Petri context metric values, or None.

Returns

list[ControlAction] The control actions proposed for the current state.

Source code in src/scpn_phase_orchestrator/supervisor/policy.py

def decide(
    self,
    upde_state: UPDEState,
    boundary_state: BoundaryState,
    petri_ctx: dict[str, float] | None = None,
) -> list[ControlAction]:
    """Evaluate regime and return control actions for the current state.

    Parameters
    ----------
    upde_state : UPDEState
        The current UPDE state.
    boundary_state : BoundaryState
        The current boundary-observer state.
    petri_ctx : dict[str, float] | None
        Petri context metric values, or ``None``.

    Returns
    -------
    list[ControlAction]
        The control actions proposed for the current state.
    """
    proposed = self._proposed_regime(upde_state, boundary_state, petri_ctx)
    regime = self._regime_manager.transition(proposed)

    actions = self._actions_for_regime(regime, upde_state)
    if self._admission_gate is None:
        self._last_admission_records = ()
    else:
        admitted = self._admission_gate.admit_actions(
            actions, upde_state, boundary_state
        )
        actions = list(admitted.actions)
        self._last_admission_records = admitted.records
    logger.info(
        "supervisor decide: regime=%s actions=%d",
        regime.value,
        len(actions),
        extra={
            "regime": regime.value,
            "n_actions": len(actions),
            "n_violations": len(boundary_state.violations),
            "stability_proxy": upde_state.stability_proxy,
            "knobs": [a.knob for a in actions],
        },
    )
    return actions

Three-Factor Hebbian Plasticity

Coupling adaptation rule: K_ij += lr × eligibility × modulator × gate.

1. Eligibility: cos(θ_j - θ_i) — pairwise Hebbian trace
2. Modulator: scalar neuromodulatory signal from L16 director
3. Phase gate: Boolean from topological-integration gate

Grounded in Friston 2005 on free energy and synaptic plasticity.

plasticity

Validated three-factor plasticity updates for coupling matrices.

The module computes pairwise phase eligibility traces and applies a modulator-gated Hebbian update to K_nm. Public functions reject boolean, non-numeric, non-finite, non-vector, non-square, and shape-mismatched inputs so plasticity cannot corrupt coupling state silently. The update preserves the Kuramoto coupling contract by requiring non-negative zero-diagonal K_nm, bounded zero-diagonal eligibility traces, and finite real scalar controls.

Functions:

compute_eligibility

compute_eligibility(phases: FloatArray) -> FloatArray

Pairwise Hebbian eligibility trace: cos(theta_j - theta_i).

Returns shape (n, n) with zero diagonal.

Parameters

phases : FloatArray Oscillator phases in radians, shape (N,).

Returns

FloatArray The pairwise Hebbian eligibility trace cos(θ_j − θ_i).

Source code in src/scpn_phase_orchestrator/coupling/plasticity.py

def compute_eligibility(phases: FloatArray) -> FloatArray:
    """Pairwise Hebbian eligibility trace: cos(theta_j - theta_i).

    Returns shape (n, n) with zero diagonal.

    Parameters
    ----------
    phases : FloatArray
        Oscillator phases in radians, shape ``(N,)``.

    Returns
    -------
    FloatArray
        The pairwise Hebbian eligibility trace ``cos(θ_j − θ_i)``.
    """
    phases = _validate_phase_vector(phases, name="phases")
    diffs = phases[np.newaxis, :] - phases[:, np.newaxis]
    elig = np.cos(diffs)
    np.fill_diagonal(elig, 0.0)
    result: FloatArray = elig
    return result

three_factor_update

three_factor_update(
    knm: FloatArray,
    eligibility: FloatArray,
    modulator: float,
    phase_gate: bool,
    lr: float = 0.01,
) -> FloatArray

Three-factor plasticity rule: K_ij += lr * eligibility_ij * M * gate.

Factors

1. eligibility — pairwise phase correlation (Hebbian trace)
2. modulator — scalar reward/error signal from L16 director
3. phase_gate — boolean from the topological-integration gate

Friston 2005, Philos. Trans. R. Soc. B 360:815-836 (free energy & synaptic plasticity).

Parameters

knm : FloatArray current coupling matrix, shape (n, n). eligibility : FloatArray Hebbian trace, shape (n, n). modulator : float scalar neuromodulatory signal. phase_gate : bool if False, no update occurs (integration gate below threshold). lr : float learning rate.

Returns

FloatArray Updated coupling matrix (new array, does not mutate input).

Raises

TypeError If an argument has the wrong type. ValueError If the eligibility or coupling shapes mismatch.

Source code in src/scpn_phase_orchestrator/coupling/plasticity.py

def three_factor_update(
    knm: FloatArray,
    eligibility: FloatArray,
    modulator: float,
    phase_gate: bool,
    lr: float = 0.01,
) -> FloatArray:
    """Three-factor plasticity rule: K_ij += lr * eligibility_ij * M * gate.

    Factors:
        1. eligibility — pairwise phase correlation (Hebbian trace)
        2. modulator — scalar reward/error signal from L16 director
        3. phase_gate — boolean from the topological-integration gate

    Friston 2005, Philos. Trans. R. Soc. B
    360:815-836 (free energy & synaptic plasticity).

    Parameters
    ----------
    knm : FloatArray
        current coupling matrix, shape (n, n).
    eligibility : FloatArray
        Hebbian trace, shape (n, n).
    modulator : float
        scalar neuromodulatory signal.
    phase_gate : bool
        if False, no update occurs (integration gate below threshold).
    lr : float
        learning rate.

    Returns
    -------
    FloatArray
        Updated coupling matrix (new array, does not mutate input).

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    ValueError
        If the eligibility or coupling shapes mismatch.
    """
    knm = _validate_coupling_matrix(knm)
    eligibility = _validate_eligibility_matrix(eligibility)
    if eligibility.shape != knm.shape:
        raise ValueError(
            "eligibility shape "
            f"{eligibility.shape} does not match knm shape {knm.shape}"
        )
    modulator = _validate_finite_real(modulator, name="modulator")
    if not isinstance(phase_gate, bool):
        raise TypeError("phase_gate must be a bool")
    lr = _validate_learning_rate(lr)
    if not phase_gate:
        return knm.copy()
    delta = lr * eligibility * modulator
    updated = np.maximum(knm + delta, 0.0)
    np.fill_diagonal(updated, 0.0)
    result: FloatArray = updated
    return result

Transfer Entropy Adaptive Coupling

K_ij(t+1) = (1-decay)·K_ij(t) + lr·TE(i→j)

Unlike symmetric Hebbian learning, transfer entropy breaks symmetry to detect causal direction. Coupling adapts based on directed information flow (Lizier 2012).

te_adaptive

Transfer-entropy-guided coupling adaptation for offline matrix updates.

te_adapt_coupling derives a directed transfer-entropy matrix from phase history and combines it with the current coupling matrix under learning-rate and decay parameters. The Python fallback clamps the returned coupling to non-negative values and clears self-coupling; the optional Rust path preserves the same dense N x N output contract. The helper returns a new matrix and does not mutate live solver state or apply actuation.

Functions:

te_adapt_coupling

te_adapt_coupling(
    knm: FloatArray,
    phase_history: FloatArray,
    lr: float = 0.01,
    decay: float = 0.0,
    n_bins: int = 8,
) -> FloatArray

Adapt coupling matrix using transfer entropy as learning signal.

K_ij(t+1) = (1-decay) * K_ij(t) + lr * TE(i→j)

Strengthens coupling along causal information flow channels. Weakens where there is no causal influence.

Lizier 2012, "Local Information Transfer as a Spatiotemporal Filter for Complex Systems," Physical Review E 77(2):026110.

Parameters

knm : FloatArray current (n, n) coupling matrix. phase_history : FloatArray (n, T) recent phase trajectories. lr : float learning rate for TE-based update. decay : float coupling decay rate per update (0 = no decay). n_bins : int histogram bins for TE estimation.

Returns

FloatArray FloatArray The coupling matrix adapted by the transfer-entropy learning signal.

Raises

RuntimeError If the transfer-entropy backend fails.

Source code in src/scpn_phase_orchestrator/coupling/te_adaptive.py

def te_adapt_coupling(
    knm: FloatArray,
    phase_history: FloatArray,
    lr: float = 0.01,
    decay: float = 0.0,
    n_bins: int = 8,
) -> FloatArray:
    """Adapt coupling matrix using transfer entropy as learning signal.

    K_ij(t+1) = (1-decay) * K_ij(t) + lr * TE(i→j)

    Strengthens coupling along causal information flow channels.
    Weakens where there is no causal influence.

    Lizier 2012, "Local Information Transfer as a Spatiotemporal Filter
    for Complex Systems," Physical Review E 77(2):026110.

    Parameters
    ----------
    knm : FloatArray
        current (n, n) coupling matrix.
    phase_history : FloatArray
        (n, T) recent phase trajectories.
    lr : float
        learning rate for TE-based update.
    decay : float
        coupling decay rate per update (0 = no decay).
    n_bins : int
        histogram bins for TE estimation.

    Returns
    -------
    FloatArray
        FloatArray The coupling matrix adapted by the transfer-entropy learning signal.

    Raises
    ------
    RuntimeError
        If the transfer-entropy backend fails.
    """
    knm = _validate_knm(knm)
    n = knm.shape[0]
    phase_history = _validate_phase_history(phase_history, n=n)
    lr = _validate_non_negative_real(lr, name="lr")
    decay = _validate_decay(decay)
    n_bins = _validate_n_bins(n_bins)
    te = _validate_transfer_entropy_scores(
        transfer_entropy_matrix(phase_history, n_bins=n_bins),
        n=n,
    )
    if _HAS_RUST:
        k_flat = np.ascontiguousarray(knm.ravel(), dtype=np.float64)
        t_flat = np.ascontiguousarray(te.ravel(), dtype=np.float64)
        result_flat = np.asarray(
            _rust_te_adapt(k_flat, t_flat, n, lr, decay),
            dtype=np.float64,
        )
        if result_flat.size != n * n:
            raise RuntimeError("TE adaptive backend returned wrong shape")
        return _validate_adapted_coupling(result_flat.reshape(n, n), n=n)
    knm_new = (1.0 - decay) * knm + lr * te
    np.fill_diagonal(knm_new, 0.0)
    result: FloatArray = np.maximum(knm_new, 0.0)
    return result

Audit Trail with Deterministic Replay

SHA256-chained JSONL audit log with per-step regime, R values, actions, and coupling state. Enables deterministic replay and cryptographic verification of simulation reproducibility.

audit

Audit logging, event streaming, and deterministic replay entry points.

The audit package owns append-only JSONL records, optional hash-chained event streams, replay reconstruction, and integrity verification. Public helpers fail closed on malformed signatures or key material while preserving unsigned development logs for local reproducibility workflows.

Functions:

__dir__

__dir__() -> list[str]

Return module attributes plus lazy public audit exports.

Source code in src/scpn_phase_orchestrator/audit/__init__.py

def __dir__() -> list[str]:
    """Return module attributes plus lazy public audit exports."""
    return sorted({*globals(), *__all__})

__getattr__

__getattr__(name: str) -> Any

Lazily expose audit package exports without import cycles.

Source code in src/scpn_phase_orchestrator/audit/__init__.py

def __getattr__(name: str) -> Any:
    """Lazily expose audit package exports without import cycles."""
    module_name = _EXPORT_MODULES.get(name)
    if module_name is None:
        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

    from importlib import import_module

    value = getattr(import_module(module_name), name)
    globals()[name] = value
    return value

What makes this closed-loop in production

Most oscillator libraries expose observability but stop before actuation. SPO’s control surface pushes into action selection with three constraints in the same loop:

• Prediction: expected coherence trend via MPC proxy.
• Safety: regime and evidence checks before promotion.
• Governance: audit-ready proposal records for every non-trivial action.

This changes operational posture from “watch and decide” to “predict-then-verify-then-act” under explicit constraints.

Operator operating model

For day-to-day deployment, teams typically configure:

1. A baseline monitor that maps R and chimera_index to supervisory inputs.
2. A policy layer with conservative defaults.
3. A Petri-NET-safe transition set for state changes.
4. An audit sink that captures state transitions and proposal rationale.

That model keeps tuning sessions reviewable and supports post-incident replay without manual reconstruction of transient state.

Stability and rollback behavior

Because state transitions are explicit and regime-gated, operators can define clear rollback boundaries: if coherence drops or monitor evidence regresses after an action, policy proposals can be bounded and reverted before the next control cycle.

Operational placement in closed-loop projects

Use this page when the target outcome is a stable control loop, not just a monitoring dashboard. The control stack in this repository is built in layers:

• prediction: MPC and monitors produce a near-term risk estimate,
• policy and regime selection: Petri-Net and policy DSL select candidate actions,
• constraint application: projector, boundaries, and imprint constraints limit per-cycle movement,
• governance: audit logging and replay preserve every non-trivial change.

A common rollout order is:

1. choose monitor set (R, PLV, boundary metrics),
2. define objectives and regime thresholds,
3. dry-run policy and projection limits,
4. replay the same sequence with fixed seeds,
5. promote only when both expected and observed lock metrics match the decision gate.

Keep this page as a production entry for teams that need a predictable control path after data onboarding and domain calibration.

Control posture and evidence contract

Every production setup using this page should keep three artifacts versioned:

• policy definition files (rules, cooldowns, caps),
• latest audit configuration and lockfile choice,
• baseline replay traces from the last accepted tuning cycle.

That set is what allows an operator to compare one control action against history and answer whether the action improved or degraded the system trajectory.

The supervisor surfaces are intentionally split into prediction, policy, and governance layers. If one layer is changed, replay the full sequence before promotion so you can isolate the effect of that change without mixing it with backend or extractor drift.