Skip to content

Actuation

The actuation subsystem converts high-level supervisor decisions into bounded, rate-limited control commands. It sits between the supervisor (which decides what to change) and the physical/virtual actuators (which execute the change). This separation of concerns is critical for safety: the supervisor can propose aggressive actions, but the actuation layer enforces physical constraints.

Pipeline position

SupervisorPolicy.decide()
         │
         ↓
  list[ControlAction]
         │
         ↓
  ActuationMapper.map_actions()     ← routing by knob + scope
         │
         ↓
  ActionProjector.project()         ← rate limit + value bounds
         │
         ↓
  Actuator commands (dict)          → Modbus/gRPC/HTTP output
         │
         ↓
  UPDEEngine.step(knm + ΔK, zeta + Δζ, ...)  ← next cycle

The actuation subsystem is the output adapter of the SPO pipeline. Without it, supervisor decisions would be unbounded floating-point values that could crash the integrator.

Control Actions

A ControlAction is the universal message format between supervisor and actuators:

ControlAction (dataclass)

Field Type Description
knob str Parameter to change: K, zeta, psi, alpha
scope str Target: global or layer_{n}
value float Proposed parameter value
ttl_s float Time-to-live in seconds
justification str Human-readable reason (audit trail)

Knob semantics

Knob Engine parameter Effect
K Coupling strength K_ij Increases/decreases synchronisation pull
zeta External drive amplitude ζ Damping or excitation
psi External drive phase Ψ Phase of external reference
alpha Phase lag α_ij Shifts preferred phase relationships

Actuation Mapper

Maps control actions to actuator-specific command dictionaries.

ActuatorMapping (dataclass)

Field Type Description
name str Actuator identifier
knob str Which control knob it responds to
scope str Which layer(s) it affects
limits tuple[float, float] (lo, hi) value bounds

ActuationMapper

ActuationMapper(mappings: list[ActuatorMapping])

Methods:

Method Signature Description
map_actions (actions: list[ControlAction]) → list[dict] Route actions to actuators
validate_action (action: ControlAction) → bool Check if any actuator handles this knob+scope

Routing rules

  1. Action's knob must match an actuator's knob
  2. Action's scope must match an actuator's scope (or "global" matches all)
  3. Action's value is clamped to actuator's limits
  4. Unroutable actions are silently dropped (no matching actuator)

Usage

from scpn_phase_orchestrator.actuation.mapper import ActuationMapper, ControlAction
from scpn_phase_orchestrator.binding.types import ActuatorMapping

mappings = [
    ActuatorMapping(name="K_amp", knob="K", scope="global", limits=(0.0, 5.0)),
    ActuatorMapping(name="zeta_drive", knob="zeta", scope="global", limits=(0.0, 1.0)),
]
mapper = ActuationMapper(mappings)

actions = [ControlAction(knob="K", scope="global", value=3.0, ttl_s=5.0,
                         justification="MPC pre-emptive boost")]
commands = mapper.map_actions(actions)

Edge cases

Input Behaviour
Empty mappings map_actions() returns [] for any input
Empty actions Returns []
No matching actuator Action silently dropped
validate_action() on unroutable Returns False

Performance: map_actions() < 10 μs.

mapper

Map validated control actions onto configured actuator records.

The mapper is deliberately data-only: it validates binding-level actuator metadata, clamps finite action values to each actuator limit, and returns command dictionaries for a transport or hardware layer to consume. Invalid action values are not sent onward, and invalid mapping definitions fail at construction time.

Classes

ControlAction dataclass 

ControlAction(
    knob: str,
    scope: str,
    value: float,
    ttl_s: float,
    justification: str,
)

A single control command targeting a specific knob and scope.

ActuationMapper 

ActuationMapper(actuator_mappings: list[ActuatorMapping])

Convert ControlActions to actuator-specific command dicts.

Source code in src/scpn_phase_orchestrator/actuation/mapper.py 
45
46
47
48
49
50
51
52
53
def __init__(self, actuator_mappings: list[ActuatorMapping]):
    from scpn_phase_orchestrator.binding.types import ActuatorMapping

    self._by_knob: dict[str, list[ActuatorMapping]] = {}
    for am in actuator_mappings:
        if not isinstance(am, ActuatorMapping):
            raise ValueError("actuator_mappings entries must be ActuatorMapping")
        _validate_mapping(am)
        self._by_knob.setdefault(am.knob, []).append(am)
Methods:
map_actions 
map_actions(
    actions: list[ControlAction],
) -> list[dict[str, Any]]

Convert ControlActions into actuator command dicts, clamping to limits.

Parameters

actions : list[ControlAction] The control actions.

Returns

list[dict[str, Any]] One command dict (actuator, knob, scope, value, ttl_s) per matching actuator, with the value clamped to that actuator's limits; actions with a non-finite value are dropped.

Source code in src/scpn_phase_orchestrator/actuation/mapper.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def map_actions(self, actions: list[ControlAction]) -> list[dict[str, Any]]:
    """Convert ControlActions into actuator command dicts, clamping to limits.

    Parameters
    ----------
    actions : list[ControlAction]
        The control actions.

    Returns
    -------
    list[dict[str, Any]]
        One command dict (``actuator``, ``knob``, ``scope``, ``value``,
        ``ttl_s``) per matching actuator, with the value clamped to that
        actuator's limits; actions with a non-finite value are dropped.
    """
    commands = []
    for action in actions:
        if not _finite_real(action.value):
            continue
        mappings = self._by_knob.get(action.knob, [])
        for am in mappings:
            if am.scope == action.scope or action.scope == "global":
                commands.append(
                    {
                        "actuator": am.name,
                        "knob": action.knob,
                        "scope": action.scope,
                        "value": max(am.limits[0], min(action.value, am.limits[1])),
                        "ttl_s": action.ttl_s,
                    }
                )
    return commands
validate_action 
validate_action(action: ControlAction) -> bool

Return True if knob is valid and value is within limits.

Parameters

action : ControlAction The control action.

Returns

bool True if knob is valid and value is within limits.

Source code in src/scpn_phase_orchestrator/actuation/mapper.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def validate_action(self, action: ControlAction) -> bool:
    """Return True if knob is valid and value is within limits.

    Parameters
    ----------
    action : ControlAction
        The control action.

    Returns
    -------
    bool
        True if knob is valid and value is within limits.
    """
    from scpn_phase_orchestrator.binding.types import VALID_KNOBS

    if action.knob not in VALID_KNOBS:
        return False
    if not _finite_real(action.value):
        return False
    mappings = self._by_knob.get(action.knob, [])
    for am in mappings:
        if am.scope == action.scope or action.scope == "global":
            lo, hi = am.limits
            if lo <= action.value <= hi:
                return True
    return False

Action Projector

Safety layer that enforces value bounds and rate limits on control actions before they reach actuators.

Safety requirements

ID Requirement Enforcement
SR-1 Output value within [lo, hi] Value bounds clamp
SR-2 Maximum change ≤ rate_limit per step Rate limit clamp

Constructor

ActionProjector(
    rate_limits: dict[str, float],    # {"K": 0.1, "zeta": 0.05}
    value_bounds: dict[str, tuple[float, float]],  # {"K": (0.0, 1.0)}
)

project()

def project(action: ControlAction, previous_value: float) -> ControlAction

Given previous value v_prev and proposed value v_new:

Δ = v_new - v_prev
Δ_clamped = clamp(Δ, -rate_limit, +rate_limit)
v_projected = clamp(v_prev + Δ_clamped, v_min, v_max)

The returned ControlAction has the same knob, scope, ttl_s, and justification — only value is modified.

Rate limit motivation

Rate limits prevent discontinuous jumps that destabilise the phase dynamics. A coupling strength that jumps from 0.1 to 5.0 in one step can cause the Euler integrator to diverge (CFL violation). The projector ensures smooth transitions.

Consecutive-step guarantee

Over N consecutive steps, the maximum total change is bounded by N × rate_limit. This provides a formal guarantee on the maximum slew rate of any actuated parameter.

Unbounded knobs

Knobs not in rate_limits or value_bounds pass through unmodified. This allows domain-specific knobs to bypass the projector when safety constraints are not applicable.

Performance: project() < 10 μs.

constraints

Projection constraints for bounded supervisor control actions.

ActionProjector is the last deterministic clamp before a control proposal is handed to actuator mapping. It preserves the requested knob/scope metadata and only changes the scalar value, first applying configured absolute bounds and then per-step rate limits relative to the previous actuator value.

Classes

ActionProjector 

ActionProjector(
    rate_limits: dict[str, float],
    value_bounds: dict[str, tuple[float, float]],
)

Clip control actions to value bounds and rate limits.

Rate limits and value bounds are empirical — see docs/ASSUMPTIONS.md § Rate Limits.

Source code in src/scpn_phase_orchestrator/actuation/constraints.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
def __init__(
    self,
    rate_limits: dict[str, float],
    value_bounds: dict[str, tuple[float, float]],
):
    if not isinstance(rate_limits, dict):
        raise TypeError("rate_limits must be a dict[str, float]")
    if not isinstance(value_bounds, dict):
        raise TypeError("value_bounds must be a dict[str, tuple[float, float]]")
    for knob, limit in rate_limits.items():
        if not isinstance(knob, str) or not knob.strip():
            raise ValueError(
                f"rate-limit knob name must be non-empty str, got {knob!r}"
            )
        if isinstance(limit, bool) or not isinstance(limit, Real):
            raise TypeError(
                f"rate limit for {knob!r} must be finite real, got {limit!r}"
            )
        if not isfinite(float(limit)) or float(limit) < 0.0:
            raise ValueError(
                f"rate limit for {knob!r} must be finite >= 0, got {limit!r}"
            )
    for knob, bounds in value_bounds.items():
        if not isinstance(knob, str) or not knob.strip():
            raise ValueError(
                f"value-bound knob name must be non-empty str, got {knob!r}"
            )
        if not isinstance(bounds, tuple) or len(bounds) != 2:
            raise TypeError(
                f"value bounds for {knob!r} must be a 2-tuple "
                f"(lo, hi), got {bounds!r}"
            )
        lo, hi = bounds
        if any(isinstance(v, bool) or not isinstance(v, Real) for v in (lo, hi)):
            raise TypeError(
                f"value bounds for {knob!r} must be finite reals, got {bounds!r}"
            )
        lo_f = float(lo)
        hi_f = float(hi)
        if not isfinite(lo_f) or not isfinite(hi_f):
            raise ValueError(
                f"value bounds for {knob!r} must be finite reals, got {bounds!r}"
            )
        if lo_f > hi_f:
            raise ValueError(
                f"value bounds for {knob!r} require lo <= hi, got {bounds!r}"
            )
    self._rate_limits = rate_limits
    self._value_bounds = value_bounds
Methods:
from_actuator_mappings classmethod 
from_actuator_mappings(
    actuators: Iterable[ActuatorMapping],
) -> ActionProjector

Build projector bounds and slew limits from binding-spec actuators.

ActionProjector is knob-indexed. A binding that maps the same knob to multiple actuator records must therefore provide identical limits and identical rate_limit_per_step values for those records; otherwise the binding is ambiguous and projection fails closed.

Parameters

actuators : Iterable[ActuatorMapping] Actuator mapping declarations.

Returns

ActionProjector Projector bounds and slew limits from binding-spec actuators.

Raises

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

Source code in src/scpn_phase_orchestrator/actuation/constraints.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
@classmethod
def from_actuator_mappings(
    cls,
    actuators: Iterable[ActuatorMapping],
) -> ActionProjector:
    """Build projector bounds and slew limits from binding-spec actuators.

    `ActionProjector` is knob-indexed. A binding that maps the same knob to
    multiple actuator records must therefore provide identical limits and
    identical `rate_limit_per_step` values for those records; otherwise the
    binding is ambiguous and projection fails closed.

    Parameters
    ----------
    actuators : Iterable[ActuatorMapping]
        Actuator mapping declarations.

    Returns
    -------
    ActionProjector
        Projector bounds and slew limits from binding-spec actuators.

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    ValueError
        If the inputs are invalid or inconsistent.
    """
    from scpn_phase_orchestrator.binding.types import ActuatorMapping

    rate_limits: dict[str, float] = {}
    value_bounds: dict[str, tuple[float, float]] = {}
    for actuator in actuators:
        if not isinstance(actuator, ActuatorMapping):
            raise TypeError(
                "actuators must contain ActuatorMapping instances, "
                f"got {actuator!r}"
            )
        bounds = (float(actuator.limits[0]), float(actuator.limits[1]))
        existing_bounds = value_bounds.get(actuator.knob)
        if existing_bounds is not None and existing_bounds != bounds:
            raise ValueError(
                f"conflicting value bounds for actuator knob {actuator.knob!r}"
            )
        value_bounds[actuator.knob] = bounds
        if actuator.rate_limit_per_step is None:
            continue
        rate_limit = float(actuator.rate_limit_per_step)
        existing_rate = rate_limits.get(actuator.knob)
        if existing_rate is not None and existing_rate != rate_limit:
            raise ValueError(
                f"conflicting rate limits for actuator knob {actuator.knob!r}"
            )
        rate_limits[actuator.knob] = rate_limit
    return cls(rate_limits=rate_limits, value_bounds=value_bounds)
project 
project(
    action: ControlAction, previous_value: float
) -> ControlAction

Clamp action value to bounds and rate limit relative to previous_value.

Parameters

action : ControlAction The control action. previous_value : float The previous knob value.

Returns

ControlAction A copy of action whose value is clamped to the knob's absolute bounds and then limited to at most the per-step rate change from previous_value; all other action metadata is preserved.

Raises

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

Source code in src/scpn_phase_orchestrator/actuation/constraints.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
def project(self, action: ControlAction, previous_value: float) -> ControlAction:
    """Clamp action value to bounds and rate limit relative to *previous_value*.

    Parameters
    ----------
    action : ControlAction
        The control action.
    previous_value : float
        The previous knob value.

    Returns
    -------
    ControlAction
        A copy of ``action`` whose value is clamped to the knob's absolute
        bounds and then limited to at most the per-step rate change from
        ``previous_value``; all other action metadata is preserved.

    Raises
    ------
    TypeError
        If an argument has the wrong type.
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if not isinstance(action, ControlAction):
        raise TypeError(f"action must be ControlAction, got {action!r}")
    if not isfinite(float(action.value)):
        raise ValueError(f"action.value must be finite real, got {action.value!r}")
    if isinstance(previous_value, bool) or not isinstance(previous_value, Real):
        raise TypeError(
            f"previous_value must be a finite real scalar, got {previous_value!r}"
        )
    if not isfinite(float(previous_value)):
        raise ValueError(
            f"previous_value must be a finite real scalar, got {previous_value!r}"
        )
    lo, hi = self._value_bounds.get(action.knob, (float("-inf"), float("inf")))
    clamped = max(lo, min(action.value, hi))

    rate_limit = self._rate_limits.get(action.knob)
    if rate_limit is not None:
        delta = clamped - previous_value
        if abs(delta) > rate_limit:
            clamped = previous_value + rate_limit * (1.0 if delta > 0 else -1.0)
        clamped = max(lo, min(clamped, hi))

    return replace(action, value=clamped)

Closed-loop feedback example

from scpn_phase_orchestrator.actuation.constraints import ActionProjector
from scpn_phase_orchestrator.actuation.mapper import ActuationMapper, ControlAction
from scpn_phase_orchestrator.supervisor.policy import SupervisorPolicy
from scpn_phase_orchestrator.supervisor.regimes import RegimeManager
from scpn_phase_orchestrator.upde.engine import UPDEEngine
from scpn_phase_orchestrator.upde.order_params import compute_order_parameter

# Setup
eng = UPDEEngine(n=8, dt=0.01)
pol = SupervisorPolicy(RegimeManager())
proj = ActionProjector(
    rate_limits={"K": 0.1, "zeta": 0.05},
    value_bounds={"K": (0.0, 5.0), "zeta": (0.0, 1.0)},
)

# Feedback loop
k_current = 0.5
zeta_current = 0.0
for _ in range(1000):
    phases = eng.step(phases, omegas, knm, zeta_current, 0.0, alpha)
    r, psi = compute_order_parameter(phases)
    state = build_upde_state(r, psi)
    actions = pol.decide(state, boundary)
    for a in actions:
        if a.knob == "K":
            safe = proj.project(a, previous_value=k_current)
            k_current = safe.value
        elif a.knob == "zeta":
            safe = proj.project(a, previous_value=zeta_current)
            zeta_current = safe.value

Output protocols

The actuation subsystem can drive multiple output protocols:

Protocol Adapter Use case
Modbus/TLS modbus_tls Industrial controllers (PLC, DCS)
gRPC grpc_service Distributed SPO nodes
HTTP/REST server Web dashboard, external APIs
Redis redis_store State persistence, pub/sub
Direct In-process Same-process engine feedback

For in-process use (most common), the actuation output feeds directly back into the next UPDEEngine.step() call without any serialisation overhead.

TTL (time-to-live) semantics

Each ControlAction carries a ttl_s field. The actuation layer tracks active actions and expires them after TTL elapses. This prevents stale control commands from persisting indefinitely if the supervisor stops producing updates.

TTL Meaning
1.0 s Short-lived corrective action
5.0 s Standard policy action
30.0 s Sustained regime response
∞ Permanent override (not recommended)

Safety invariants

The actuation subsystem guarantees:

  1. Bounded output: every actuated value is within [lo, hi]
  2. Bounded rate: |Δv| ≤ rate_limit per step
  3. Monotonic convergence: consecutive project() calls converge toward the proposed value at the rate limit
  4. No side effects: project() is pure — same inputs → same output
  5. Metadata preservation: project() only modifies value, all other ControlAction fields are immutable

Performance summary

Operation Budget Notes
ActuationMapper.map_actions() < 10 μs Dict construction
ActionProjector.project() < 10 μs Two clamp operations
Full closed-loop overhead < 70 μs decide + project + map

CFL stability interaction

The ActionProjector's rate limits interact with the CFL stability condition dt × (max_ω + max_K) < π:

  • If rate_limit_K is too large, a single step could violate CFL
  • The recommended rate limit is rate_limit_K ≤ π/(dt × N) - max_ω/N
  • The numerics module's check_stability() should be called after applying actuation to verify the new K_nm is stable

This is not enforced automatically — the rate limits in the binding spec must be chosen to be CFL-compatible. The docs/ASSUMPTIONS.md file documents the derivation.

Relationship to other subsystems

Subsystem Interaction
Supervisor Produces ControlAction list
Binding Declares ActuatorMapping list
Numerics CFL check after actuation
Audit Logs every actuation command
Engine Consumes modified K_nm, ζ, Ψ

HDL Synthesis Compiler

The KuramotoVerilogCompiler provides a path from high-level topological learning to hard real-time hardware execution. It compiles a stabilized Kuramoto network (\(K_{nm}\), \(\omega\)) directly into structural Verilog code.

Experimental HDL synthesis path

The HDL synthesis path is a research feature that emits structural Verilog from a Kuramoto network. Motivation: a CPU (even the Rust kernel) can introduce OS scheduling jitter, so mapping the integration loop to parallel hardware can, in principle, reduce latency and timing variance.

This path is experimental and unvalidated. It has no field evidence and no safety certification; it must not be treated as a controller for any live system (fusion, medical, grid, or otherwise). Any latency, jitter, or throughput figure is design-dependent and must be measured on real hardware before use — the words "zero jitter" or a specific "nanosecond" latency are not claimed here as facts.

Implementation Details

The compiler generates a structural Verilog module that implements: 1. State Registers: Fixed-point or floating-point registers for each \(\theta_i\). 2. Interaction Matrix: Parallel instantiation of sine-calculators (CORDIC or LUT). 3. Euler Integration: Single-clock cycle updates for the entire manifold.

hdl_compiler

Kuramoto topology to synthesisable Verilog.

Translates a learned coupling matrix K_nm and natural-frequency vector omega into a fixed-point Verilog module that can be compiled by standard FPGA toolchains (Vivado, Quartus). The emitted design instantiates the shared cordic_sincos primitive from spo-kernel/crates/spo-fpga/src/kuramoto_core.v for every non-zero K_ij entry and performs Euler integration in Q16.16 fixed-point.

The compiler is arithmetic-complete — no real, no system tasks, no $sin — so its output is synthesisable rather than simulation-only.

Classes

KuramotoVerilogCompiler 

KuramotoVerilogCompiler(
    n_oscillators: int,
    bit_width: int = _WORD_BITS,
    cordic_stages: int = 4,
)

Compile a Kuramoto topology into synthesisable Verilog.

Each oscillator becomes a Q16.16 state register driven by Euler integration of

dθ_i/dt = ω_i + Σ_j K_ij · sin(θ_j − θ_i)

where the sin() primitive is the cordic_sincos pipelined CORDIC from the FPGA kuramoto_core library. Only non-zero coupling entries materialise in hardware.

Parameters

n_oscillators: Size of the mesh; matches the rows/columns of the coupling matrix. bit_width: Data width in bits (default 32 for Q16.16). cordic_stages: Number of CORDIC rotation stages (default 4). Must match the parameter used on the instantiated cordic_sincos module.

Source code in src/scpn_phase_orchestrator/actuation/hdl_compiler.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def __init__(
    self,
    n_oscillators: int,
    bit_width: int = _WORD_BITS,
    cordic_stages: int = 4,
) -> None:
    n_oscillators = _require_integer(n_oscillators, name="n_oscillators")
    bit_width = _require_integer(bit_width, name="bit_width")
    cordic_stages = _require_integer(cordic_stages, name="cordic_stages")
    if n_oscillators < 1:
        raise ValueError(f"n_oscillators must be >= 1, got {n_oscillators}")
    if bit_width != _WORD_BITS:
        raise ValueError(
            f"only {_WORD_BITS}-bit Q16.16 is supported, got {bit_width}"
        )
    if cordic_stages < 1:
        raise ValueError(f"cordic_stages must be >= 1, got {cordic_stages}")
    self.n = n_oscillators
    self.width = bit_width
    self.cordic_stages = cordic_stages
Methods:
compile 
compile(
    knm: FloatArray, omegas: FloatArray, dt: float
) -> str

Generate a synthesisable Verilog module for the mesh.

The returned string is a complete module suitable for writing to a .v file and feeding to the synthesis step of an FPGA tool. The module depends on cordic_sincos being available on the include path (usually via \`include "kuramoto_core.v" or an IP library).

Parameters

knm : FloatArray Coupling matrix K_nm, shape (N, N). omegas : FloatArray Natural frequencies in rad/s, shape (N,). dt : float Integration step size.

Returns

str A synthesisable Verilog module for the mesh.

Source code in src/scpn_phase_orchestrator/actuation/hdl_compiler.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def compile(self, knm: FloatArray, omegas: FloatArray, dt: float) -> str:
    r"""Generate a synthesisable Verilog module for the mesh.

    The returned string is a complete module suitable for writing to a
    ``.v`` file and feeding to the synthesis step of an FPGA tool.
    The module depends on ``cordic_sincos`` being available on the
    include path (usually via ``\`include "kuramoto_core.v"`` or an
    IP library).

    Parameters
    ----------
    knm : FloatArray
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    omegas : FloatArray
        Natural frequencies in rad/s, shape ``(N,)``.
    dt : float
        Integration step size.

    Returns
    -------
    str
        A synthesisable Verilog module for the mesh.
    """
    knm = _require_finite_array(
        knm,
        name="knm",
        shape=(self.n, self.n),
    )
    omegas = _require_finite_array(omegas, name="omegas", shape=(self.n,))
    dt = _require_positive_finite_float(dt, name="dt")

    dt_q = _hex_q16_16(dt)
    lines: list[str] = []
    lines.append("// Automatically generated by SCPN Phase Orchestrator.")
    lines.append(
        "// Requires cordic_sincos (spo-fpga/kuramoto_core.v) on include path."
    )
    lines.append(f"module kuramoto_mesh_{self.n} #(")
    lines.append(f"    parameter WIDTH = {self.width},")
    lines.append(f"    parameter CORDIC_STAGES = {self.cordic_stages}")
    lines.append(") (")
    lines.append("    input  wire clk,")
    lines.append("    input  wire rst_n,")
    lines.append(f"    output wire signed [WIDTH-1:0] theta_out [0:{self.n - 1}]")
    lines.append(");")
    lines.append(f"    localparam signed [WIDTH-1:0] DT_Q = {dt_q};")

    # Omega constants
    for i in range(self.n):
        lines.append(
            f"    localparam signed [WIDTH-1:0] OMEGA_{i} = "
            f"{_hex_q16_16(float(omegas[i]))};"
        )

    # Non-zero coupling constants (flattened to (i,j) tuples)
    active_edges: list[tuple[int, int, float]] = []
    for i in range(self.n):
        for j in range(self.n):
            if i == j:
                continue
            k_ij = float(knm[i, j])
            if k_ij == 0.0:
                continue
            active_edges.append((i, j, k_ij))

    for i, j, k_ij in active_edges:
        lines.append(
            f"    localparam signed [WIDTH-1:0] K_{i}_{j} = {_hex_q16_16(k_ij)};"
        )

    # State registers + output wiring
    for i in range(self.n):
        lines.append(f"    reg signed [WIDTH-1:0] theta_{i};")
        lines.append(f"    assign theta_out[{i}] = theta_{i};")

    # CORDIC instances for every active edge
    for i, j, _ in active_edges:
        lines.append(f"    wire signed [WIDTH-1:0] diff_{i}_{j};")
        lines.append(f"    assign diff_{i}_{j} = theta_{j} - theta_{i};")
        lines.append(f"    wire signed [WIDTH-1:0] sin_{i}_{j};")
        lines.append(f"    wire signed [WIDTH-1:0] cos_{i}_{j};  // unused")
        lines.append(
            f"    cordic_sincos #(.WIDTH(WIDTH), .STAGES(CORDIC_STAGES)) "
            f"u_cordic_{i}_{j} ("
        )
        lines.append("        .clk(clk),")
        lines.append("        .rst_n(rst_n),")
        lines.append("        .valid_in(1'b1),")
        lines.append(f"        .angle_in(diff_{i}_{j}),")
        lines.append("        .valid_out(/* unused */),")
        lines.append(f"        .sin_out(sin_{i}_{j}),")
        lines.append(f"        .cos_out(cos_{i}_{j})")
        lines.append("    );")

    # Q16.16 multiply helper — combinational widened product + truncation.
    # Each K_ij · sin_ij is computed as a 64-bit product, shifted right
    # by FRAC_BITS to return a Q16.16 value.
    for i, j, _ in active_edges:
        lines.append(f"    wire signed [2*WIDTH-1:0] prod_{i}_{j};")
        lines.append(
            f"    assign prod_{i}_{j} = $signed(K_{i}_{j}) * $signed(sin_{i}_{j});"
        )
        lines.append(f"    wire signed [WIDTH-1:0] term_{i}_{j};")
        lines.append(
            f"    assign term_{i}_{j} = prod_{i}_{j}[2*WIDTH-1:{_FRAC_BITS}];"
        )

    # Aggregate sum per oscillator (ω_i + Σ_j term_ij)
    for i in range(self.n):
        row_terms = [f"OMEGA_{i}"]
        for e_i, e_j, _ in active_edges:
            if e_i == i:
                row_terms.append(f"term_{e_i}_{e_j}")
        lines.append(f"    wire signed [WIDTH-1:0] dtheta_{i};")
        lines.append(f"    assign dtheta_{i} = " + " + ".join(row_terms) + ";")
        lines.append(f"    wire signed [2*WIDTH-1:0] dtheta_dt_{i};")
        lines.append(
            f"    assign dtheta_dt_{i} = $signed(dtheta_{i}) * $signed(DT_Q);"
        )
        lines.append(f"    wire signed [WIDTH-1:0] dtheta_scaled_{i};")
        lines.append(
            f"    assign dtheta_scaled_{i} = dtheta_dt_{i}[2*WIDTH-1:{_FRAC_BITS}];"
        )

    # Sequential Euler update (Q16.16)
    lines.append("    integer reset_i;")
    lines.append("    always @(posedge clk or negedge rst_n) begin")
    lines.append("        if (!rst_n) begin")
    for i in range(self.n):
        lines.append(f"            theta_{i} <= {self.width}'sd0;")
    lines.append("        end else begin")
    for i in range(self.n):
        lines.append(f"            theta_{i} <= theta_{i} + dtheta_scaled_{i};")
    lines.append("        end")
    lines.append("    end")
    lines.append("endmodule")

    return "\n".join(lines)

Verified neural Control Barrier Function safety filter

actuation.control_barrier is a stronger safety layer than the bounds clamp: a Control Barrier Function h(x) defines a safe set S = {x : h(x) ≥ 0}, and the filter admits the supervisor action closest to its proposal that still satisfies the discrete-time CBF condition h(x_{k+1}) ≥ (1 − γ)·h(x_k) under the one-step plant model x_{k+1} ≈ x_k + f + g·u. With the first-order form ∇h(x)·(f + g·u) ≥ −γ·h(x) this is an analytic projection of the nominal control onto a state-dependent half-space, then a clip to the actuator bounds — a constraint derived from the barrier, not a fixed box.

NeuralBarrier is a pure-NumPy ReLU network (no training-framework dependency) exposing value, gradient (reverse-mode), and interval_bounds (sound IBP). ControlBarrierFilter.verify_forward_invariance returns a sound BarrierCertificate: it partitions the state box, bounds h per cell by IBP, and requires that on every boundary-shell cell an actuator-admissible control restores the CBF condition. Because IBP over-approximates h, a passing certificate is never a false guarantee. The certificate now carries both a filter_digest and a verification_digest; runtime callers can prove the certificate belongs to the exact barrier weights, CBF parameters, actuator bounds, control-effect vector, state/drift box, and verifier settings that produced it. Review-only: the filter shapes a proposed action; it never actuates.

control_barrier

Verified neural Control Barrier Function (CBF) safety filter.

A heuristic policy gate clamps actions to bounds; it cannot prove that the admitted action keeps the system inside a safe set. A Control Barrier Function can. A barrier h(x) defines the safe set S = {x : h(x) ≥ 0}; the discrete-time CBF condition

h(x_{k+1}) ≥ (1 − γ) · h(x_k),   γ ∈ (0, 1],

keeps h non-negative once it starts non-negative, so S is forward invariant. The control enters the one-step plant model x_{k+1} ≈ x_k + f + g·u (f the uncontrolled drift, g the per-knob control sensitivity), so the filter admits the action closest to the supervisor's proposal that still satisfies the first-order CBF condition

∇h(x)·(f + g·u) ≥ −γ · h(x),

an analytic projection of the nominal control onto that half-space, then a clip to the actuator bounds. This is strictly stronger than a bounds clamp: it is a state-dependent constraint derived from the barrier, not a fixed box.

The barrier is a neural ReLU network (pure NumPy — no training-framework dependency), and the filter is verified: :func:verify_forward_invariance certifies, soundly, that on the boundary shell {x : 0 ≤ h(x) ≤ shell} an admissible control always restores the CBF condition. The certificate is built by interval bound propagation (IBP, Gowal et al. 2018) over a partition of the state box: IBP over-approximates h on every cell, so a passing certificate can never be a false guarantee — at worst the sound over-approximation refuses to certify a barrier that is in fact valid.

References

  • Ames, Coogan, Egerstedt, Notomista, Sreenath & Tabuada 2019, ECC — control barrier functions: theory and applications.
  • Agrawal & Sreenath 2017, RSS — discrete-time control barrier functions.
  • Gowal et al. 2018, arXiv:1810.12715 — interval bound propagation for verified neural-network bounds.

Classes

NeuralBarrier dataclass 

NeuralBarrier(
    weights: tuple[FloatArray, ...],
    biases: tuple[FloatArray, ...],
)

A ReLU feed-forward neural control barrier function h(x).

The network maps a state vector to a scalar barrier value; the safe set is {x : h(x) ≥ 0}. Hidden layers use ReLU activations and the output layer is linear (so h can take any sign). Weights are supplied at construction (trained or designed elsewhere); this class evaluates, differentiates, and soundly bounds the network.

Attributes

weights : tuple[FloatArray, ...] Per-layer weight matrices, each shape (out, in). biases : tuple[FloatArray, ...] Per-layer bias vectors; biases[i] has length weights[i].shape[0].

Attributes
input_dim property 
input_dim: int

Dimension of the state vector the barrier consumes.

Methods:
value 
value(state: FloatArray) -> float

Return the barrier value h(state) (safe when ≥ 0).

Parameters

state : FloatArray A state vector of length :attr:input_dim.

Returns

float The scalar barrier value.

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def value(self, state: FloatArray) -> float:
    """Return the barrier value ``h(state)`` (safe when ``≥ 0``).

    Parameters
    ----------
    state : FloatArray
        A state vector of length :attr:`input_dim`.

    Returns
    -------
    float
        The scalar barrier value.
    """
    activation = self._validate_state(state)
    for layer, (weight, bias) in enumerate(
        zip(self.weights, self.biases, strict=True)
    ):
        activation = weight @ activation + bias
        if layer < len(self.weights) - 1:
            activation = np.maximum(activation, 0.0)
    return float(activation[0])
gradient 
gradient(state: FloatArray) -> FloatArray

Return ∂h/∂state at state by reverse-mode differentiation.

Parameters

state : FloatArray A state vector of length :attr:input_dim.

Returns

FloatArray The gradient vector, shape (input_dim,).

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def gradient(self, state: FloatArray) -> FloatArray:
    """Return ``∂h/∂state`` at ``state`` by reverse-mode differentiation.

    Parameters
    ----------
    state : FloatArray
        A state vector of length :attr:`input_dim`.

    Returns
    -------
    FloatArray
        The gradient vector, shape ``(input_dim,)``.
    """
    activation = self._validate_state(state)
    pre_masks: list[FloatArray] = []
    for layer, (weight, bias) in enumerate(
        zip(self.weights, self.biases, strict=True)
    ):
        activation = weight @ activation + bias
        if layer < len(self.weights) - 1:
            mask = (activation > 0.0).astype(np.float64)
            pre_masks.append(mask)
            activation = activation * mask
    grad = np.ones(1, dtype=np.float64)
    for layer in range(len(self.weights) - 1, -1, -1):
        grad = self.weights[layer].T @ grad
        if layer > 0:
            grad = grad * pre_masks[layer - 1]
    return np.ascontiguousarray(grad, dtype=np.float64)
interval_bounds 
interval_bounds(
    lower: FloatArray, upper: FloatArray
) -> tuple[float, float]

Return sound [min, max] bounds of h over a state box (IBP).

Interval bound propagation pushes the input interval [lower, upper] through each affine layer (in centre/radius form) and through ReLU, yielding an over-approximation: the true range of h over the box is contained in the returned interval.

Parameters

lower, upper : FloatArray Per-dimension lower and upper bounds of the state box.

Returns

tuple[float, float] A sound (h_min, h_max) enclosure over the box.

Raises

ValueError If the box is malformed or upper < lower in any dimension.

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
def interval_bounds(
    self, lower: FloatArray, upper: FloatArray
) -> tuple[float, float]:
    """Return sound ``[min, max]`` bounds of ``h`` over a state box (IBP).

    Interval bound propagation pushes the input interval ``[lower, upper]``
    through each affine layer (in centre/radius form) and through ReLU,
    yielding an over-approximation: the true range of ``h`` over the box is
    contained in the returned interval.

    Parameters
    ----------
    lower, upper : FloatArray
        Per-dimension lower and upper bounds of the state box.

    Returns
    -------
    tuple[float, float]
        A sound ``(h_min, h_max)`` enclosure over the box.

    Raises
    ------
    ValueError
        If the box is malformed or ``upper < lower`` in any dimension.
    """
    lo = self._validate_state(lower)
    hi = self._validate_state(upper)
    if np.any(hi < lo):
        raise ValueError("interval upper bound must be >= lower bound")
    for layer, (weight, bias) in enumerate(
        zip(self.weights, self.biases, strict=True)
    ):
        centre = (lo + hi) / 2.0
        radius = (hi - lo) / 2.0
        out_centre = weight @ centre + bias
        out_radius = np.abs(weight) @ radius
        lo = out_centre - out_radius
        hi = out_centre + out_radius
        if layer < len(self.weights) - 1:
            lo = np.maximum(lo, 0.0)
            hi = np.maximum(hi, 0.0)
    return float(lo[0]), float(hi[0])

BarrierCertificate dataclass 

BarrierCertificate(
    verified: bool,
    cells_checked: int,
    boundary_cells: int,
    worst_margin: float,
    boundary_shell: float,
    gamma: float,
    filter_digest: str = "",
    verification_digest: str = "",
)

Sound forward-invariance verdict for a CBF filter over a state box.

Attributes

verified : bool Whether every boundary-shell cell admits a control restoring the CBF condition (a sound guarantee — never a false positive). cells_checked : int Number of partition cells inspected. boundary_cells : int Number of cells on the safety boundary (where h may reach 0). worst_margin : float Smallest best_h_next − (1 − γ)·h_upper over boundary cells; ≥ 0 iff verified. inf when no boundary cell exists in the box. boundary_shell : float The boundary-shell half-width used (cells with h_min ≤ shell). gamma : float The CBF decrease rate used. filter_digest : str SHA-256 digest of the exact :class:ControlBarrierFilter configuration the certificate verifies. verification_digest : str SHA-256 digest of the filter digest plus the state/drift boxes and verification parameters used to produce the certificate.

Methods:
to_dict 
to_dict() -> dict[str, bool | int | float | str]

Return a JSON-serialisable mapping of the certificate.

Returns

dict[str, bool | int | float | str] The verdict, cell counts, worst margin, shell width, gamma, and digests binding the certificate to its verified filter/envelope.

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
def to_dict(self) -> dict[str, bool | int | float | str]:
    """Return a JSON-serialisable mapping of the certificate.

    Returns
    -------
    dict[str, bool | int | float | str]
        The verdict, cell counts, worst margin, shell width, gamma, and
        digests binding the certificate to its verified filter/envelope.
    """
    return {
        "verified": self.verified,
        "cells_checked": self.cells_checked,
        "boundary_cells": self.boundary_cells,
        "worst_margin": self.worst_margin,
        "boundary_shell": self.boundary_shell,
        "gamma": self.gamma,
        "filter_digest": self.filter_digest,
        "verification_digest": self.verification_digest,
    }

ControlBarrierFilter dataclass 

ControlBarrierFilter(
    barrier: NeuralBarrier,
    gamma: float,
    control_lo: float,
    control_hi: float,
    control_effect: FloatArray,
)

A CBF-QP safety filter over a single scalar control knob.

Attributes

barrier : NeuralBarrier The neural control barrier function. gamma : float Discrete CBF decrease rate γ ∈ (0, 1]; the barrier may fall by at most a factor γ of its value per step. control_lo, control_hi : float Actuator bounds on the scalar control u. control_effect : FloatArray The per-unit control sensitivity g = ∂x/∂u, shape (input_dim,).

Attributes
filter_digest property 
filter_digest: str

Return a stable SHA-256 digest of the filter configuration.

The digest binds a runtime CBF gate to the exact barrier weights, biases, CBF parameters, control bounds, and control-effect vector that a certificate was generated against.

Methods:
validate_certificate 
validate_certificate(
    certificate: BarrierCertificate,
) -> None

Raise ValueError unless certificate verifies this filter.

Parameters

certificate : BarrierCertificate Forward-invariance certificate generated by :meth:verify_forward_invariance.

Raises

ValueError If the certificate failed, lacks a binding digest, or was generated for a different filter configuration.

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def validate_certificate(self, certificate: BarrierCertificate) -> None:
    """Raise ``ValueError`` unless ``certificate`` verifies this filter.

    Parameters
    ----------
    certificate : BarrierCertificate
        Forward-invariance certificate generated by
        :meth:`verify_forward_invariance`.

    Raises
    ------
    ValueError
        If the certificate failed, lacks a binding digest, or was generated
        for a different filter configuration.
    """
    if not certificate.verified:
        raise ValueError("barrier_certificate must be verified")
    if certificate.filter_digest == "":
        raise ValueError("barrier_certificate must carry a filter_digest")
    if certificate.filter_digest != self.filter_digest:
        raise ValueError("barrier_certificate does not match barrier_filter")
    if certificate.gamma != self.gamma:
        raise ValueError("barrier_certificate gamma does not match barrier_filter")
filter 
filter(
    nominal_control: float,
    state: FloatArray,
    drift: FloatArray,
) -> tuple[float, bool]

Return the safe control nearest the nominal one, and whether it changed.

Parameters

nominal_control : float The supervisor's proposed scalar control u_nom. state : FloatArray The current state vector. drift : FloatArray The uncontrolled one-step state change f (same shape as state).

Returns

tuple[float, bool] (safe_control, intervened) — the admitted control clipped to the actuator bounds, and whether it differs from the (bound-clipped) nominal control.

Raises

ValueError If inputs are malformed.

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
def filter(
    self,
    nominal_control: float,
    state: FloatArray,
    drift: FloatArray,
) -> tuple[float, bool]:
    """Return the safe control nearest the nominal one, and whether it changed.

    Parameters
    ----------
    nominal_control : float
        The supervisor's proposed scalar control ``u_nom``.
    state : FloatArray
        The current state vector.
    drift : FloatArray
        The uncontrolled one-step state change ``f`` (same shape as state).

    Returns
    -------
    tuple[float, bool]
        ``(safe_control, intervened)`` — the admitted control clipped to the
        actuator bounds, and whether it differs from the (bound-clipped)
        nominal control.

    Raises
    ------
    ValueError
        If inputs are malformed.
    """
    u_nom = _as_finite_float(nominal_control, name="nominal_control")
    state_vec = self.barrier._validate_state(state)
    drift_vec = _as_float_vector(drift, name="drift")
    if drift_vec.shape[0] != self.barrier.input_dim:
        raise ValueError("drift dimension must match barrier input")

    grad = self.barrier.gradient(state_vec)
    h = self.barrier.value(state_vec)
    # First-order discrete CBF: grad·(drift + g·u) >= -gamma·h
    #   => (grad·g)·u >= -gamma·h - grad·drift
    lie_g = float(grad @ self.control_effect)
    rhs = -self.gamma * h - float(grad @ drift_vec)

    clipped_nom = min(self.control_hi, max(self.control_lo, u_nom))
    feasible = self._project(clipped_nom, lie_g, rhs)
    intervened = not np.isclose(feasible, clipped_nom, rtol=0.0, atol=1e-12)
    return feasible, intervened
verify_forward_invariance 
verify_forward_invariance(
    state_lo: FloatArray,
    state_hi: FloatArray,
    drift_lo: FloatArray,
    drift_hi: FloatArray,
    *,
    cells_per_axis: int = 16,
    boundary_shell: float = 0.25,
) -> BarrierCertificate

Soundly certify forward invariance of the safe set over a state box.

The box is partitioned into cells_per_axis cells per dimension. For each cell, IBP bounds h over the cell; a cell is on the boundary when its lower bound is ≤ shell and its range straddles or approaches 0. For boundary cells, the next-state box x + f + g·u is formed by interval arithmetic (worst-case drift [drift_lo, drift_hi]) for each control endpoint, IBP bounds h on it, and the best endpoint must keep h_next_lower ≥ (1 − γ)·h_upper. Because IBP over-approximates, a verified certificate is sound.

Parameters

state_lo, state_hi : FloatArray The state box to certify. drift_lo, drift_hi : FloatArray Worst-case interval enclosure of the uncontrolled drift f. cells_per_axis : int Partition resolution per dimension (>= 1). boundary_shell : float Boundary-shell half-width; only cells reaching h ≤ shell are checked for the CBF condition.

Returns

BarrierCertificate The sound forward-invariance verdict.

Raises

ValueError If inputs are malformed.

Source code in src/scpn_phase_orchestrator/actuation/control_barrier.py 
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
def verify_forward_invariance(
    self,
    state_lo: FloatArray,
    state_hi: FloatArray,
    drift_lo: FloatArray,
    drift_hi: FloatArray,
    *,
    cells_per_axis: int = 16,
    boundary_shell: float = 0.25,
) -> BarrierCertificate:
    """Soundly certify forward invariance of the safe set over a state box.

    The box is partitioned into ``cells_per_axis`` cells per dimension. For
    each cell, IBP bounds ``h`` over the cell; a cell is on the boundary when
    its lower bound is ``≤ shell`` and its range straddles or approaches 0.
    For boundary cells, the next-state box ``x + f + g·u`` is formed by
    interval arithmetic (worst-case drift ``[drift_lo, drift_hi]``) for each
    control endpoint, IBP bounds ``h`` on it, and the best endpoint must keep
    ``h_next_lower ≥ (1 − γ)·h_upper``. Because IBP over-approximates, a
    ``verified`` certificate is sound.

    Parameters
    ----------
    state_lo, state_hi : FloatArray
        The state box to certify.
    drift_lo, drift_hi : FloatArray
        Worst-case interval enclosure of the uncontrolled drift ``f``.
    cells_per_axis : int
        Partition resolution per dimension (``>= 1``).
    boundary_shell : float
        Boundary-shell half-width; only cells reaching ``h ≤ shell`` are
        checked for the CBF condition.

    Returns
    -------
    BarrierCertificate
        The sound forward-invariance verdict.

    Raises
    ------
    ValueError
        If inputs are malformed.
    """
    lo = self.barrier._validate_state(state_lo)
    hi = self.barrier._validate_state(state_hi)
    d_lo = _as_float_vector(drift_lo, name="drift_lo")
    d_hi = _as_float_vector(drift_hi, name="drift_hi")
    if d_lo.shape[0] != self.barrier.input_dim or d_hi.shape[0] != lo.shape[0]:
        raise ValueError("drift bounds must match barrier input dimension")
    if np.any(hi < lo):
        raise ValueError("state_hi must be >= state_lo")
    if np.any(d_hi < d_lo):
        raise ValueError("drift_hi must be >= drift_lo")
    if isinstance(cells_per_axis, bool) or not isinstance(cells_per_axis, Integral):
        raise ValueError("cells_per_axis must be a positive integer")
    if cells_per_axis < 1:
        raise ValueError("cells_per_axis must be >= 1")
    boundary_shell_value = _as_finite_float(boundary_shell, name="boundary_shell")
    if boundary_shell_value < 0.0:
        raise ValueError("boundary_shell must be non-negative")

    dim = lo.shape[0]
    edges = [np.linspace(lo[i], hi[i], cells_per_axis + 1) for i in range(dim)]
    worst_margin = float("inf")
    cells_checked = 0
    boundary_cells = 0
    verified = True

    for cell_index in np.ndindex(*([cells_per_axis] * dim)):
        cell_lo = np.array([edges[i][cell_index[i]] for i in range(dim)])
        cell_hi = np.array([edges[i][cell_index[i] + 1] for i in range(dim)])
        cells_checked += 1
        h_min, h_max = self.barrier.interval_bounds(cell_lo, cell_hi)
        # Inside the safe set and away from the boundary: nothing to enforce.
        if h_min > boundary_shell_value:
            continue
        # Entirely outside the safe set: not part of S, skip.
        if h_max < 0.0:
            continue
        boundary_cells += 1
        cell_margin = self._cell_margin(cell_lo, cell_hi, d_lo, d_hi, h_max)
        worst_margin = min(worst_margin, cell_margin)
        if cell_margin < 0.0:
            verified = False

    filter_digest = self.filter_digest
    verification_digest = _sha256_json(
        {
            "schema": "scpn_phase_orchestrator.control_barrier_verification.v1",
            "filter_digest": filter_digest,
            "state_lo": _array_payload(lo),
            "state_hi": _array_payload(hi),
            "drift_lo": _array_payload(d_lo),
            "drift_hi": _array_payload(d_hi),
            "cells_per_axis": int(cells_per_axis),
            "boundary_shell": boundary_shell_value,
        }
    )
    return BarrierCertificate(
        verified=verified,
        cells_checked=cells_checked,
        boundary_cells=boundary_cells,
        worst_margin=worst_margin,
        boundary_shell=boundary_shell_value,
        gamma=self.gamma,
        filter_digest=filter_digest,
        verification_digest=verification_digest,
    )

Foundation-model governor

actuation.foundation_model_governor is the harness that makes an external controller — a foundation-model forecaster, a learned policy, any advisory source SPO does not trust — deployable. FoundationModelGovernor.govern takes the proposal as an advisory scalar control and admits only a safe action by composing the trust stack SPO already owns: actuator bounds (clamp), a rate limit against the last admitted action, an optional certified Control Barrier Function projection (forward-invariance, with the state-left-the-safe-set case flagged when h(x) < 0), and any number of named safety predicates (an STL-derived check, an operating-envelope rule) that veto the action. Supplying a CBF filter without a verified matching BarrierCertificate is rejected at governor construction, so the runtime path cannot silently use an uncertified or stale neural barrier. Each call returns a GovernorDecision recording the admitted action, the status (admitted / constrained / rejected), the ordered envelope stages that touched the proposal, the violations, the barrier value, and a canonical-JSON SHA-256 seal — the same hashing the assurance bundle uses, so the governance record is tamper-evident.

The governor competes on governance, not prediction: it never forecasts, and it is review-only — it returns a safe action and a decision, it never actuates a plant. This is the runtime embodiment of EU AI Act Art. 14 (human oversight) and Art. 12 (logging / traceability).

foundation_model_governor

Govern an externally-proposed control through SPO's safety envelope.

A foundation model (a Panda-class forecaster, a learned policy, any external controller) may out-predict SPO's own observer, but it offers no safety guarantee, no bound on its output, and no audit trail. This module is the harness that makes such an advisory proposal deployable: :class:FoundationModelGovernor takes the proposal as an advisory scalar control and admits only a safe action, by running it through the trust stack SPO already owns —

  1. actuator bounds — clamp to [control_lo, control_hi];
  2. rate limit — bound the step against the last admitted action (|u − u_prev| ≤ max_rate);
  3. Control Barrier Function — project through an optional, certified :class:~scpn_phase_orchestrator.actuation.control_barrier.ControlBarrierFilter so the admitted action keeps the system inside the certified forward-invariant safe set, and flag when the state has already left it (h(x) < 0);
  4. safety predicates — veto the action if any supplied predicate (an STL-derived check, an operating-envelope rule, …) rejects it.

Every decision is sealed into a content-addressed :class:GovernorDecision (the same canonical-JSON SHA-256 the assurance bundle uses), so the governance record is tamper-evident and the chain of which envelope stages touched the proposal is explicit. The governor competes on governance, not prediction: it never forecasts and it is review-only — it returns a safe action and a decision; it never actuates a plant.

References

  • EU AI Act 2024/1689 Art. 14 (human oversight) and Art. 12 (logging / traceability) — the review-only, audited posture this envelope implements.

Classes

GovernorDecision dataclass 

GovernorDecision(
    proposed_action: float,
    admitted_action: float,
    status: str,
    stages_applied: tuple[str, ...],
    violations: tuple[str, ...],
    barrier_value: float | None,
)

The audited outcome of governing one proposed control.

Attributes

proposed_action : float The advisory action as received from the external source. admitted_action : float The safe action the governor admits (the reviewed output). status : str :data:ADMITTED, :data:CONSTRAINED, or :data:REJECTED. stages_applied : tuple[str, ...] Envelope stages that modified the proposal, in order (bounds, rate_limit, cbf). violations : tuple[str, ...] Reasons the action was rejected; empty unless status is :data:REJECTED. barrier_value : float | None The barrier value h(state) when a Control Barrier Function is configured, otherwise None; a negative value means the state has left the certified safe set. content_hash : str SHA-256 of the canonical decision record (excluding this field); computed on construction.

Methods:
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the decision.

Returns

dict[str, object] The canonical payload plus the computed content_hash.

Source code in src/scpn_phase_orchestrator/actuation/foundation_model_governor.py 
137
138
139
140
141
142
143
144
145
146
147
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the decision.

    Returns
    -------
    dict[str, object]
        The canonical payload plus the computed ``content_hash``.
    """
    record = self._canonical_payload()
    record["content_hash"] = self.content_hash
    return record

FoundationModelGovernor dataclass 

FoundationModelGovernor(
    control_lo: float,
    control_hi: float,
    max_rate: float,
    barrier_filter: ControlBarrierFilter | None = None,
    barrier_certificate: BarrierCertificate | None = None,
    safety_predicates: tuple[
        tuple[str, SafetyPredicate], ...
    ] = (),
    hold_on_reject: bool = True,
)

Admit an externally-proposed scalar control through the safety envelope.

Attributes

control_lo : float Lower actuator bound. control_hi : float Upper actuator bound (> control_lo). max_rate : float Maximum admitted change per call, |u − u_prev| (> 0). barrier_filter : ControlBarrierFilter | None Optional Control Barrier Function gate; None skips the CBF stage. Supplying a filter also requires a verified matching barrier_certificate. barrier_certificate : BarrierCertificate | None Verified forward-invariance certificate generated for barrier_filter. Runtime construction fails closed when the certificate is missing, failed, or bound to a different filter digest. safety_predicates : tuple[tuple[str, SafetyPredicate], ...] Named (label, predicate) safety checks run on the candidate action; any predicate returning ok=False rejects the action. hold_on_reject : bool On rejection, hold the previous action (True) or fall back to the bound-clamped neutral action 0 (False).

Methods:
govern 
govern(
    proposed_action: float,
    state: FloatArray,
    drift: FloatArray,
    *,
    previous_action: float = 0.0,
) -> GovernorDecision

Govern one advisory control proposal and return an audited decision.

Parameters

proposed_action : float The advisory action from the external source (e.g. a foundation model), in actuator units. state : FloatArray Current system state passed to the Control Barrier Function and the safety predicates. drift : FloatArray Uncontrolled state drift f(x) passed to the Control Barrier Function. previous_action : float The last admitted action, used for the rate limit and as the rejection fallback when hold_on_reject is set.

Returns

GovernorDecision The admitted action, status, applied stages, any violations, the barrier value, and a sealing hash.

Raises

ValueError If the proposal, previous action, or state arrays are not finite reals of the expected shape.

Source code in src/scpn_phase_orchestrator/actuation/foundation_model_governor.py 
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
def govern(
    self,
    proposed_action: float,
    state: FloatArray,
    drift: FloatArray,
    *,
    previous_action: float = 0.0,
) -> GovernorDecision:
    """Govern one advisory control proposal and return an audited decision.

    Parameters
    ----------
    proposed_action : float
        The advisory action from the external source (e.g. a foundation
        model), in actuator units.
    state : FloatArray
        Current system state passed to the Control Barrier Function and the
        safety predicates.
    drift : FloatArray
        Uncontrolled state drift ``f(x)`` passed to the Control Barrier
        Function.
    previous_action : float
        The last admitted action, used for the rate limit and as the
        rejection fallback when ``hold_on_reject`` is set.

    Returns
    -------
    GovernorDecision
        The admitted action, status, applied stages, any violations, the
        barrier value, and a sealing hash.

    Raises
    ------
    ValueError
        If the proposal, previous action, or state arrays are not finite
        reals of the expected shape.
    """
    proposed = _real_scalar(proposed_action, "proposed_action")
    previous = _real_scalar(previous_action, "previous_action")
    state_vec = _finite_vector(state, "state")
    drift_vec = _finite_vector(drift, "drift")

    stages: list[str] = []
    action = proposed
    clamped = min(max(action, self.control_lo), self.control_hi)
    if clamped != action:
        stages.append(_BOUNDS)
        action = clamped
    limited = self._rate_limit(action, previous)
    if limited != action:
        stages.append(_RATE_LIMIT)
        action = limited

    violations: list[str] = []
    barrier_value = self._apply_barrier(action, state_vec, drift_vec, stages)
    if barrier_value is not None:
        action = barrier_value[0]
        if barrier_value[1] < 0.0:
            violations.append("barrier: state outside certified safe set (h<0)")

    violations.extend(self._predicate_violations(action, state_vec))
    barrier = None if barrier_value is None else barrier_value[1]
    return self._decide(proposed, action, previous, stages, violations, barrier)