Supervisor¶
The supervisor subsystem adds a regime-classification and control-proposal layer over oscillator dynamics. It classifies configured regimes, can generate model-based risk estimates, and proposes bounded corrections for review—it does not establish domain-event prediction or close a control loop on hardware.
Pipeline position¶
UPDEEngine.step() ──→ phases ──→ compute_order_parameter()
│
↓
UPDEState (R, ψ, locks)
│
┌─────────────────┼─────────────────┐
↓ ↓ ↓
RegimeManager PetriNetAdapter PredictiveSupervisor
│ │ │
└─────────┬───────┘ │
↓ ↓
SupervisorPolicy.decide() ←──────┘
│
├──→ CausalInterventionEngine
│ (baseline vs intervention rollout)
│
↓
list[ControlAction]
│
↓
ActionProjector.project()
│
↓
ActuationMapper.map_actions()
The supervisor sits between the engine output and the next engine step.
It consumes UPDEState and BoundaryState, produces ControlAction
instructions that modify K_nm, ζ, Ψ, or ω for the next step.
Regime Manager¶
Finite state machine for synchronisation regimes with hysteresis, cooldown, and event logging.
Regime enum¶
| Value | Meaning | R range (default thresholds) |
|---|---|---|
NOMINAL |
Healthy synchronisation | R ≥ 0.6 |
DEGRADED |
Partial desynchronisation | 0.3 ≤ R < 0.6 |
CRITICAL |
Synchronisation failure | R < 0.3 or hard violation |
RECOVERY |
Transitioning from CRITICAL | CRITICAL → R improving |
Safety requirement SR-3¶
CRITICAL must pass through RECOVERY before reaching NOMINAL. Direct CRITICAL → NOMINAL is forbidden. This prevents premature resumption of normal operation after a synchronisation failure.
Constructor¶
RegimeManager(
hysteresis: float = 0.05, # band around thresholds
cooldown_steps: int = 10, # steps before next transition
event_bus: EventBus | None = None,
hysteresis_hold_steps: int = 0, # consecutive proposals needed
)
Methods¶
| Method | Signature | Description |
|---|---|---|
evaluate |
(UPDEState, BoundaryState) → Regime |
Proposes regime from metrics |
transition |
(Regime) → Regime |
Applies FSM rules, returns actual |
force_transition |
(Regime) → Regime |
Bypasses cooldown |
Hysteresis¶
To prevent oscillation between regimes when R is near a threshold, the manager applies a hysteresis band:
NOMINAL → DEGRADED: requires R < threshold - hysteresis
DEGRADED → NOMINAL: requires R > threshold + hysteresis
hysteresis_hold_steps adds an additional guard: the proposed regime
must be proposed for N consecutive steps before the transition fires.
CRITICAL always bypasses this hold (safety override).
Cooldown¶
After a transition, subsequent non-CRITICAL transitions are blocked
for cooldown_steps evaluations. CRITICAL always bypasses cooldown.
Transition history¶
transition_history: deque[tuple[int, Regime, Regime]] stores the
last 100 transitions as (step_number, old_regime, new_regime).
Performance: evaluate() < 10 μs.
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
Attributes¶
current_regime
property
¶
The regime established after the most recent transition.
Returns¶
Regime The regime established after the most recent transition.
Methods:¶
evaluate ¶
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
transition ¶
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
force_transition ¶
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
Higher-Order Topology Adaptation¶
HigherOrderTopologySupervisor is the first supervisor-side topology editor.
It consumes live phases plus the current pairwise K_nm matrix and returns a
next-step topology:
- bounded pairwise coupling updates from local phase alignment
- optional triadic
Hyperedgeproposals when global coherence is below target - pruning of stale or incoherent higher-order edges
- serialisable audit metadata for added/pruned simplices and pairwise delta norm
The core control knob is TopologyMutationPolicy.mutation_rate. A value of
0.0 freezes topology; larger values increase the maximum per-step pairwise
and triadic changes while preserving non-negative couplings and a zero
diagonal. TopologyMutationPolicy.simplex_pairwise_support_floor is the
policy-hardening gate for deployment reviews: a candidate 2-simplex is only
created when every pairwise edge inside that triad is already at or above the
configured support floor.
import numpy as np
from scpn_phase_orchestrator.supervisor import (
HigherOrderTopologySupervisor,
TopologyMutationPolicy,
)
from scpn_phase_orchestrator.upde.hypergraph import HypergraphEngine
policy = TopologyMutationPolicy(mutation_rate=0.2, coherence_floor=0.8)
topology = HigherOrderTopologySupervisor(policy)
result = topology.mutate(phases, knm)
engine = HypergraphEngine(len(phases), dt=0.01, hyperedges=list(result.hyperedges))
next_phases = engine.step(phases, omegas, pairwise_knm=result.knm)
audit_payload = result.to_audit_record()
This slice does not claim autonomous online structural control. It provides the auditable mutation primitive that existing policy, causal, STL, simplicial, and hypergraph paths can gate before applying a topology change.
Domainpack demos:
domainpacks/plasma_control/topology_adaptation_demo.pyruns one guarded mutation against the plasma-control binding and prints the audit payload as JSON.domainpacks/traffic_flow/topology_adaptation_demo.pybuilds pairwise support from transfer-entropy evidence before proposing traffic-corridor simplices, then records Lyapunov before/after energy and basin evidence for the proposed mutation.domainpacks/network_security/topology_adaptation_demo.pybuilds pairwise support from transfer-entropy evidence before proposing traffic/attack/defence simplices, then records Lyapunov before/after energy evidence for the proposed mutation.
topology ¶
Supervisor-side higher-order topology mutation utilities.
The functions here do not replace the UPDE, simplicial, or hypergraph
engines. They prepare the next-step coupling topology from live phase
evidence so an existing engine can consume pairwise K_nm and optional
triadic hyperedges.
Classes¶
TopologyMutationPolicy
dataclass
¶
TopologyMutationPolicy(
mutation_rate: float = 0.1,
coherence_floor: float = 0.75,
pairwise_threshold: float = 0.85,
simplex_threshold: float = 0.9,
max_pairwise_delta: float = 0.05,
max_simplex_strength: float = 0.2,
max_new_simplices: int = 4,
prune_threshold: float = 0.2,
simplex_pairwise_support_floor: float = 0.0,
max_coupling: float = 10.0,
)
Policy knobs for one topology mutation step.
mutation_rate is the main supervisor knob: zero freezes topology;
one applies the maximum allowed per-step pairwise and triadic changes.
TopologyMutationResult
dataclass
¶
TopologyMutationResult(
knm: FloatArray,
hyperedges: tuple[Hyperedge, ...],
added_simplices: tuple[Hyperedge, ...],
pruned_simplices: tuple[Hyperedge, ...],
pairwise_delta_norm: float,
global_coherence: float,
)
Result of a supervisor topology mutation step.
Methods:¶
to_audit_record ¶
Return a serialisable audit payload for topology mutation.
Returns¶
dict[str, object] Return a serialisable audit payload for topology mutation.
Source code in src/scpn_phase_orchestrator/supervisor/topology.py
HigherOrderTopologySupervisor ¶
Edit pairwise and triadic topology from live phase evidence.
Source code in src/scpn_phase_orchestrator/supervisor/topology.py
Methods:¶
mutate ¶
mutate(
phases: FloatArray,
knm: FloatArray,
hyperedges: tuple[Hyperedge, ...] | None = None,
) -> TopologyMutationResult
Return a mutated topology for the next supervisor actuation step.
Parameters¶
phases : FloatArray
Oscillator phases in radians, shape (N,).
knm : FloatArray
Coupling matrix K_nm, shape (N, N).
hyperedges : tuple[Hyperedge, ...] | None
Existing hyperedges, or None.
Returns¶
TopologyMutationResult The mutated topology for the next actuation step.
Source code in src/scpn_phase_orchestrator/supervisor/topology.py
Functions:¶
Hierarchical Orchestration Summaries¶
build_hierarchical_orchestration_plan() is the generic nested-supervisor
foundation. Child supervisors exchange bounded summaries only: child name,
channel, R, psi, regime, confidence, and optional metadata. The parent
planner converts those summaries into a reduced UPDEState, computes
cross-child phase alignment, and emits escalation records for low confidence,
degraded coherence, critical coherence, or explicit child-regime escalation.
from scpn_phase_orchestrator.supervisor import (
ChildSupervisorSummary,
build_hierarchical_orchestration_plan,
)
plan = build_hierarchical_orchestration_plan(
[
ChildSupervisorSummary("edge-a", "power", R=0.9, psi=0.0),
ChildSupervisorSummary("edge-b", "thermal", R=0.5, psi=1.2),
],
degraded_threshold=0.65,
critical_threshold=0.35,
)
parent_state = plan.parent_state
audit_payload = plan.to_audit_record()
The same reduced summaries can be wrapped in deterministic sync envelopes for JSONL replay, message-bus transport, or parent-side cloud ingestion. The parent ingestion helper rejects stale or duplicate sequence numbers per source node and protocol-version mismatches before building the parent orchestration plan. Direct envelope JSON parsing uses canonical finite JSON semantics: non-finite constants and duplicate object keys are rejected before the reduced summary is validated or admitted to the parent watermark ledger.
from scpn_phase_orchestrator.supervisor import (
build_hierarchy_sync_envelope,
ingest_hierarchy_sync_envelopes,
)
envelope = build_hierarchy_sync_envelope(
ChildSupervisorSummary("edge-a", "power", R=0.9, psi=0.0),
source_node="edge-node-a",
sequence=42,
)
ledger = ingest_hierarchy_sync_envelopes(
[envelope],
previous_sequences={"edge-node-a": 41},
)
sync_audit = ledger.to_audit_record()
HierarchyTransportRuntime is the next live-transport boundary. Caller-owned
REST, gRPC, Kafka, file, or hardware adapters can pass decoded mappings or JSON
strings into the runtime; the runtime parses reduced sync records, maintains
per-source sequence watermarks across batches, and emits the same parent
ledger. It still owns no socket, thread, broker client, or actuator handle.
from scpn_phase_orchestrator.supervisor import HierarchyTransportRuntime
runtime = HierarchyTransportRuntime()
batch_ledger = runtime.ingest_batch([envelope.to_json()])
runtime_audit = runtime.to_audit_record()
For offline distributed-edge testing, simulate_hierarchy_gossip_consensus()
replays local consensus over accepted sync envelopes and a caller-supplied
neighbour map. Each node updates only its reduced coherence, phase, confidence,
and audit metadata; no sockets are opened and no raw observations enter the
consensus state.
from scpn_phase_orchestrator.supervisor import simulate_hierarchy_gossip_consensus
rounds = simulate_hierarchy_gossip_consensus(
[envelope],
neighbour_map={"edge-node-a": ()},
rounds=1,
)
consensus_audit = [round_record.to_audit_record() for round_record in rounds]
This slice does not open sockets, run a gossip protocol, or perform direct actuation. It gives existing regime, policy, FEP, causal, STL, and audit paths a common parent-level state built from reduced child evidence without moving raw time series, local coupling matrices, or actuator targets across hierarchy boundaries.
Domainpack demos:
domainpacks/power_grid/hierarchy_sync_demo.pyreplays generation and demand/renewable edge summaries through the sync-envelope ingestion path.domainpacks/cardiac_rhythm/hierarchy_sync_demo.pyreplays pacemaker/atrial and ventricular/recovery summaries through the same parent planner.
hierarchy ¶
Reduced-evidence hierarchy summaries, envelopes, ledgers, and consensus.
The hierarchy package enforces a boundary where parent supervisors receive only bounded child summaries: coherence, phase, regime, confidence, channel, and metadata. Raw phases, time series, coupling matrices, event payloads, and actuator targets are rejected from metadata and transport envelopes. The boundary types and validation live in one core module, with the orchestration plan, sync transport, and gossip consensus split into their own modules behind a stable re-export surface. Builders and runtimes are socket-free and return audit-ready plans or ledgers.
Classes¶
ChildSupervisorSummary
dataclass
¶
ChildSupervisorSummary(
name: str,
channel: str,
R: float,
psi: float,
regime: str = _REGIME_NOMINAL,
confidence: float = 1.0,
metadata: Mapping[str, object] = dict(),
)
Bounded child-supervisor evidence for parent orchestration.
The summary intentionally carries reduced coherence evidence only. Raw child phases, time series, local coupling matrices, and actuator targets do not cross the hierarchy boundary in this foundation slice.
Attributes¶
weighted_R
property
¶
Return coherence weighted by summary confidence.
Returns¶
float Return coherence weighted by summary confidence.
Methods:¶
to_audit_record ¶
Return a JSON-safe reduced child summary.
Returns¶
dict[str, object] Return a JSON-safe reduced child summary.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/boundary.py
HierarchyEscalation
dataclass
¶
HierarchyEscalation(
child: str,
channel: str,
severity: str,
reason: str,
R: float,
confidence: float,
child_regime: str,
)
Bounded evidence escalated from a child to the parent supervisor.
Methods:¶
to_audit_record ¶
Return a JSON-safe escalation record.
Returns¶
dict[str, object] Return a JSON-safe escalation record.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/boundary.py
HierarchySyncEnvelope
dataclass
¶
HierarchySyncEnvelope(
protocol_version: str,
source_node: str,
sequence: int,
summary: ChildSupervisorSummary,
monotonic_time_s: float | None = None,
)
Transport-neutral hierarchy summary exchanged by edge/cloud nodes.
Methods:¶
to_audit_record ¶
Return a JSON-safe transport envelope audit record.
Returns¶
dict[str, object] Return a JSON-safe transport envelope audit record.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/boundary.py
HierarchyConsensusRound
dataclass
¶
HierarchyConsensusRound(
round_index: int,
states: tuple[HierarchyConsensusState, ...],
plan: HierarchicalOrchestrationPlan,
rejected: tuple[dict[str, object], ...] = (),
)
Deterministic non-networked gossip/local-consensus replay result.
Methods:¶
to_audit_record ¶
Return a JSON-safe consensus-round audit record.
Returns¶
dict[str, object] Return a JSON-safe consensus-round audit record.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/consensus.py
HierarchyConsensusState
dataclass
¶
HierarchicalOrchestrationPlan
dataclass
¶
HierarchicalOrchestrationPlan(
hierarchy: str,
children: tuple[ChildSupervisorSummary, ...],
parent_state: UPDEState,
escalations: tuple[HierarchyEscalation, ...],
parent_R: float,
parent_psi: float,
audit_scope: str = _AUDIT_SCOPE_REDUCED_SUMMARIES,
)
Parent orchestration input built from reduced child summaries.
Methods:¶
to_audit_record ¶
Return a serialisable plan record for hierarchy audit logs.
Returns¶
dict[str, object] Return a serialisable plan record for hierarchy audit logs.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/plan.py
HierarchySyncLedger
dataclass
¶
HierarchySyncLedger(
accepted: tuple[HierarchySyncEnvelope, ...],
rejected: tuple[dict[str, object], ...],
plan: HierarchicalOrchestrationPlan,
)
Parent-side ingestion result for sync envelopes.
Methods:¶
to_audit_record ¶
Return a serialisable sync-ingestion audit payload.
Returns¶
dict[str, object] Return a serialisable sync-ingestion audit payload.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
HierarchyTransportRuntime ¶
HierarchyTransportRuntime(
*,
previous_sequences: Mapping[str, int] | None = None,
hierarchy: str = "edge_cloud_summary_sync",
degraded_threshold: float = 0.65,
critical_threshold: float = 0.35,
min_confidence: float = 0.5,
protocol_version: str = _DEFAULT_HIERARCHY_SYNC_PROTOCOL,
)
Socket-free runtime state for hierarchy transport adapters.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
Attributes¶
previous_sequences
property
¶
Return the accepted per-source sequence watermarks.
Returns¶
dict[str, int] Return the accepted per-source sequence watermarks.
Methods:¶
ingest ¶
ingest(
records: Sequence[
HierarchySyncEnvelope | Mapping[str, object] | str
],
) -> HierarchySyncLedger
Parse a transport batch, ingest it, and advance accepted watermarks.
Parameters¶
records : Sequence[HierarchySyncEnvelope | Mapping[str, object] | str] The transport records to ingest.
Returns¶
HierarchySyncLedger The sync ledger with advanced watermarks.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
ingest_batch ¶
ingest_batch(
records: Sequence[
HierarchySyncEnvelope | Mapping[str, object] | str
],
) -> HierarchySyncLedger
Alias for adapter batch ingestion.
Parameters¶
records : Sequence[HierarchySyncEnvelope | Mapping[str, object] | str] The transport records to ingest.
Returns¶
HierarchySyncLedger The sync ledger for the ingested batch.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
to_audit_record ¶
Return socket-free runtime state for audit logging.
Returns¶
dict[str, object] Return socket-free runtime state for audit logging.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
Functions:¶
simulate_hierarchy_gossip_consensus ¶
simulate_hierarchy_gossip_consensus(
envelopes: Sequence[HierarchySyncEnvelope],
*,
neighbour_map: Mapping[str, Sequence[str]],
rounds: int = 1,
self_weight: float = 0.5,
hierarchy: str = "offline_hierarchy_gossip_consensus",
previous_sequences: Mapping[str, int] | None = None,
degraded_threshold: float = 0.65,
critical_threshold: float = 0.35,
min_confidence: float = 0.5,
protocol_version: str = _DEFAULT_HIERARCHY_SYNC_PROTOCOL,
) -> tuple[HierarchyConsensusRound, ...]
Replay local consensus over hierarchy sync envelopes without networking.
Each round updates every accepted node from its own reduced summary and the summaries of configured neighbours. The update averages confidence-weighted coherence and circular phase only; raw child observations never enter the consensus state. This is a deterministic simulation surface for testing distributed orchestration policies before any live gossip transport exists.
Parameters¶
envelopes : Sequence[HierarchySyncEnvelope]
The ordered transport envelopes.
neighbour_map : Mapping[str, Sequence[str]]
Per-node neighbour lists for gossip.
rounds : int
Number of gossip rounds.
self_weight : float
Self-weight in the gossip consensus update.
hierarchy : str
Hierarchy label.
previous_sequences : Mapping[str, int] | None
Accepted per-source sequence watermarks, or None.
degraded_threshold : float
Coherence threshold below which a child is degraded.
critical_threshold : float
Coherence threshold below which a child is critical.
min_confidence : float
Minimum child summary confidence to include.
protocol_version : str
Hierarchy sync protocol version.
Returns¶
tuple[HierarchyConsensusRound, ...] The per-round gossip consensus states.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/consensus.py
79 80 81 82 83 84 85 86 87 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 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 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 | |
build_hierarchical_orchestration_plan ¶
build_hierarchical_orchestration_plan(
children: Iterable[ChildSupervisorSummary],
*,
hierarchy: str = "child_supervisors_to_parent",
degraded_threshold: float = 0.65,
critical_threshold: float = 0.35,
min_confidence: float = 0.5,
) -> HierarchicalOrchestrationPlan
Build a parent UPDE state and escalation set from child summaries.
This is a non-networked hierarchy foundation. It composes child coherence
summaries into a parent-level UPDEState so existing regime, policy, FEP,
causal, and audit paths can reason over nested supervisors without reading
raw child observations.
Parameters¶
children : Iterable[ChildSupervisorSummary] Child supervisor summaries. hierarchy : str Hierarchy label. degraded_threshold : float Coherence threshold below which a child is degraded. critical_threshold : float Coherence threshold below which a child is critical. min_confidence : float Minimum child summary confidence to include.
Returns¶
HierarchicalOrchestrationPlan The parent plan and escalation set.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/plan.py
build_hierarchy_sync_envelope ¶
build_hierarchy_sync_envelope(
summary: ChildSupervisorSummary,
*,
source_node: str,
sequence: int,
protocol_version: str = _DEFAULT_HIERARCHY_SYNC_PROTOCOL,
monotonic_time_s: float | None = None,
) -> HierarchySyncEnvelope
Build a deterministic edge/cloud hierarchy sync envelope.
The envelope is transport-neutral: callers may write it to JSONL, send it over a message bus, or hand it to tests without this module opening sockets or performing live deployment work.
Parameters¶
summary : ChildSupervisorSummary
The child supervisor summary.
source_node : str
Identifier of the source node.
sequence : int
Monotonic envelope sequence number.
protocol_version : str
Hierarchy sync protocol version.
monotonic_time_s : float | None
Monotonic timestamp in seconds, or None.
Returns¶
HierarchySyncEnvelope The deterministic hierarchy sync envelope.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
ingest_hierarchy_sync_envelopes ¶
ingest_hierarchy_sync_envelopes(
envelopes: Sequence[HierarchySyncEnvelope],
*,
previous_sequences: Mapping[str, int] | None = None,
hierarchy: str = "edge_cloud_summary_sync",
degraded_threshold: float = 0.65,
critical_threshold: float = 0.35,
min_confidence: float = 0.5,
protocol_version: str = _DEFAULT_HIERARCHY_SYNC_PROTOCOL,
) -> HierarchySyncLedger
Validate envelopes and build a parent plan from accepted summaries.
Parent nodes reject stale or duplicate sequence numbers per source node and reject protocol-version mismatches. Accepted envelopes are sorted by source node and sequence before parent-state composition, making JSONL replay and cloud ingestion deterministic.
Parameters¶
envelopes : Sequence[HierarchySyncEnvelope]
The ordered transport envelopes.
previous_sequences : Mapping[str, int] | None
Accepted per-source sequence watermarks, or None.
hierarchy : str
Hierarchy label.
degraded_threshold : float
Coherence threshold below which a child is degraded.
critical_threshold : float
Coherence threshold below which a child is critical.
min_confidence : float
Minimum child summary confidence to include.
protocol_version : str
Hierarchy sync protocol version.
Returns¶
HierarchySyncLedger The sync ledger built from accepted summaries.
Raises¶
ValueError If an envelope fails validation.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 | |
load_hierarchy_sync_envelope ¶
load_hierarchy_sync_envelope(
record: HierarchySyncEnvelope
| Mapping[str, object]
| str,
) -> HierarchySyncEnvelope
Parse a JSON string or decoded mapping into a strict sync envelope.
Parameters¶
record : HierarchySyncEnvelope | Mapping[str, object] | str A sync envelope, decoded mapping, or JSON string.
Returns¶
HierarchySyncEnvelope The parsed strict sync envelope.
Raises¶
ValueError If the record cannot be parsed into a strict envelope.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy/sync.py
Hierarchy Adapter Boundaries¶
hierarchy_adapters adds decoded JSONL, REST-payload, and WebSocket-frame
helpers over HierarchyTransportRuntime. These helpers are transport
boundaries only: they do not open sockets, own HTTP servers, start event loops,
or apply actuation. They return HierarchyAdapterResult records containing
accepted/rejected counts, sequence watermarks, parent-plan summaries, and the
underlying sync ledger.
hierarchy_adapters ¶
Decoded REST, WebSocket-frame, and JSONL hierarchy adapter boundaries.
The adapter helpers validate already-decoded payloads, content-type headers,
frame kinds, and envelope batches before passing records into a socket-free
HierarchyTransportRuntime. They return audit-safe result records with
watermarks and ledgers. The module deliberately owns no HTTP server, WebSocket,
filesystem tailer, or retry loop.
Classes¶
HierarchyAdapterResult
dataclass
¶
HierarchyAdapterResult(
boundary: str,
ledger: HierarchySyncLedger,
watermarks: Mapping[str, int],
frame_kind: str | None = None,
status: str = "accepted",
)
Audit-safe result returned by decoded hierarchy adapter boundaries.
Attributes¶
accepted_count
property
¶
Return the number of envelopes accepted by the runtime.
Returns¶
int Return the number of envelopes accepted by the runtime.
rejected_count
property
¶
Return the number of envelopes rejected by the runtime.
Returns¶
int Return the number of envelopes rejected by the runtime.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe adapter audit payload.
Returns¶
dict[str, object] Return a deterministic JSON-safe adapter audit payload.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy_adapters.py
Functions:¶
replay_hierarchy_jsonl ¶
replay_hierarchy_jsonl(
lines: Iterable[
str | Mapping[str, object] | HierarchySyncEnvelope
],
*,
runtime: HierarchyTransportRuntime | None = None,
) -> HierarchyAdapterResult
Replay decoded or JSONL hierarchy records through a socket-free runtime.
Parameters¶
lines : Iterable[str | Mapping[str, object] | HierarchySyncEnvelope]
Decoded or JSONL hierarchy records.
runtime : HierarchyTransportRuntime | None
The socket-free transport runtime, or None.
Returns¶
HierarchyAdapterResult The adapter result for the replayed records.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy_adapters.py
handle_hierarchy_rest_payload ¶
handle_hierarchy_rest_payload(
payload: Mapping[str, object],
*,
headers: Mapping[str, object],
runtime: HierarchyTransportRuntime | None = None,
) -> HierarchyAdapterResult
Handle a decoded REST request payload without owning an HTTP server.
Parameters¶
payload : Mapping[str, object]
The decoded REST request payload.
headers : Mapping[str, object]
Decoded request headers.
runtime : HierarchyTransportRuntime | None
The socket-free transport runtime, or None.
Returns¶
HierarchyAdapterResult The adapter result for the REST payload.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy_adapters.py
handle_hierarchy_frame ¶
handle_hierarchy_frame(
frame: Mapping[str, object],
*,
runtime: HierarchyTransportRuntime | None = None,
) -> HierarchyAdapterResult
Handle a decoded WebSocket-style frame without owning a socket.
Parameters¶
frame : Mapping[str, object]
The decoded WebSocket-style frame.
runtime : HierarchyTransportRuntime | None
The socket-free transport runtime, or None.
Returns¶
HierarchyAdapterResult The adapter result for the frame.
Raises¶
ValueError If the frame is malformed.
Source code in src/scpn_phase_orchestrator/supervisor/hierarchy_adapters.py
Byzantine Meta-Orchestrator Manifest¶
build_bft_meta_orchestrator_manifest() turns signed child-supervisor policy
proposals into an offline quorum-review manifest. The manifest records the
winning payload hash, accepted and rejected node IDs, hash-linked audit parent,
blocked reasons when quorum is absent, and a canonical manifest hash.
The helper verifies HMAC-SHA256 proposal signatures against a supplied keyring, but it does not open network transport or permit direct actuation. Accepted manifests still have to pass the normal supervisor review gate before use.
byzantine ¶
Offline Byzantine-tolerant policy proposal consensus manifests.
Functions:¶
sign_policy_proposal ¶
sign_policy_proposal(
node_id: str,
payload: Mapping[str, object],
previous_audit_hash: str,
signing_key: str,
) -> dict[str, object]
Return a deterministic signed policy proposal record.
Parameters¶
node_id : str Identifier of the proposing node. payload : Mapping[str, object] The policy-proposal payload to sign. previous_audit_hash : str Hash of the previous audit record in the chain. signing_key : str HMAC signing key for the record.
Returns¶
dict[str, object] The deterministic signed policy-proposal record.
Source code in src/scpn_phase_orchestrator/supervisor/byzantine.py
build_bft_meta_orchestrator_manifest ¶
build_bft_meta_orchestrator_manifest(
proposals: Sequence[Mapping[str, object]],
keyring: Mapping[str, str],
*,
quorum: int,
) -> dict[str, object]
Build a review-only three-node BFT consensus manifest.
Parameters¶
proposals : Sequence[Mapping[str, object]] Signed policy proposals from the participating nodes. keyring : Mapping[str, str] Mapping of node id to its verification key. quorum : int Number of agreeing nodes required for consensus.
Returns¶
dict[str, object] The review-only BFT consensus manifest.
Raises¶
ValueError If the proposals fail signature or quorum checks.
Source code in src/scpn_phase_orchestrator/supervisor/byzantine.py
69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 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 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 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | |
Strange-Loop Supervisor Monitor¶
StrangeLoopSupervisor is the first self-referential supervisor slice. It
treats the supervisor's own action stream as a four-dimensional control
channel over K, alpha, zeta, and Psi. The monitor records recent
action bundles, computes a control phase, control coherence, drift score,
oscillation score, and over-control score, then returns conservative damping
recommendations for a normal policy or safety gate to approve.
from scpn_phase_orchestrator.supervisor import StrangeLoopSupervisor
loop = StrangeLoopSupervisor(overcontrol_threshold=0.2)
assessment = loop.observe(actions_from_supervisor_policy)
if assessment.recommended_actions:
audit_payload = assessment.to_audit_record()
This slice does not hot-patch the supervisor or claim autonomous self-awareness. It provides an auditable meta-control signal that can detect policy drift, control-loop oscillation, and excessive actuation before those dynamics are fed back into the plant.
Long-run drift scenario helpers exercise that monitor across deterministic
40-step review traces for stable power-grid trims, cardiac policy drift,
traffic-control oscillation, and plasma over-control. The fixture corpus stays
non-actuating and execution-disabled, publishes stable scenario/result hashes,
and is gated in the reference suite so drift, oscillation, and over-control
threshold behavior remains reproducible across releases.
Studio renders the resulting audit records through the public
scpn_phase_orchestrator.studio.build_strange_loop_studio_panel() facade,
which preserves the
strange_loop_drift_review_not_live_actuation boundary, validates SHA-256
evidence hashes and finite metric ranges, and keeps all recommendations behind
the normal review and safety gate.
strange_loop ¶
Self-monitoring supervisor action-history diagnostics.
StrangeLoopSupervisor embeds recent control-action bundles into native
control-knob space and measures drift, oscillation, coherence, and over-control
from the bounded history. Recommendations are conservative damping proposals
for an outer policy gate to approve. The monitor records diagnostics only and
does not apply actions or alter the underlying supervisor.
Classes¶
StrangeLoopDriftScenario
dataclass
¶
StrangeLoopDriftScenario(
domain: str,
scenario_id: str,
description: str,
expected_trigger: str,
action_schedule: tuple[tuple[ControlAction, ...], ...],
non_actuating: bool = True,
execution_disabled: bool = True,
claim_boundary: str = _DRIFT_SCENARIO_BOUNDARY,
)
Deterministic long-run action-history scenario for strange-loop review.
Methods:¶
scenario_hash ¶
Return a deterministic scenario hash over the full action schedule.
Returns¶
str Return a deterministic scenario hash over the full action schedule.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
to_audit_record ¶
Return a JSON-safe long-run scenario record.
Returns¶
dict[str, object] Return a JSON-safe long-run scenario record.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
StrangeLoopDriftScenarioResult
dataclass
¶
StrangeLoopDriftScenarioResult(
domain: str,
scenario_id: str,
expected_trigger: str,
step_count: int,
max_drift_score: float,
max_oscillation_score: float,
max_overcontrol_score: float,
min_control_coherence: float,
triggered_recommendation_count: int,
final_recommended_knobs: tuple[str, ...],
passed_expected_trigger: bool,
scenario_hash: str,
result_hash: str,
non_actuating: bool = True,
execution_disabled: bool = True,
claim_boundary: str = _DRIFT_SCENARIO_BOUNDARY,
)
Audit-ready result for one long-run strange-loop drift scenario.
Methods:¶
to_audit_record ¶
Return a JSON-safe drift scenario result.
Returns¶
dict[str, object] Return a JSON-safe drift scenario result.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
StrangeLoopAssessment
dataclass
¶
StrangeLoopAssessment(
control_phase: float,
control_coherence: float,
drift_score: float,
oscillation_score: float,
overcontrol_score: float,
recommended_actions: tuple[ControlAction, ...],
)
Audit-ready metrics for supervisor self-control dynamics.
Methods:¶
to_audit_record ¶
Return a JSON-serialisable record for supervisor audit logs.
Returns¶
dict[str, object] Return a JSON-serialisable record for supervisor audit logs.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
StrangeLoopSupervisor ¶
StrangeLoopSupervisor(
*,
history_size: int = 12,
drift_threshold: float = 0.25,
oscillation_threshold: float = 0.5,
overcontrol_threshold: float = 0.2,
damping_gain: float = 0.05,
ttl_s: float = 3.0,
)
Treat supervisor action history as a self-referential control channel.
The monitor embeds each recent action bundle into the four native control
knobs (K, alpha, zeta, Psi). It then measures whether the supervisor is
drifting, oscillating, or over-actuating and emits conservative damping
recommendations for an outer policy gate to approve.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
Methods:¶
observe ¶
Record one supervisor action bundle and assess self-control state.
Parameters¶
actions : list[ControlAction] The control actions to apply or assess.
Returns¶
StrangeLoopAssessment The self-control assessment for the action bundle.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
Functions:¶
build_strange_loop_drift_scenarios ¶
Build deterministic long-run strange-loop drift review scenarios.
Returns¶
tuple[StrangeLoopDriftScenario, ...] Build deterministic long-run strange-loop drift review scenarios.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
evaluate_strange_loop_drift_scenarios ¶
evaluate_strange_loop_drift_scenarios(
scenarios: Sequence[StrangeLoopDriftScenario]
| None = None,
) -> tuple[StrangeLoopDriftScenarioResult, ...]
Evaluate long-run drift scenarios through StrangeLoopSupervisor.
Parameters¶
scenarios : Sequence[StrangeLoopDriftScenario] | None
The drift scenarios to evaluate, or None for the defaults.
Returns¶
tuple[StrangeLoopDriftScenarioResult, ...] The drift-scenario results.
Source code in src/scpn_phase_orchestrator/supervisor/strange_loop.py
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 | |
Morphogenetic Topology Field¶
MorphogeneticTopologySupervisor evolves a persistent field over the pairwise
coupling topology. Each step combines:
- pairwise phase-alignment reaction terms
- incident-edge diffusion over the current topology field
- bounded growth and shrink rates
- a hard maximum per-step coupling delta
The result is a next-step K_nm, a carried MorphogeneticFieldState, grown and
shrunk edge lists, and compact field statistics for audit logs.
from scpn_phase_orchestrator.supervisor import (
MorphogeneticTopologySupervisor,
build_morphogenetic_field_snapshot,
render_morphogenetic_field_svg,
)
supervisor = MorphogeneticTopologySupervisor()
result = supervisor.step(phases, knm)
next_knm = result.knm
field_state = result.field_state
audit_payload = result.to_audit_record()
snapshot = build_morphogenetic_field_snapshot(result, top_k=5)
heatmap_rows = snapshot.heatmap_rows
svg_artifact = render_morphogenetic_field_svg(result, top_k=5)
This slice provides a reviewable grow/shrink primitive for topology shaping. It does not bypass the existing policy, causal, STL, or action-projection gates. The field snapshot helper is dependency-free and emits JSON-safe statistics, ASCII heatmap rows, and strongest-edge records for reports or later UI rendering. Coupling and carried topology-field matrices are strict off-diagonal graph objects: boolean and complex aliases are rejected before float coercion, and non-zero self-edge diagonals are rejected before any field evolution, snapshot, or SVG rendering.
render_morphogenetic_field_svg() is the first richer UI rendering surface for
the same field state. It produces a deterministic, dependency-free SVG heatmap
plus top-edge labels and snapshot metadata. The renderer is passive: it turns an
already computed field into a review artefact and does not mutate policy,
coupling, or actuation state.
Studio packages those SVG artefacts through the public
scpn_phase_orchestrator.studio.build_morphogenetic_field_studio_panel()
facade, which validates complete SVG
documents, fixed-width heatmap rows, field-energy statistics, and sorted
off-diagonal topology edges before exposing the panel as passive operator
evidence.
domainpacks/swarm_robotics/morphogenetic_field_demo.py provides a deterministic
domainpack proof: it evaluates a split-flock phase state and emits the
morphogenetic field audit payload plus snapshot rows without live actuation.
domainpacks/power_grid/morphogenetic_field_demo.py provides the same
non-actuating proof for a stressed grid replay: generator rotor and area
frequency layers remain near-synchronised while tie-line, load-demand, and
renewable layers drift, producing reviewable grown/shrunk field-edge records.
domainpacks/traffic_flow/morphogenetic_field_demo.py extends the demo set with
a corridor spillback replay: corridor, network, and equity-pressure layers
remain locally aligned while intersection, demand, and weather phases stress
the field, again without live actuation.
domainpacks/plasma_control/morphogenetic_field_demo.py adds a research plasma
replay: transport-barrier, current-profile, and global-equilibrium layers remain
locally aligned while turbulence, tearing, ELM, and wall-interaction phases
stress the field, again without live actuation.
domainpacks/network_security/morphogenetic_field_demo.py adds a
lateral-movement replay: normal-traffic and defence-response layers remain
locally aligned while the attack-vector layer stresses the field, again without
live actuation.
morphogenetic ¶
Morphogenetic topology-field diagnostics for bounded coupling proposals.
The supervisor evolves a persistent normalized field from phase alignment, diffusion, and coherence-target reactions, then returns a clipped coupling proposal plus audit summaries of grown and shrunk edges. Snapshot and SVG helpers render review artifacts from computed fields. The module does not apply coupling updates to external systems or perform actuation.
Classes¶
MorphogeneticFieldPolicy
dataclass
¶
MorphogeneticFieldPolicy(
growth_rate: float = 0.2,
shrink_rate: float = 0.15,
diffusion_rate: float = 0.1,
coherence_target: float = 0.75,
max_delta: float = 0.05,
max_coupling: float = 10.0,
)
Knobs for reaction-diffusion-style topology field evolution.
MorphogeneticFieldState
dataclass
¶
Persistent topology field carried between supervisor ticks.
Methods:¶
to_audit_snapshot ¶
Return compact, serialisable field statistics for audit logs.
Returns¶
dict[str, object] Return compact, serialisable field statistics for audit logs.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
MorphogeneticFieldResult
dataclass
¶
MorphogeneticFieldResult(
knm: FloatArray,
field_state: MorphogeneticFieldState,
grown_edges: tuple[tuple[int, int, float], ...],
shrunk_edges: tuple[tuple[int, int, float], ...],
delta_norm: float,
global_coherence: float,
)
Output of one morphogenetic topology field step.
Methods:¶
to_audit_record ¶
Return a serialisable topology-field audit payload.
Returns¶
dict[str, object] Return a serialisable topology-field audit payload.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
MorphogeneticFieldSnapshot
dataclass
¶
MorphogeneticFieldSnapshot(
shape: tuple[int, int],
mean: float,
minimum: float,
maximum: float,
l2_norm: float,
heatmap_rows: tuple[str, ...],
top_edges: tuple[tuple[int, int, float], ...],
)
Compact visual snapshot of a morphogenetic topology field.
Methods:¶
to_audit_record ¶
Return a JSON-safe field snapshot for docs, reports, and audits.
Returns¶
dict[str, object] Return a JSON-safe field snapshot for docs, reports, and audits.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
MorphogeneticFieldSVG
dataclass
¶
Dependency-free SVG rendering of a morphogenetic topology field.
Methods:¶
to_audit_record ¶
Return a JSON-safe SVG artefact record for review tooling.
Returns¶
dict[str, object] Return a JSON-safe SVG artefact record for review tooling.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
MorphogeneticTopologySupervisor ¶
Grow or shrink pairwise topology from a persistent coherence field.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
Methods:¶
step ¶
step(
phases: FloatArray,
knm: FloatArray,
field_state: MorphogeneticFieldState | None = None,
) -> MorphogeneticFieldResult
Evolve the topology field and return the next pairwise coupling.
Parameters¶
phases : FloatArray
Oscillator phases in radians, shape (N,).
knm : FloatArray
Coupling matrix K_nm, shape (N, N).
field_state : MorphogeneticFieldState | None
The morphogenetic field state, or None.
Returns¶
MorphogeneticFieldResult The next pairwise coupling field result.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
Functions:¶
build_morphogenetic_field_snapshot ¶
build_morphogenetic_field_snapshot(
field_state: MorphogeneticFieldState
| MorphogeneticFieldResult,
*,
top_k: int = 5,
palette: str = " .:-=+*#%@",
) -> MorphogeneticFieldSnapshot
Build a compact visual snapshot for a topology field.
The snapshot is dependency-free and audit-oriented: it exposes summary statistics, ASCII heatmap rows, and the strongest non-diagonal field edges.
Parameters¶
field_state : MorphogeneticFieldState | MorphogeneticFieldResult
The morphogenetic field state, or None.
top_k : int
Number of strongest entries to retain.
palette : str
Colour palette name for the snapshot.
Returns¶
MorphogeneticFieldSnapshot The compact morphogenetic field snapshot.
Raises¶
ValueError
If the field state or top_k is invalid.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
render_morphogenetic_field_svg ¶
render_morphogenetic_field_svg(
field_state: MorphogeneticFieldState
| MorphogeneticFieldResult,
*,
top_k: int = 5,
cell_size: int = 28,
title: str = "Morphogenetic topology field",
) -> MorphogeneticFieldSVG
Render a dependency-free SVG heatmap for a topology field.
The renderer is passive: it produces a review artefact from an already computed field and does not mutate policy, coupling, or actuation state.
Parameters¶
field_state : MorphogeneticFieldState | MorphogeneticFieldResult
The morphogenetic field state, or None.
top_k : int
Number of strongest entries to retain.
cell_size : int
SVG cell size in pixels.
title : str
Title rendered on the SVG.
Returns¶
MorphogeneticFieldSVG The dependency-free SVG heatmap artefact.
Raises¶
ValueError If the field state or rendering parameters are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/morphogenetic.py
299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 | |
Sheaf Coherence Supervisor¶
SheafCoherenceSupervisor evaluates N-channel node states against directed
restriction maps. It builds the block sheaf Laplacian, computes edge residuals,
and reports obstruction metrics for audit logs.
Inputs are fail-closed real-valued tensors: node_states must have shape
(n_nodes, n_channels) and restriction_maps must have shape
(n_nodes, n_nodes, n_channels, n_channels). Boolean aliases, complex values,
non-finite values, and malformed object payloads are rejected before Laplacian
assembly so the obstruction score cannot depend on implicit dtype coercion.
This supervisor-facing sheaf-cohomology slice exposes obstruction score, consistency energy, approximate kernel dimension, obstruction dimension, and a review-only obstruction-aware control primitive. It does not claim a complete formal proof system or autonomous sheaf-control loop.
from scpn_phase_orchestrator.supervisor import (
SheafCoherenceSupervisor,
build_sheaf_obstruction_summary,
)
supervisor = SheafCoherenceSupervisor(tolerance=1e-8)
result = supervisor.assess(node_states, restriction_maps)
summary = build_sheaf_obstruction_summary(result)
if result.obstruction_score > 0.1:
audit_payload = summary.to_audit_record()
propose_sheaf_obstruction_control() projects an obstructed section one
bounded step down the sheaf-Laplacian consistency-energy gradient. The use case
is operator review: identify a mathematically justified state correction that
reduces obstruction while recording before/after cohomology dimensions. The
proposal is always non-actuating, execution-disabled, and review-required.
from scpn_phase_orchestrator.supervisor import (
propose_sheaf_obstruction_control,
)
proposal = propose_sheaf_obstruction_control(
node_states,
restriction_maps,
step_size=0.25,
max_update_norm=0.4,
)
assert proposal.projected_consistency_energy <= proposal.baseline_consistency_energy
assert proposal.execution_disabled
domainpacks/edge_consensus_nchannel/sheaf_obstruction_demo.py provides a
heterogeneous-domain replay: P, I, S, Load, Trust, and
ConsensusHealth node states are evaluated across edge, gateway, and parent
restriction maps, producing nominal and stressed obstruction audit records
without live actuation.
domainpacks/power_grid/sheaf_obstruction_demo.py adds a second
heterogeneous-domain replay. It evaluates generation, tie-line, load, and
renewable regions over rotor-angle, frequency-deviation, tie-flow, demand, and
renewable-ramp channels, then reports nominal versus line-fault obstruction
summaries.
domainpacks/network_security/sheaf_obstruction_demo.py adds a security replay.
It evaluates normal-traffic, attack-vector, and defence-response cohorts over
traffic-rate, threat-level, defence-phase, and trust-score channels, then
reports nominal versus lateral-movement obstruction summaries.
build_sheaf_obstruction_summary() hardens the raw obstruction metric into a
reviewable triage record. It classifies nominal, warning, and critical
states from explicit thresholds and reports the strongest residual edges so
operators can see which directed restrictions are failing.
Studio exposes this evidence through
build_sheaf_cohomology_studio_panel(records, summaries, control_proposals).
That panel keeps obstruction records, residual-edge summaries, and bounded
review-only control proposals together while preserving disabled execution and
actuation gates.
sheaf ¶
Sheaf-Laplacian coherence assessment for N-channel supervisor states.
Classes¶
SheafCoherenceResult
dataclass
¶
SheafCoherenceResult(
laplacian: FloatArray,
residuals: FloatArray,
obstruction_score: float,
consistency_energy: float,
kernel_dimension: int,
obstruction_dimension: int,
edge_count: int,
tolerance: float,
)
Audit-ready obstruction assessment for a cellular-sheaf state.
Methods:¶
to_audit_record ¶
Return a compact serialisable payload for supervisor audit logs.
Returns¶
dict[str, object] Return a compact serialisable payload for supervisor audit logs.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
SheafObstructionSummary
dataclass
¶
SheafObstructionSummary(
severity: str,
top_residual_edges: tuple[
tuple[int, int, float, tuple[float, ...]], ...
],
obstruction_score: float,
warning_threshold: float,
critical_threshold: float,
)
Review summary for obstruction hardening and audit triage.
Methods:¶
to_audit_record ¶
Return a JSON-serialisable obstruction summary.
Returns¶
dict[str, object] Return a JSON-serialisable obstruction summary.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
SheafControlProposal
dataclass
¶
SheafControlProposal(
baseline_obstruction_score: float,
projected_obstruction_score: float,
baseline_consistency_energy: float,
projected_consistency_energy: float,
baseline_kernel_dimension: int,
projected_kernel_dimension: int,
baseline_obstruction_dimension: int,
projected_obstruction_dimension: int,
recommended_update: FloatArray,
projected_node_states: FloatArray,
update_norm: float,
step_size: float,
max_update_norm: float,
accepted_for_review: bool,
non_actuating: bool,
execution_disabled: bool,
operator_review_required: bool,
blocked_reasons: tuple[str, ...],
)
Review-only obstruction-aware sheaf-Laplacian control proposal.
Methods:¶
to_audit_record ¶
Return a compact serialisable payload for operator review.
Returns¶
dict[str, object] Return a compact serialisable payload for operator review.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
SheafCoherenceSupervisor ¶
Assess whether N-channel states agree across restriction maps.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
Methods:¶
assess ¶
Return sheaf obstruction metrics for one supervisor tick.
Parameters¶
node_states : FloatArray
Per-node channel states, shape (N, C).
restriction_maps : FloatArray
Directed sheaf restriction maps.
Returns¶
SheafCoherenceResult The sheaf obstruction metrics for the tick.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
Functions:¶
build_sheaf_obstruction_summary ¶
build_sheaf_obstruction_summary(
result: SheafCoherenceResult,
*,
warning_threshold: float = 0.05,
critical_threshold: float = 0.25,
top_k: int = 5,
) -> SheafObstructionSummary
Build a passive triage summary from a sheaf-coherence result.
Parameters¶
result : SheafCoherenceResult The sheaf-coherence result to summarise. warning_threshold : float Obstruction value above which a warning is raised. critical_threshold : float Coherence threshold below which a child is critical. top_k : int Number of strongest entries to retain.
Returns¶
SheafObstructionSummary The passive triage summary.
Raises¶
ValueError If the result or thresholds are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
propose_sheaf_obstruction_control ¶
propose_sheaf_obstruction_control(
node_states: FloatArray,
restriction_maps: FloatArray,
*,
step_size: float = 0.25,
max_update_norm: float = 0.25,
tolerance: float = 1e-08,
max_backtracking_steps: int = 12,
) -> SheafControlProposal
Propose a review-only correction along the sheaf-Laplacian gradient.
The proposal minimises the cellular-sheaf consistency energy
x.T @ L @ x by a bounded explicit gradient step. A small deterministic
backtracking line search is used so accepted proposals never increase the
measured obstruction energy. The result is an audit artefact only:
execution is disabled and any live actuation requires a separate operator
approval path.
Parameters¶
node_states : FloatArray
Per-node channel states, shape (N, C).
restriction_maps : FloatArray
Directed sheaf restriction maps.
step_size : float
Gradient step size for the proposed correction.
max_update_norm : float
Maximum norm of the proposed correction.
tolerance : float
Numerical tolerance.
max_backtracking_steps : int
Maximum backtracking line-search steps.
Returns¶
SheafControlProposal The review-only sheaf correction proposal.
Raises¶
ValueError If the node states or step parameters are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
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 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 | |
sheaf_coherence ¶
sheaf_coherence(
node_states: FloatArray,
restriction_maps: FloatArray,
tolerance: float = 1e-08,
) -> SheafCoherenceResult
Measure cross-channel consistency over a directed cellular sheaf.
Parameters¶
node_states : FloatArray
N-channel node state matrix with shape (n_nodes, n_channels).
restriction_maps : FloatArray
Directed restriction maps with shape (n_nodes, n_nodes, n_channels,
n_channels). Entry restriction_maps[i, j] maps node j into node i.
tolerance : float
Numerical threshold used for approximate nullity and obstruction counts.
Returns¶
SheafCoherenceResult
A SheafCoherenceResult with the block sheaf Laplacian, directed residual
tensor, obstruction score, consistency energy, and audit-visible approximate
dimensions.
Raises¶
ValueError If the node states or restriction maps are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
sheaf_laplacian ¶
Build the block sheaf Laplacian from directed restriction maps.
Parameters¶
restriction_maps : FloatArray Directed sheaf restriction maps. tolerance : float Numerical tolerance.
Returns¶
FloatArray The block sheaf Laplacian.
Raises¶
ValueError If the restriction maps are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/sheaf.py
Value-Alignment Guard¶
ValueAlignmentGuard is a hard safety wrapper around proposed
ControlAction lists. It evaluates explicit objective constraints, blocks
violating actions, and returns a forced fallback action set when the proposal
does not satisfy the configured score threshold.
The guard is intentionally simple and auditable: no hidden reward model is
loaded at runtime. Domainpacks can translate their safety or objective priors
into ValueConstraint entries and attach the resulting decision record to the
normal audit trace.
Policies may also include ValueParetoObjective entries. When present,
ValueAlignmentGuard.evaluate(..., objective_deltas={...}) requires finite
objective deltas, blocks regressions beyond each objective's allowed tolerance,
and requires at least one positive configured objective to improve. Missing
objective evidence fails closed and forces the same safe fallback path. Audit
records include pareto_violations with the observed delta, required delta,
allowed regression, and counterfactual reason.
Binding specs may carry the same policy as a reviewable value_alignment
template:
value_alignment:
minimum_score: 0.8
constraints:
- name: limit-coupling
knob: K
scope: global
max_abs_value: 0.1
weight: 2.0
fallback_actions:
- knob: zeta
scope: global
value: 0.0
ttl_s: 1.0
justification: value guard safe hold
pareto_objectives:
- name: safety_margin
min_delta: 0.01
max_regression: 0.0
Use value_alignment_policy_from_binding_spec(spec) to convert that template
into a ValueAlignmentPolicy. Audit records include hard bound violations,
Pareto objective violations, and score-threshold counterfactuals so reviewers
can distinguish a blocked unsafe action, a candidate that regresses a protected
objective, and a fallback forced by the policy's minimum alignment score.
Domainpack templates now include review-time examples for cardiac rhythm, power grid, network security, fusion equilibrium, neuroscience EEG, brain connectome, sleep architecture, circadian biology, epidemic SIR, and other simulation/replay domainpacks. These templates are guard priors for reviewable candidate actions; they are not live medical, grid, vehicle, financial, public-health, or security operating policies.
from scpn_phase_orchestrator.actuation.mapper import ControlAction
from scpn_phase_orchestrator.supervisor import (
ValueAlignmentGuard,
ValueAlignmentPolicy,
ValueParetoObjective,
ValueConstraint,
value_alignment_policy_from_binding_spec,
)
policy = ValueAlignmentPolicy(
constraints=(ValueConstraint("limit-coupling", knob="K", max_abs_value=0.1),),
fallback_actions=(
ControlAction("zeta", "global", 0.0, 1.0, "alignment fallback: hold"),
),
pareto_objectives=(
ValueParetoObjective("safety_margin", min_delta=0.01, max_regression=0.0),
),
)
decision = ValueAlignmentGuard(policy).evaluate(
proposed_actions,
objective_deltas={"safety_margin": 0.02},
)
actions_to_apply = decision.actions_to_apply
audit_payload = decision.to_audit_record()
templated_policy = value_alignment_policy_from_binding_spec(binding_spec)
alignment ¶
Pareto-style value guard for supervisor actuation proposals.
Classes¶
ValueConstraint
dataclass
¶
ValueConstraint(
name: str,
knob: str = "*",
scope: str = "*",
min_value: float | None = None,
max_value: float | None = None,
max_abs_value: float | None = None,
weight: float = 1.0,
)
A hard value constraint over a proposed control action.
knob and scope accept "*" wildcards. Bounds are inclusive.
weight controls how much this constraint contributes to the
reported alignment score; a failed hard constraint always blocks the
action regardless of weight.
Methods:¶
applies_to ¶
Return whether this constraint applies to action.
Parameters¶
action : ControlAction The control action to test against the constraints.
Returns¶
bool
True when the constraint applies to the action.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
violations_for ¶
Return failed bound names for action.
Parameters¶
action : ControlAction The control action to test against the constraints.
Returns¶
tuple[str, ...] The failed bound names for the action.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
ValueViolation
dataclass
¶
ValueViolation(
constraint: str,
knob: str,
scope: str,
proposed_value: float,
failed_bounds: tuple[str, ...],
counterfactual: str,
)
A blocked action and the value constraint it violated.
Methods:¶
to_audit_record ¶
Return a serialisable violation record.
Returns¶
dict[str, object] Return a serialisable violation record.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
ValueScoreCounterfactual
dataclass
¶
Counterfactual record explaining a score-threshold fallback.
Methods:¶
to_audit_record ¶
Return a serialisable score-threshold counterfactual.
Returns¶
dict[str, object] Return a serialisable score-threshold counterfactual.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
ValueParetoObjective
dataclass
¶
A named objective delta that must stay on the review Pareto frontier.
ValueParetoViolation
dataclass
¶
ValueParetoViolation(
objective: str,
observed_delta: float | None,
required_delta: float,
allowed_regression: float,
counterfactual: str,
)
A failed Pareto objective review condition.
Methods:¶
to_audit_record ¶
Return a serialisable Pareto violation record.
Returns¶
dict[str, object] Return a serialisable Pareto violation record.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
ValueAlignmentPolicy
dataclass
¶
ValueAlignmentPolicy(
constraints: tuple[ValueConstraint, ...],
fallback_actions: ActionTuple = (),
minimum_score: float = 0.0,
pareto_objectives: tuple[
ValueParetoObjective, ...
] = (),
)
Configured objective constraints and fallback actuation.
ValueAlignmentDecision
dataclass
¶
ValueAlignmentDecision(
approved_actions: ActionTuple,
blocked_actions: ActionTuple,
fallback_actions: ActionTuple,
violations: tuple[ValueViolation, ...],
score_counterfactuals: tuple[
ValueScoreCounterfactual, ...
],
pareto_violations: tuple[ValueParetoViolation, ...],
alignment_score: float,
minimum_score: float,
)
Result of applying value constraints to proposed actions.
Attributes¶
satisfied
property
¶
Return whether the proposed action set passed the guard.
Returns¶
bool Return whether the proposed action set passed the guard.
actions_to_apply
property
¶
Return approved actions or the forced safe fallback path.
Returns¶
ActionTuple Return approved actions or the forced safe fallback path.
Methods:¶
to_audit_record ¶
Return a serialisable guard decision.
Returns¶
dict[str, object] Return a serialisable guard decision.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
ValueAlignmentGuard ¶
Block supervisor actions that violate configured value constraints.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
Methods:¶
evaluate ¶
evaluate(
actions: list[ControlAction] | ActionTuple,
*,
objective_deltas: ObjectiveDeltas | None = None,
) -> ValueAlignmentDecision
Evaluate proposed actions and return an auditable decision.
Parameters¶
actions : list[ControlAction] | ActionTuple
The control actions to apply or assess.
objective_deltas : ObjectiveDeltas | None
Per-objective deltas for the proposed actions, or None.
Returns¶
ValueAlignmentDecision The auditable value-alignment decision.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
Functions:¶
calibrate_value_alignment_replay_evidence ¶
calibrate_value_alignment_replay_evidence(
policy: ValueAlignmentPolicy,
replay_cases: Mapping[
str, list[ControlAction] | ActionTuple
],
*,
evidence_label: str = "value_alignment_replay_calibration",
) -> dict[str, object]
Calibrate a value-alignment policy against replayed action proposals.
The returned artifact is deterministic and review-only: it records what the guard would approve, block, or divert to fallback on replayed cases, but it never authorises live actuation. This gives production reviewers evidence for guard behaviour before any deployment-tier enforcement is claimed.
Parameters¶
policy : ValueAlignmentPolicy The value-alignment policy to calibrate. replay_cases : Mapping[str, list[ControlAction] | ActionTuple] Replay action proposals keyed by case name. evidence_label : str Label recorded with the calibration evidence.
Returns¶
dict[str, object] The calibration evidence for the policy.
Raises¶
ValueError If the policy or replay cases are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 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 | |
value_alignment_policy_from_binding_spec ¶
Build a policy from BindingSpec.value_alignment when present.
Parameters¶
spec : object The binding spec to read value-alignment configuration from.
Returns¶
ValueAlignmentPolicy | None
The value-alignment policy, or None if not present.
Raises¶
ValueError If the value-alignment configuration is malformed.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
value_alignment_policy_from_template ¶
Build a value-alignment policy from a binding-spec template mapping.
Expected shape::
value_alignment:
minimum_score: 0.8
constraints:
- name: limit-coupling
knob: K
max_abs_value: 0.1
fallback_actions:
- knob: zeta
scope: global
value: 0.0
ttl_s: 1.0
justification: safe hold
Parameters¶
template : Mapping[str, object] The value-alignment template mapping.
Returns¶
ValueAlignmentPolicy The value-alignment policy built from the template.
Source code in src/scpn_phase_orchestrator/supervisor/alignment.py
Policy Engine¶
Rule-based evaluation of supervisor actions.
SupervisorPolicy¶
SupervisorPolicy(
regime_manager: RegimeManager,
petri_adapter: PetriNetAdapter | None = None,
gains: SupervisorPolicyGains | None = None,
admission_gate: PolicyCBFAdmissionGate | None = None,
)
decide()¶
def decide(
upde_state: UPDEState,
boundary_state: BoundaryState,
petri_ctx: dict[str, float] | None = None,
) -> list[ControlAction]
Returns a list of ControlAction instructions. Each action specifies:
| Field | Type | Example |
|---|---|---|
knob |
str |
"K", "zeta", "psi" |
scope |
str |
"global", "layer_0" |
value |
float |
0.05 (K boost), 0.1 (zeta damp) |
ttl_s |
float |
5.0 (action expires after 5s) |
justification |
str |
"degraded: K boost" |
When an optional PolicyCBFAdmissionGate is supplied, matching supervisor
actions are admitted through verified neural CBF filters before decide()
returns. last_admission_records exposes deterministic audit records for the
latest call, including the CBF filter digest, certificate digest, admission
status, admitted value, and SMT-LIB artefact hash.
Regime-action mapping¶
| Regime | Actions |
|---|---|
| NOMINAL | None (no intervention) |
| DEGRADED | K boost +0.05 (global) |
| CRITICAL | ζ damping +0.1 + K reduce -0.03 (worst layer) |
| RECOVERY | K restore +0.025 (half boost, global) |
Hard violation override¶
Hard boundary violations (BoundaryState.hard_violations) force
CRITICAL regardless of R values.
Policy CBF Admission¶
PolicyCBFAdmissionGate is the opt-in bridge between heuristic supervisor
proposals and certificate-bound neural CBF admission. Each PolicyCBFChannel
selects one action knob/scope, validates a matching BarrierCertificate for the
provided ControlBarrierFilter, extracts named runtime metrics from UPDEState
and BoundaryState, and emits a deterministic SMT-LIB admission artefact for
the scalar CBF half-space checked at that decision. The gate does not execute
Z3 locally and does not actuate; it constrains, admits, or rejects proposal
values before downstream projection.
cbf_admission ¶
Certificate-bound CBF admission for supervisor ControlAction proposals.
SupervisorPolicy emits bounded, non-actuating action proposals. This module
adds an optional admission layer for deployments that have a verified neural
Control Barrier Function (CBF): matching actions are passed through the existing
certificate-bound CBF governor, and every decision emits a deterministic SMT-LIB
admission artefact. The artefact captures the exact scalar CBF half-space,
control bounds, selected action, and filter/certificate digests; it does not run
Z3 locally and does not grant actuation.
Classes¶
PolicyCBFAdmissionRecord
dataclass
¶
PolicyCBFAdmissionRecord(
knob: str,
scope: str,
proposed_value: float,
admitted_value: float,
status: str,
stages_applied: tuple[str, ...],
violations: tuple[str, ...],
barrier_value: float | None,
filter_digest: str,
certificate_verification_digest: str,
smt_artifact: FormalTextArtifact,
smt_artifact_hash: str,
)
Audit record for one CBF-admitted supervisor action.
Attributes¶
knob, scope : str
Action channel admitted by the CBF gate.
proposed_value : float
Original supervisor proposal.
admitted_value : float
Value admitted by the CBF governor.
status : str
Governor status: admitted, constrained, or rejected.
stages_applied : tuple[str, ...]
Envelope stages that modified the proposal.
violations : tuple[str, ...]
Rejection reasons, if any.
barrier_value : float | None
Current CBF value h(x).
filter_digest : str
Digest of the CBF filter configuration.
certificate_verification_digest : str
Digest of the certificate envelope used to validate the filter.
smt_artifact : FormalTextArtifact
Deterministic SMT-LIB admission artefact for this decision.
smt_artifact_hash : str
SHA-256 hash of :attr:smt_artifact.
content_hash : str
SHA-256 hash of the audit record excluding the SMT text.
Methods:¶
to_audit_record ¶
Return a JSON-safe CBF admission audit record.
Returns¶
dict[str, object] Admission decision, barrier/certificate digests, SMT artefact hash, and deterministic content hash.
Source code in src/scpn_phase_orchestrator/supervisor/cbf_admission.py
PolicyCBFAdmissionResult
dataclass
¶
PolicyCBFAdmissionResult(
actions: tuple[ControlAction, ...],
records: tuple[PolicyCBFAdmissionRecord, ...],
)
PolicyCBFChannel
dataclass
¶
PolicyCBFChannel(
knob: str,
scope: str,
barrier_filter: ControlBarrierFilter,
barrier_certificate: BarrierCertificate,
state_metrics: tuple[str, ...],
drift_bounds: tuple[float, ...],
previous_action: float = 0.0,
max_rate: float | None = None,
)
Certificate-bound CBF admission channel for one action knob/scope.
Parameters¶
knob, scope : str
Action selector. Only exact (knob, scope) matches are admitted by
this channel.
barrier_filter : ControlBarrierFilter
Verified CBF filter for the scalar action value.
barrier_certificate : BarrierCertificate
Certificate that validates :attr:barrier_filter.
state_metrics : tuple[str, ...]
Names of UPDE/boundary metrics used as the CBF state vector.
drift_bounds : tuple[float, ...]
Deterministic drift vector supplied to the CBF filter for admission.
previous_action : float
Held fallback and rate-limit reference for rejected decisions.
max_rate : float | None
Optional per-call rate limit. None uses the full control span.
Methods:¶
matches ¶
Return whether action belongs to this CBF channel.
Parameters¶
action : ControlAction Supervisor action proposal to compare with this channel's selector.
Returns¶
bool
True when both knob and scope match exactly.
Source code in src/scpn_phase_orchestrator/supervisor/cbf_admission.py
admit ¶
admit(
action: ControlAction,
upde_state: UPDEState,
boundary_state: BoundaryState,
) -> tuple[ControlAction, PolicyCBFAdmissionRecord]
Admit one matching action through the verified CBF governor.
Parameters¶
action : ControlAction
Supervisor action proposal. It must match :meth:matches.
upde_state : UPDEState
Current UPDE metrics used to build the CBF state vector.
boundary_state : BoundaryState
Current boundary metrics used to build the CBF state vector.
Returns¶
tuple[ControlAction, PolicyCBFAdmissionRecord] The admitted action and its deterministic audit record.
Raises¶
ValueError
If action does not match this channel.
Source code in src/scpn_phase_orchestrator/supervisor/cbf_admission.py
PolicyCBFAdmissionGate ¶
Apply configured CBF channels to supervisor action proposals.
Source code in src/scpn_phase_orchestrator/supervisor/cbf_admission.py
Methods:¶
admit_actions ¶
admit_actions(
actions: Sequence[ControlAction],
upde_state: UPDEState,
boundary_state: BoundaryState,
) -> PolicyCBFAdmissionResult
Admit matching actions and return transformed actions plus records.
Parameters¶
actions : Sequence[ControlAction] Supervisor action proposals. upde_state : UPDEState Current UPDE metrics. boundary_state : BoundaryState Current boundary-observer metrics.
Returns¶
PolicyCBFAdmissionResult Admitted action tuple and CBF audit records for matched actions.
Source code in src/scpn_phase_orchestrator/supervisor/cbf_admission.py
Performance: decide() < 50 μs.
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
Attributes¶
last_admission_records
property
¶
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
Causal Counterfactual Rollouts¶
CausalInterventionEngine evaluates proposed supervisor actions by running
paired UPDE trajectories from the same state:
- baseline: no action
- intervention: action-adjusted
K,alpha,zeta, orPsi
The result is a CounterfactualRollout with R and Psi trajectories,
final and mean R deltas, signed final phase delta, and a serialisable audit
payload.
Counterfactual phases, frequency vectors, coupling matrices, phase-lag
matrices, and lagged causal traces are validated as finite real-valued numeric
arrays before simulation or causal scoring. Boolean aliases and
complex/object-complex payloads are rejected before float coercion so rollouts
and lagged-linear influence estimates stay on the real Kuramoto state space.
from scpn_phase_orchestrator.supervisor import CausalInterventionEngine
engine = CausalInterventionEngine(n_oscillators=8, dt=0.01, horizon=20)
rollout = engine.evaluate_actions(phases, omegas, knm, alpha, 0.0, 0.0, actions)
record = rollout.to_audit_record()
attribution = rollout.attribute(threshold=1e-3).to_audit_record()
This is the first causal-supervision slice: it does not claim formal do-calculus yet, but it makes every proposed actuation comparable against a no-action counterfactual under the same UPDE dynamics.
CounterfactualRollout.attribute() compresses the final and mean R deltas
into an audit-ready effect label: stabilising, neutral, or destabilising.
learn_causal_graph() adds a lightweight live causal-model learner. It
estimates signed directed edges from lagged monitor traces and appends explicit
do(knob:scope) -> R edges from paired counterfactual rollouts. The output is
a CausalGraphEstimate with JSON-safe nodes, edge weights, confidence scores,
lags, and evidence labels for the audit trail.
from scpn_phase_orchestrator.supervisor import learn_causal_graph
graph = learn_causal_graph(
{"R_good": good_trace, "R_bad": bad_trace},
[rollout],
lag=1,
min_abs_weight=1e-4,
)
audit_graph = graph.to_audit_record()
build_temporal_causal_hypergraph_experiment() is the research-screening layer
for temporal-causal hypergraph candidates. It compares each proposed
time-symmetric hyperedge against a deterministic family of conventional
baselines before any claim can be made:
- lagged-linear graph edge score from
learn_causal_graph(); - lagged Pearson correlation between source and future target;
- lagged-delta Pearson correlation between source and target increment;
- pairwise Granger-style residual improvement over target history;
- target-persistence null correlation.
Candidate hyperedges are accepted for review only when their score beats the strongest baseline by the configured margin. The manifest stays research-only: production claims, hot patches, and actuation are disabled, and non-winning candidates are retained as blocked evidence for audit comparison. Use this for offline discovery of higher-order temporal coupling hypotheses, not for real-time causal intervention.
from scpn_phase_orchestrator.supervisor import (
build_temporal_causal_hypergraph_experiment,
)
manifest = build_temporal_causal_hypergraph_experiment(
{
"driver": driver_trace,
"response": response_trace,
"distractor": distractor_trace,
},
[
{
"sources": ["driver", "response"],
"target": "response",
"time_offsets": [-1, 0],
"score": candidate_score,
}
],
lag=1,
min_abs_weight=1e-4,
required_baseline_margin=0.1,
)
assert manifest["production_claim_permitted"] is False
assert manifest["baseline"]["strongest_baseline"] in {
"lagged_linear_graph",
"lagged_pearson",
"lagged_delta_pearson",
"granger_residual_improvement",
"target_persistence_null",
}
Domainpack demos:
domainpacks/cardiac_rhythm/causal_attribution_demo.pyevaluates a pacing-drive candidate against a ventricular-disturbance baseline.domainpacks/power_grid/causal_attribution_demo.pyevaluates a governor droop coupling candidate against a no-action load-step baseline.domainpacks/traffic_flow/causal_attribution_demo.pyevaluates a signal-cycle coupling candidate against a no-action corridor-spillback baseline.domainpacks/network_security/causal_attribution_demo.pyevaluates a firewall-coupling candidate against a no-action lateral-movement baseline.
Backend and cost: each evaluation performs two UPDE rollouts over the
configured horizon, so work scales with 2 * horizon engine steps. It uses
the existing UPDEEngine backend dispatcher; Rust acceleration is used when
available, otherwise the NumPy path is used.
causal ¶
Causal graph learning and counterfactual supervisor rollout diagnostics.
The module estimates directed influence from traces and compares baseline UPDE trajectories against parameter-intervention rollouts derived from proposed control actions. Inputs are validated for finite dimensions before simulation; action application mutates local copies of coupling and phase-lag matrices only. Outputs are audit-ready records and attribution summaries, not live actuation.
Classes¶
CausalAttribution
dataclass
¶
CausalAttribution(
effect: str,
trajectory_consistency: float,
score: float,
delta_R_final: float,
delta_R_mean: float,
threshold: float,
)
Attribution summary derived from a paired counterfactual rollout.
trajectory_consistency reports the fraction of the rollout horizon over
which the per-step order-parameter delta holds the attributed effect's sign
(or, for a neutral verdict, stays within the neutral band |delta| <=
threshold). It is a deterministic property of the single paired
trajectory — how steadily the intervention pushes R in the attributed
direction — not a statistical or frequentist confidence: the rollout draws no
samples, so there is no sampling distribution and no p-value to report. A
value near 1.0 means the effect is steady across the whole horizon; a
lower value means the sign only settles late or oscillates. Effect magnitude
lives in score/delta_R_final/delta_R_mean, kept separate so a
steady-but-small effect is not confused with a large-but-transient one.
Methods:¶
to_audit_record ¶
Return a JSON-serialisable attribution payload.
Returns¶
dict[str, object] Return a JSON-serialisable attribution payload.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
CausalInfluenceEdge
dataclass
¶
CausalInfluenceEdge(
source: str,
target: str,
weight: float,
confidence: float,
lag: int,
evidence: str,
)
Signed directed influence estimate between live causal graph nodes.
Methods:¶
to_audit_record ¶
Return a JSON-serialisable causal-edge payload.
Returns¶
dict[str, object] Return a JSON-serialisable causal-edge payload.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
CausalGraphEstimate
dataclass
¶
CausalGraphEstimate(
nodes: tuple[str, ...],
edges: tuple[CausalInfluenceEdge, ...],
lag: int,
min_abs_weight: float,
)
Audit-ready directed causal graph learned from traces and interventions.
Methods:¶
to_audit_record ¶
Return a JSON-serialisable causal graph estimate.
Returns¶
dict[str, object] Return a JSON-serialisable causal graph estimate.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
InterventionParameters
dataclass
¶
UPDE parameters after applying a supervisor intervention.
CounterfactualRollout
dataclass
¶
CounterfactualRollout(
baseline_R: list[float],
intervention_R: list[float],
baseline_psi: list[float],
intervention_psi: list[float],
delta_R_final: float,
delta_R_mean: float,
delta_psi_final: float,
actions: tuple[ControlAction, ...],
)
Paired baseline/intervention rollout summary for audit logging.
Methods:¶
to_audit_record ¶
Return a JSON-serialisable counterfactual audit payload.
Returns¶
dict[str, object] Return a JSON-serialisable counterfactual audit payload.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
attribute ¶
Summarise whether the intervention caused a measurable R change.
The verdict is decided by the trajectory-averaged effect score; its
steadiness across the horizon is reported separately as
trajectory_consistency (see :class:CausalAttribution), an honest
deterministic measure rather than a statistical confidence.
Parameters¶
threshold : float Decision threshold.
Returns¶
CausalAttribution The causal attribution of the intervention.
Raises¶
ValueError
If threshold is invalid.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
CausalInterventionEngine ¶
CausalInterventionEngine(
n_oscillators: int,
dt: float,
horizon: int = 20,
method: str = "rk4",
*,
layer_membership: Mapping[str, Sequence[int]]
| None = None,
)
Counterfactual UPDE rollouts for supervisor actions.
The engine answers the first causal supervision question: from the same state, what would the order-parameter trajectory look like with and without the proposed intervention?
Parameters¶
n_oscillators : int
Number of oscillators in the network.
dt : float
Integration timestep.
horizon : int
Number of rollout steps.
method : str
UPDE integration method.
layer_membership : Mapping[str, Sequence[int]] or None
Optional named layers with their member oscillator indices, enabling
do(K, layer_<name>) interventions. "layer_<name>" (default) or
"layer_<name>.within" perturbs the within-layer coupling sub-block;
"layer_<name>.incident" perturbs every coupling incident to a layer
member (the set generalisation of oscillator_). Without it, any
layer-scoped action is rejected.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
Methods:¶
evaluate_actions ¶
evaluate_actions(
phases: FloatArray,
omegas: FloatArray,
knm: FloatArray,
alpha: FloatArray,
zeta: float,
psi: float,
actions: list[ControlAction]
| tuple[ControlAction, ...],
) -> CounterfactualRollout
Compare no-action and intervened trajectories.
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.
zeta : float
External drive strength ζ.
psi : float
External drive reference phase Ψ in radians.
actions : list[ControlAction] | tuple[ControlAction, ...]
The control actions to apply or assess.
Returns¶
CounterfactualRollout The counterfactual rollout comparing no-action and intervened trajectories.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
apply_actions ¶
apply_actions(
knm: FloatArray,
alpha: FloatArray,
zeta: float,
psi: float,
actions: tuple[ControlAction, ...],
) -> InterventionParameters
Apply supported supervisor actions to UPDE parameters.
Parameters¶
knm : FloatArray
Coupling matrix K_nm, shape (N, N).
alpha : FloatArray
Phase-lag matrix in radians, shape (N, N), or None for no lag.
zeta : float
External drive strength ζ.
psi : float
External drive reference phase Ψ in radians.
actions : tuple[ControlAction, ...]
The control actions to apply or assess.
Returns¶
InterventionParameters The UPDE parameters after applying the actions.
Raises¶
ValueError If an action is unsupported.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
Functions:¶
learn_causal_graph ¶
learn_causal_graph(
trace: dict[str, list[float]],
rollouts: list[CounterfactualRollout]
| tuple[CounterfactualRollout, ...] = (),
*,
lag: int = 1,
min_abs_weight: float = 1e-06,
) -> CausalGraphEstimate
Estimate a signed live causal graph from traces and interventions.
Trace edges use lagged linear influence from source[t] to
target[t + lag] - target[t]. Intervention edges summarise paired
counterfactual rollouts as explicit do(knob:scope) -> R effects. The
estimate is intentionally lightweight and audit-first; it is not a formal
do-calculus proof.
Parameters¶
trace : dict[str, list[float]] Signal trace keyed by variable name, each a sequence of floats. rollouts : list[CounterfactualRollout] | tuple[CounterfactualRollout, ...] Counterfactual rollouts used to estimate causal edges. lag : int Lag in samples for the causal estimate. min_abs_weight : float Minimum absolute edge weight retained in the graph.
Returns¶
CausalGraphEstimate The estimated signed causal graph.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
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 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 | |
build_temporal_causal_hypergraph_experiment ¶
build_temporal_causal_hypergraph_experiment(
trace: dict[str, list[float]],
candidate_hyperedges: list[dict[str, object]]
| tuple[dict[str, object], ...],
*,
lag: int = 1,
min_abs_weight: float = 1e-06,
required_baseline_margin: float = 0.0,
) -> dict[str, object]
Build a research-only temporal-causal hypergraph experiment manifest.
The manifest compares candidate time-symmetric hyperedges against a deterministic family of conventional causal baselines: lagged-linear graph edges, lagged Pearson correlation, lagged-delta correlation, Granger-style residual improvement, and target persistence. It never permits production claims, hot-patching, or actuation; baseline failure keeps all candidates blocked as research evidence only.
Parameters¶
trace : dict[str, list[float]] Signal trace keyed by variable name, each a sequence of floats. candidate_hyperedges : list[dict[str, object]] | tuple[dict[str, object], ...] Candidate causal hyperedges to test. lag : int Lag in samples for the causal estimate. min_abs_weight : float Minimum absolute edge weight retained in the graph. required_baseline_margin : float Minimum baseline margin a hyperedge must beat.
Returns¶
dict[str, object] The temporal-causal hypergraph experiment manifest.
Raises¶
ValueError If the trace or candidate hyperedges are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/causal.py
604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 | |
Policy Rules (Declarative)¶
Declarative rules loaded from YAML/JSON configuration.
Data model¶
PolicyCondition(metric: str, layer: int | None, op: str, threshold: float)
CompoundCondition(conditions: list[PolicyCondition], logic: str = "AND")
PolicyAction(knob: str, scope: str, value: float, ttl_s: float)
PolicyRule(
name: str,
regimes: list[str], # active in these regimes
condition: PolicyCondition | CompoundCondition,
actions: list[PolicyAction],
cooldown_s: float = 0.0, # min seconds between firings
max_fires: int = 0, # 0 = unlimited
)
STL Monitors In Policy YAML¶
Policy files may also declare reviewable Signal Temporal Logic monitors under
top-level stl_monitors. These monitors do not emit control actions directly;
they evaluate scalar traces and return audit records that can be used by the
runtime gate or safety review job.
rules: []
stl_monitors:
- name: keep_sync
spec: always (R >= 0.3)
severity: hard
- name: eventual_recovery
spec: eventually (R >= 0.8)
from scpn_phase_orchestrator.supervisor.policy_rules import (
evaluate_policy_stl_specs,
load_policy_stl_specs,
)
specs = load_policy_stl_specs("policy.yaml")
results = evaluate_policy_stl_specs(specs, {"R": [0.2, 0.4, 0.9]})
audit_payloads = [result.to_audit_record() for result in results]
PolicyEngine¶
engine = PolicyEngine(rules)
engine.advance_clock(dt)
actions = engine.evaluate(regime, upde_state, good_layers, bad_layers)
Rules are evaluated in list order. Each rule fires if:
1. Current regime is in rule.regimes
2. Condition evaluates True against UPDEState metrics
3. Cooldown has expired since last firing
4. max_fires not exceeded
load_policy_rules(path) loads rules from YAML/JSON file.
policy_rules ¶
Policy DSL records, loaders, STL monitors, and bounded rule evaluation.
This module validates policy conditions, compound logic, action declarations,
cooldowns, max-fire limits, and STL monitor specifications before evaluation.
PolicyEngine returns ControlAction proposals when regime and metric
conditions match, while loaders cap rule, condition, and action counts. Policy
evaluation is local and does not apply actuation.
Classes¶
PolicyCondition
dataclass
¶
List the metric names known to the policy DSL.
Known metrics: R, R_good, R_bad, stability_proxy, pac_max, mean_amplitude, subcritical_fraction, amplitude_spread (per-layer), mean_amplitude_layer (per-layer).
CompoundCondition
dataclass
¶
AND/OR combinator over multiple PolicyConditions.
PolicyAction
dataclass
¶
Action emitted by a policy rule: knob, scope, target value, and TTL.
PolicyRule
dataclass
¶
PolicyRule(
name: str,
regimes: list[str],
condition: PolicyCondition | CompoundCondition,
actions: list[PolicyAction],
cooldown_s: float = 0.0,
max_fires: int = 0,
)
Named rule: fires actions when regime and condition match.
PolicySTLSpec
dataclass
¶
Named STL monitor declared by the policy DSL.
PolicySTLResult
dataclass
¶
PolicySTLAutomaton
dataclass
¶
Policy-level synthesized STL automaton with monitor name and severity.
Methods:¶
to_audit_record ¶
Return a JSON-safe policy STL automaton audit record.
Returns¶
dict[str, object] Return a JSON-safe policy STL automaton audit record.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
PolicyEngine ¶
Evaluate domainpack policy rules against current state.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
Methods:¶
advance_clock ¶
Advance the internal clock used for cooldown tracking.
Parameters¶
dt : float Integration step size.
Raises¶
ValueError
If dt is not positive.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
evaluate ¶
evaluate(
regime: Regime,
upde_state: UPDEState,
good_layers: list[int],
bad_layers: list[int],
) -> list[ControlAction]
Evaluate all rules against current state and return triggered actions.
Parameters¶
regime : Regime The current control regime. upde_state : UPDEState The current UPDE state. good_layers : list[int] Indices of the maintain (good) layers. bad_layers : list[int] Indices of the suppress (bad) layers.
Returns¶
list[ControlAction] The control actions triggered by the matching rules.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
Functions:¶
load_policy_stl_specs ¶
Load top-level stl_monitors declarations from a policy YAML file.
Parameters¶
path : str | Path Filesystem path to the policy YAML file.
Returns¶
list[PolicySTLSpec] The STL monitor declarations from the policy YAML.
Raises¶
ValueError If the policy YAML is malformed.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
evaluate_policy_stl_specs ¶
evaluate_policy_stl_specs(
specs: list[PolicySTLSpec] | tuple[PolicySTLSpec, ...],
trace: dict[str, list[float]],
) -> list[PolicySTLResult]
Evaluate policy-declared STL monitors over a scalar trace.
Parameters¶
specs : list[PolicySTLSpec] | tuple[PolicySTLSpec, ...] The STL monitor specifications. trace : dict[str, list[float]] Signal trace keyed by variable name, each a list of floats.
Returns¶
list[PolicySTLResult] The per-monitor STL evaluation results.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
synthesise_policy_stl_automata ¶
synthesise_policy_stl_automata(
specs: list[PolicySTLSpec] | tuple[PolicySTLSpec, ...],
trace: dict[str, list[float]],
) -> list[PolicySTLAutomaton]
Synthesise audit automata for policy-declared builtin STL monitors.
Parameters¶
specs : list[PolicySTLSpec] | tuple[PolicySTLSpec, ...] The STL monitor specifications. trace : dict[str, list[float]] Signal trace keyed by variable name, each a list of floats.
Returns¶
list[PolicySTLAutomaton] The audit automata for the policy STL monitors.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
load_policy_rules ¶
Load policy rules from a YAML file.
Supports both v0.1 (single condition/action) and v0.2 (compound conditions with logic, action chains) formats.
Parameters¶
path : str | Path Filesystem path to the policy YAML file.
Returns¶
list[PolicyRule] The policy rules loaded from the YAML file.
Raises¶
ValueError If the policy YAML is malformed.
Source code in src/scpn_phase_orchestrator/supervisor/policy_rules.py
560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 | |
Policy Diagnostics¶
Dry-run helpers for validating policy reachability, overlap, cooldown, and action output before a rule set is allowed into a live supervisor path.
policy_diagnostics ¶
Offline policy-rule dry-run diagnostics over audit-log style entries.
The dry-run path reconstructs reduced UPDEState snapshots, evaluates policy
rules without applying actions, tracks rule/action fire counts, and reports
unreachable rules, overlapping rule firings, and action collisions. It mutates
only the local PolicyEngine clock used for cooldown simulation and never
touches runtime supervisor or actuation state.
Classes¶
PolicyDryRunStep
dataclass
¶
One audit step and the policy rules that fired on it.
PolicyDryRunReport
dataclass
¶
PolicyDryRunReport(
steps: int,
rules: tuple[str, ...],
fire_counts: dict[str, int],
action_counts: dict[str, int],
unreachable_rules: tuple[str, ...],
overlapping_steps: tuple[int, ...],
action_collision_steps: tuple[int, ...],
step_reports: tuple[PolicyDryRunStep, ...],
)
Summary of replayed policy behaviour over an audit log.
Functions:¶
dry_run_policy_rules ¶
dry_run_policy_rules(
rules: list[PolicyRule],
entries: list[dict[str, Any]],
*,
good_layers: list[int],
bad_layers: list[int],
) -> PolicyDryRunReport
Replay policy rules over audit steps without applying actuation.
Parameters¶
rules : list[PolicyRule] The policy rules to evaluate. entries : list[dict[str, Any]] Audit-log step entries to replay. good_layers : list[int] Indices of the maintain (good) layers. bad_layers : list[int] Indices of the suppress (bad) layers.
Returns¶
PolicyDryRunReport The policy dry-run report.
Source code in src/scpn_phase_orchestrator/supervisor/policy_diagnostics.py
Formal Export¶
Export helpers translate Petri-net, policy-rule, and policy-declared STL surfaces into model-checker artefacts for independent safety analysis. PRISM exports remain the default; TLA+ modules are available for protocol and policy transition-system checks.
The CLI supports:
spo formal-export domainpacks/my_domain/binding_spec.yaml --export protocol
spo formal-export domainpacks/my_domain/binding_spec.yaml --export policy
spo formal-export domainpacks/my_domain/binding_spec.yaml --export stl
spo formal-export domainpacks/my_domain/binding_spec.yaml --export protocol-tla
spo formal-export domainpacks/my_domain/binding_spec.yaml --export policy-tla
spo formal-export domainpacks/my_domain/binding_spec.yaml --export policy-smt
spo formal-export domainpacks/my_domain/binding_spec.yaml --export package
--export stl reads stl_monitors from the sibling policy.yaml by default
and emits signal constants plus satisfied/violated labels for the builtin STL
subset. This is a model-checker linkage surface; full temporal automata
synthesis remains future work. --export protocol-tla emits a bounded TLA+
module with Petri places as variables, transition guards as constants, Init,
Next, Spec, and Safety == TypeOK. --export policy-tla emits bounded
rule-fire counters plus reachability predicates for fired rules and emitted
actions. --export policy-smt emits an SMT-LIB v2 feasibility model for Z3:
the model declares the active regime, metric inputs, bounded rule-fire counters,
rule firing predicates, emitted-action predicates, and a final check-sat
envelope asking whether at least one rule can fire under the declared guards.
--export package emits a JSON formal verification package manifest that binds
protocol PRISM/TLA, policy PRISM, and generated policy SMT-LIB artefact hashes
to named safety properties and external PRISM/TLC/Z3 command records. The
package API also accepts reviewed Promela and SMT-LIB text artefacts through
FormalTextArtifact, linking them to non-executing SPIN and Z3
command/readiness manifests under the same hash and disabled-execution
contract. The package does not run model checkers; all command records keep
execution_permitted=false. Add
--include-checker-readiness to append non-executing checker availability
records to that JSON; --checker-path executable=/path can make CI readiness
evidence deterministic, and --checker-path executable= forces a missing
checker record without invoking anything.
build_runtime_control_certificate() turns a package, checker readiness
records, externally reviewed checker result records, and finite runtime bounds
into a deterministic FormalRuntimeCertificate. The certificate is the runtime
handoff contract for verifiable control: every required property must have a
matching available checker and a passed result bound to the exact package hash.
Missing, failed, stale, or unavailable evidence produces status="blocked".
Even status="verified_non_actuating" keeps actuation_permitted=false; it is
an auditable precondition for operator review or a separate runtime monitor,
not permission to execute hardware controls.
Remote CI owns the first external execution lane through
formal-model-checkers.yml, which installs SPIN and Z3, materialises reviewed
Promela/SMT-LIB smoke artefacts, validates disabled package/readiness metadata,
and runs those external checkers only under the CI-only execution guard.
The same lane now also materialises safety-domain packages for
cardiac_rhythm, chemical_reactor, power_grid, pll_clock,
autonomous_vehicles, satellite_constellation, power_safety_nchannel,
traffic_flow, swarm_robotics, manufacturing_spc, and robotic_cpg. Each
domain package binds a SPIN operator-approval gate and a Z3 hard-bound
feasibility artefact derived from the domainpack safety boundaries, preserving
the disabled runtime-execution contract while allowing remote CI to execute the
external checker commands in an isolated environment.
For builtin STL automata, synthesise_stl_controller_candidates() provides a
non-actuating controller-synthesis bridge. It proposes signal-level candidate
actions from the weakest violated predicate and records actuating=False; the
proposal is an audit artefact, not a live controller or bypass around policy and
actuation safety gates. project_stl_controller_candidates() can then map
those candidates through explicit policy-approved projection templates and the
standard ActionProjector, yielding bounded ControlAction proposals while
still recording actuating=False.
synthesise_stl_closed_loop_plan() combines those two stages into an offline
closed-loop review artefact: it records the feedback signals, trace length,
future review horizon, projected actions, and fail-closed blockers without
mutating runtime state or enabling actuation.
formal_export ¶
Formal-model exporters for Petri nets, policy rules, and STL monitors.
The exporter functions convert already-validated supervisor structures into
PRISM or TLA+ text plus identifier maps, sanitising names and preserving metric,
transition, rule, action, and STL mappings for auditability, split into
responsibility modules (shared identifiers, verification package, runtime
certificate, and per-formalism exporters) behind a stable re-export surface.
Export routines are pure text generation; they do not invoke model checkers,
write files, or change the source policy/Petri structures. shutil is
re-exported so checker-availability tests resolve it on this package namespace.
Classes¶
PrismExport
dataclass
¶
PrismExport(
model: str,
place_names: dict[str, str],
metric_names: dict[str, str],
transition_names: dict[str, str],
rule_names: dict[str, str] = dict(),
action_names: dict[str, str] = dict(),
stl_names: dict[str, str] = dict(),
)
PRISM model text plus the identifier mapping used during export.
TLAExport
dataclass
¶
TLAExport(
module: str,
place_names: dict[str, str],
metric_names: dict[str, str],
transition_names: dict[str, str],
rule_names: dict[str, str] = dict(),
action_names: dict[str, str] = dict(),
)
TLA+ module text plus the identifier mapping used during export.
FormalCheckerAvailability
dataclass
¶
FormalCheckerAvailability(
property_name: str,
checker: str,
artifact_name: str,
executable: str,
command: tuple[str, ...],
available: bool,
resolved_path: str | None = None,
status: str = "missing_executable",
execution_permitted: bool = False,
)
Non-executing readiness record for one external checker command.
Methods:¶
__post_init__ ¶
Validate the non-executing checker availability contract.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
to_audit_record ¶
Return a JSON-safe non-executing checker readiness record.
Returns¶
dict[str, object] Return a JSON-safe non-executing checker readiness record.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
FormalCheckerResult
dataclass
¶
FormalCheckerResult(
property_name: str,
checker: str,
artifact_name: str,
package_hash: str,
result_hash: str,
status: str,
passed: bool,
detail: str = "",
execution_permitted: bool = False,
)
Reviewed external checker result bound to one package hash.
Results are audit records supplied by CI or a human-reviewed verification workflow after materialising the package outside this library. The constructor validates identity, checker kind, package hash, and result hash; it never executes checkers and never grants actuation.
Methods:¶
__post_init__ ¶
Validate reviewed checker result identity and fail-closed status.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
to_audit_record ¶
Return a JSON-safe external checker result record.
Returns¶
dict[str, object] Return a JSON-safe external checker result record.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
FormalRuntimeCertificate
dataclass
¶
FormalRuntimeCertificate(
certificate_name: str,
package_name: str,
package_hash: str,
runtime_bounds: dict[str, float],
checker_availability: tuple[
FormalCheckerAvailability, ...
],
checker_results: tuple[FormalCheckerResult, ...],
required_property_count: int,
passed_required_count: int,
missing_required_properties: tuple[str, ...],
failed_required_properties: tuple[str, ...],
unavailable_checker_properties: tuple[str, ...],
status: str,
certificate_hash: str,
actuation_permitted: bool = False,
)
Fail-closed runtime certificate for formal supervisor evidence.
A certificate binds a formal verification package, finite runtime bounds, checker readiness, and externally supplied checker results into one deterministic hash. A verified certificate is still non-actuating; it is an auditable precondition for operator review or a separate runtime monitor.
Methods:¶
__post_init__ ¶
Validate certificate integrity and non-actuating runtime status.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
to_audit_record ¶
Return a deterministic JSON-safe runtime certificate.
Returns¶
dict[str, object] Return a deterministic JSON-safe runtime certificate.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
FormalCheckerCommand
dataclass
¶
FormalCheckerCommand(
property_name: str,
checker: str,
artifact_name: str,
command: tuple[str, ...],
execution_permitted: bool = False,
)
External model-checker command manifest for one property.
Methods:¶
to_audit_record ¶
Return a JSON-safe external checker command record.
Returns¶
dict[str, object] Return a JSON-safe external checker command record.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/verification_package.py
FormalSafetyProperty
dataclass
¶
FormalSafetyProperty(
name: str,
artifact_name: str,
checker: str,
expression: str,
description: str = "",
required: bool = True,
)
Named model-checking property bound to one exported artefact.
Methods:¶
to_audit_record ¶
Return a JSON-safe formal property record.
Returns¶
dict[str, object] Return a JSON-safe formal property record.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/verification_package.py
FormalTextArtifact
dataclass
¶
Reviewed external proof artefact text for package manifests.
This object lets operators add already-reviewed Promela or SMT-LIB artefacts to the same deterministic package contract as generated PRISM/TLA exports. It records text only; it does not generate, write, or execute external checker inputs.
FormalVerificationPackage
dataclass
¶
FormalVerificationPackage(
package_name: str,
artifact_hashes: dict[str, str],
artifact_types: dict[str, str],
properties: tuple[FormalSafetyProperty, ...],
checker_commands: tuple[FormalCheckerCommand, ...],
package_hash: str,
)
Deterministic bundle for external formal-verification workflows.
Methods:¶
to_audit_record ¶
Return a JSON-safe package manifest.
Returns¶
dict[str, object] Return a JSON-safe package manifest.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/verification_package.py
Functions:¶
export_petri_net_prism ¶
export_petri_net_prism(
net: PetriNet,
initial_marking: Marking,
*,
module_name: str = "spo_petri",
max_tokens: int | None = None,
) -> PrismExport
Serialise a Petri net into a bounded PRISM MDP model.
Guard metrics become PRISM constants, so safety properties can be checked over scenario-specific metric assignments without changing the net model.
Parameters¶
net : PetriNet
The Petri net to export.
initial_marking : Marking
The initial Petri net marking.
module_name : str
Name of the emitted model-checker module.
max_tokens : int | None
Maximum token bound per place, or None.
Returns¶
PrismExport The bounded PRISM MDP export of the Petri net.
Raises¶
PolicyError If the net or bounds violate the export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/petri_export.py
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 | |
export_petri_net_tla ¶
export_petri_net_tla(
net: PetriNet,
initial_marking: Marking,
*,
module_name: str = "SpoPetri",
max_tokens: int | None = None,
) -> TLAExport
Serialise a Petri net into a bounded TLA+ transition-system module.
Guard metrics become TLA+ constants. Places become bounded natural-number variables, and each Petri transition becomes a named next-state action that preserves all unaffected places explicitly.
Parameters¶
net : PetriNet
The Petri net to export.
initial_marking : Marking
The initial Petri net marking.
module_name : str
Name of the emitted model-checker module.
max_tokens : int | None
Maximum token bound per place, or None.
Returns¶
TLAExport The bounded TLA+ export of the Petri net.
Raises¶
PolicyError If the net or bounds violate the export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/petri_export.py
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 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 | |
export_policy_rules_prism ¶
export_policy_rules_prism(
rules: list[PolicyRule],
*,
module_name: str = "spo_policy",
) -> PrismExport
Serialise policy rules into a bounded PRISM MDP model.
Metrics and current regime are model inputs represented as PRISM constants. Each rule has a bounded fire counter; unlimited rules are represented as one-shot reachability counters for model-checking queries.
Parameters¶
rules : list[PolicyRule] The policy rules to export or validate. module_name : str Name of the emitted model-checker module.
Returns¶
PrismExport The bounded PRISM MDP export of the policy rules.
Raises¶
PolicyError If the rules violate the export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/policy_export.py
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 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 | |
export_policy_rules_tla ¶
Serialise policy rules into a bounded TLA+ transition-system module.
Parameters¶
rules : list[PolicyRule] The policy rules to export or validate. module_name : str Name of the emitted model-checker module.
Returns¶
TLAExport The bounded TLA+ export of the policy rules.
Raises¶
PolicyError If the rules violate the export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/policy_export.py
312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 | |
audit_formal_checker_availability ¶
audit_formal_checker_availability(
package: FormalVerificationPackage,
*,
executable_paths: Mapping[str, str | None]
| None = None,
) -> tuple[FormalCheckerAvailability, ...]
Return non-executing external-checker readiness records.
The audit resolves only the first command token for each package checker
command. It never materialises artefacts, writes files, launches subprocesses,
or changes the package execution policy. Tests and CI may inject
executable_paths for deterministic readiness checks; production callers
can omit it to use shutil.which against the current host.
Parameters¶
package : FormalVerificationPackage
The formal verification package.
executable_paths : Mapping[str, str | None] | None
Mapping of checker name to executable path, or None.
Returns¶
tuple[FormalCheckerAvailability, ...] The non-executing external-checker readiness records.
Raises¶
PolicyError If the package fails its fail-closed policy checks.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
build_runtime_control_certificate ¶
build_runtime_control_certificate(
package: FormalVerificationPackage,
checker_availability: Sequence[
FormalCheckerAvailability
],
checker_results: Sequence[FormalCheckerResult],
runtime_bounds: Mapping[str, object],
*,
certificate_name: str = "spo-runtime-control-certificate",
) -> FormalRuntimeCertificate
Build a fail-closed runtime certificate from formal evidence.
The certificate is verified only when every required package property has a matching available checker and a passed external result bound to the current package hash. Missing, failed, stale, or unavailable evidence produces a blocked certificate. The returned record never permits actuation.
Parameters¶
package : FormalVerificationPackage The formal verification package. checker_availability : Sequence[FormalCheckerAvailability] External-checker readiness records. checker_results : Sequence[FormalCheckerResult] External-checker result records. runtime_bounds : Mapping[str, object] Finite runtime bounds for the certificate. certificate_name : str Name for the emitted certificate.
Returns¶
FormalRuntimeCertificate The fail-closed runtime control certificate.
Raises¶
PolicyError If the formal evidence fails the fail-closed policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/runtime_certificate.py
372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 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 | |
export_policy_rules_smt ¶
export_policy_rules_smt(
rules: list[PolicyRule],
*,
module_name: str = "spo_policy",
) -> FormalTextArtifact
Serialise policy rules into a bounded SMT-LIB feasibility model.
The export declares the active regime, metric inputs, bounded rule-fire counters, rule firing predicates, and action emission predicates. The final assertion asks an SMT solver whether at least one policy rule can fire under the declared constraints. The function only generates deterministic text; it does not invoke Z3 or any other solver.
Parameters¶
rules : list[PolicyRule] The policy rules to export or validate. module_name : str Name recorded in the emitted SMT-LIB comments.
Returns¶
FormalTextArtifact An SMT-LIB v2 artifact suitable for package hashing and Z3 execution.
Raises¶
PolicyError If the rules violate the shared formal-export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/smt_export.py
83 84 85 86 87 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 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 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 | |
export_stl_specs_prism ¶
export_stl_specs_prism(
specs: list[PolicySTLSpec],
*,
module_name: str = "spo_stl",
) -> PrismExport
Serialise policy-declared STL monitors into PRISM label surfaces.
This export covers the builtin STL subset used by STLMonitor:
always (...) and eventually (...) over numeric predicate
conjunctions. The model is a single-state abstraction with signal
constants and per-monitor satisfied/violated labels for property checks.
Parameters¶
specs : list[PolicySTLSpec] Policy-declared STL monitor specifications. module_name : str Name of the emitted model-checker module.
Returns¶
PrismExport The PRISM label-surface export of the STL monitors.
Raises¶
PolicyError If the STL specs violate the export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/stl_export.py
build_formal_verification_package ¶
build_formal_verification_package(
artifacts: Mapping[
str, PrismExport | TLAExport | FormalTextArtifact
],
properties: Sequence[FormalSafetyProperty],
*,
package_name: str = "spo-formal-verification",
) -> FormalVerificationPackage
Build a deterministic manifest for external model-checker execution.
The package records exported artefact hashes, property-library entries, and exact checker commands. It never writes files or invokes external tools; CI or operators can materialise the package and run the recorded commands in a controlled environment.
Parameters¶
artifacts : Mapping[str, PrismExport | TLAExport | FormalTextArtifact] Mapping of artefact name to its formal export. properties : Sequence[FormalSafetyProperty] The formal safety properties. package_name : str Name for the verification package.
Returns¶
FormalVerificationPackage The formal verification package manifest.
Raises¶
PolicyError If the artefacts or properties fail policy checks.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/verification_package.py
264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 | |
smt_export ¶
SMT-LIB text exporter for bounded supervisor policy rule feasibility.
Classes¶
Functions:¶
export_policy_rules_smt ¶
export_policy_rules_smt(
rules: list[PolicyRule],
*,
module_name: str = "spo_policy",
) -> FormalTextArtifact
Serialise policy rules into a bounded SMT-LIB feasibility model.
The export declares the active regime, metric inputs, bounded rule-fire counters, rule firing predicates, and action emission predicates. The final assertion asks an SMT solver whether at least one policy rule can fire under the declared constraints. The function only generates deterministic text; it does not invoke Z3 or any other solver.
Parameters¶
rules : list[PolicyRule] The policy rules to export or validate. module_name : str Name recorded in the emitted SMT-LIB comments.
Returns¶
FormalTextArtifact An SMT-LIB v2 artifact suitable for package hashing and Z3 execution.
Raises¶
PolicyError If the rules violate the shared formal-export policy.
Source code in src/scpn_phase_orchestrator/supervisor/formal_export/smt_export.py
83 84 85 86 87 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 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 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 | |
export_petri_net_to_prism renders a guard-gated Petri net as a finite PRISM
MDP model, preserving the runtime engine's first-enabled transition priority and
exposing guard metrics as PRISM constants for scenario binding.
formal ¶
Formal verification exporters for guard-gated Petri nets.
Classes¶
Functions:¶
export_petri_net_to_prism ¶
export_petri_net_to_prism(
net: PetriNet,
initial: Marking,
*,
module_name: str = "supervisor",
max_tokens: int | None = None,
include_idle: bool = True,
) -> str
Export a guard-gated Petri net as a finite PRISM MDP model.
The exporter preserves the runtime engine's first-enabled transition priority by blocking each command when an earlier transition is enabled. Guard metrics become PRISM constants so verification jobs can bind them explicitly for a scenario.
Parameters¶
net : PetriNet
Guard-gated Petri net to export; its transitions define the commands
and their first-enabled priority order.
initial : Marking
Initial token count for each place.
module_name : str
Name of the generated PRISM module; sanitised to a valid identifier.
max_tokens : int or None
Upper bound on tokens per place. When None, a bound is derived from
the net structure and the initial marking.
include_idle : bool
When True, emit an [idle] self-loop that fires only when no
transition is enabled, keeping the MDP deadlock-free.
Returns¶
str The PRISM MDP model source.
Raises¶
PolicyError
If max_tokens is less than 1, or if an initial marking exceeds the
token bound.
Source code in src/scpn_phase_orchestrator/supervisor/formal.py
24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 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 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 114 115 116 117 118 119 | |
Rust Supervisor Backend Probe¶
The Python supervisor remains the default runtime-control surface. The optional
Rust spo-supervisor PyO3 bindings are validated separately through
audit_rust_supervisor_backend(), which checks required spo_kernel symbols
and runs deterministic, non-actuating smoke checks for regime classification,
boundary observation, and coherence monitoring. spo doctor reports this as
the optional rust-supervisor backend so operators can diagnose a missing or
malformed Rust supervisor FFI without changing live-control behavior.
rust_backend ¶
Optional Rust supervisor FFI readiness checks.
The live Python supervisor remains the default runtime-control path. This module
answers a narrower packaging and operations question: when the optional
spo_kernel wheel is installed, does it expose the spo-supervisor PyO3
surface and do the deterministic regime, boundary, and coherence primitives
behave like usable supervisor components?
All probes fail closed. Missing symbols, malformed smoke outputs, or import failures produce an unavailable status and never mutate production supervisor state.
Classes¶
RustSupervisorBackendStatus
dataclass
¶
RustSupervisorBackendStatus(
available: bool,
symbols: tuple[str, ...],
missing_symbols: tuple[str, ...],
detail: str,
smoke: Mapping[str, object] = dict(),
)
Readiness outcome for the optional Rust supervisor FFI backend.
Attributes¶
available: True only when the module imports, all required symbols are
present, and deterministic smoke validation succeeds.
symbols: Required PyO3 symbols checked on the module.
missing_symbols: Required symbols absent from the inspected module.
detail: Human-facing readiness explanation for ``spo doctor`` and audit
records.
smoke: Deterministic smoke observations when validation succeeds.
Attributes¶
status
property
¶
Return ok when available, otherwise warn.
Returns¶
str
ok for a usable optional backend and warn for an unavailable
optional backend.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-serialisable backend audit record.
Returns¶
dict[str, object] Backend readiness details suitable for doctor JSON output, release evidence, or internal handoff logs.
Source code in src/scpn_phase_orchestrator/supervisor/rust_backend.py
Functions:¶
audit_rust_supervisor_backend ¶
Probe the optional spo_kernel supervisor FFI surface.
Parameters¶
module : object | None
Optional module-like object used by tests. When None, spo_kernel
is imported lazily.
Returns¶
RustSupervisorBackendStatus Fail-closed readiness status for the optional Rust supervisor backend.
Source code in src/scpn_phase_orchestrator/supervisor/rust_backend.py
Petri Net FSM¶
Formal Petri net state machine enabling formal verification of safety properties: deadlock freedom, liveness, bounded token counts.
Components¶
| Class | Fields | Description |
|---|---|---|
Place |
name: str |
Token container (regime state) |
Arc |
place: str, weight: int |
Token flow edge |
Guard |
metric: str, op: str, threshold: float |
Firing condition |
Transition |
name, inputs, outputs, guard |
Guarded state change |
Marking |
tokens: dict[str, int] |
Current token distribution |
Guard operators¶
Guards support five comparison operators: >, >=, <, <=, ==.
Guard.evaluate(ctx) checks the condition against a context dictionary.
PetriNet methods¶
| Method | Description |
|---|---|
enabled(marking, ctx) |
Returns transitions whose guards pass |
fire(marking, transition) |
Moves tokens and returns new marking |
step(marking, ctx) |
Fires first enabled transition |
parse_guard("R < 0.3") parses a string into a Guard object.
Performance: enabled_transitions() < 10 μs.
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¶
Guard
dataclass
¶
Boolean guard condition on a named metric (e.g. 'stability_proxy > 0.6').
Methods:¶
evaluate ¶
Return True if the guard condition is satisfied by ctx.
Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py
Transition
dataclass
¶
Petri net transition with input/output arcs and optional guard.
Marking
dataclass
¶
Token distribution across places in a Petri net.
Methods:¶
active_places ¶
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
PetriNet ¶
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
Attributes¶
place_names
property
¶
All place names registered in this net.
Returns¶
frozenset[str] All place names registered in this net.
transitions
property
¶
All transitions in firing-priority order.
Returns¶
list[Transition] All transitions in firing-priority order.
guard_metrics
property
¶
Whitelisted context metric names used by transition guards.
Returns¶
frozenset[str] Whitelisted context metric names used by transition guards.
Methods:¶
enabled ¶
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
fire ¶
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
step ¶
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
Functions:¶
parse_guard ¶
Parse guard string like 'stability_proxy > 0.6'.
Source code in src/scpn_phase_orchestrator/supervisor/petri_net.py
Petri Net Adapter¶
Bridge between UPDEState and the Petri net FSM.
PetriNetAdapter(
net: PetriNet,
initial_marking: Marking,
place_to_regime: dict[str, str], # maps place names to regime names
event_bus: EventBus | None = None,
)
adapter.step(ctx) evaluates the Petri net with the given context and
returns the current Regime based on which place holds the token.
petri_adapter ¶
Adapter from guarded Petri-net markings into supervisor regime decisions.
PetriNetAdapter validates the Petri net, initial marking, place-to-regime
mapping, optional event bus, and finite metric context before stepping. It fires
at most one transition through the underlying net, emits a transition event when
configured, and maps active places to the highest-severity regime. It does not
emit control actions directly.
Classes¶
PetriNetAdapter ¶
PetriNetAdapter(
net: PetriNet,
initial_marking: Marking,
place_to_regime: dict[str, str],
event_bus: EventBus | None = None,
)
Map Petri net markings to Regime values.
Each place in the net maps to a Regime via place_to_regime. When multiple places are marked, the highest-severity regime wins (CRITICAL > RECOVERY > DEGRADED > NOMINAL).
Source code in src/scpn_phase_orchestrator/supervisor/petri_adapter.py
Attributes¶
marking
property
¶
Current Petri net marking (token distribution).
Returns¶
Marking Current Petri net marking (token distribution).
net
property
¶
Methods:¶
step ¶
Advance the Petri net one step and return the active regime.
Parameters¶
ctx : dict[str, float] Context metric values keyed by guard-metric name.
Returns¶
Regime The active regime after advancing the net one step.
Source code in src/scpn_phase_orchestrator/supervisor/petri_adapter.py
Event Bus¶
Publish-subscribe system for supervisor events.
RegimeEvent (frozen dataclass)¶
| Field | Type | Description |
|---|---|---|
kind |
str |
"regime_transition" or "boundary_violation" |
step |
int |
Step number when event occurred |
detail |
str |
Human-readable description |
EventBus¶
bus = EventBus(maxlen=200)
bus.subscribe(callback)
bus.post(RegimeEvent(kind="regime_transition", step=42, detail="nominal->degraded"))
bus.history # list of all events
bus.count # total events posted
Events are stored in a bounded deque (default 200). Subscribers are
called synchronously on post().
events ¶
Validated supervisor event records and an in-process bounded event bus.
RegimeEvent restricts event kinds and step/detail fields before publication,
and EventBus records a bounded chronological history while notifying
callable subscribers synchronously. The bus is process-local and passive: it
does not spawn threads, persist logs, retry subscriber failures, or emit network
traffic.
Classes¶
RegimeEvent
dataclass
¶
Immutable event emitted on regime transitions or boundary breaches.
EventBus ¶
Pub/sub bus for regime events with bounded history.
Source code in src/scpn_phase_orchestrator/supervisor/events.py
Attributes¶
history
property
¶
Chronological list of all posted events.
Returns¶
list[RegimeEvent] Chronological list of all posted events.
Methods:¶
subscribe ¶
Register a callback to receive future events.
Parameters¶
callback : object A callable invoked with each posted event.
Raises¶
ValueError
If callback is not callable.
Source code in src/scpn_phase_orchestrator/supervisor/events.py
unsubscribe ¶
Remove a previously registered callback.
Parameters¶
callback : object A callable invoked with each posted event.
Source code in src/scpn_phase_orchestrator/supervisor/events.py
post ¶
Record event in history and notify all subscribers.
Parameters¶
event : RegimeEvent The regime event to record and broadcast.
Raises¶
ValueError
If event is not a RegimeEvent.
Source code in src/scpn_phase_orchestrator/supervisor/events.py
Model-Predictive Controller (MPC)¶
Anticipatory control using Ott-Antonsen mean-field reduction.
Prediction (dataclass)¶
| Field | Type | Description |
|---|---|---|
R_predicted |
list[float] |
Predicted R trajectory (horizon steps) |
will_degrade |
bool |
R predicted to cross DEGRADED threshold |
will_critical |
bool |
R predicted to cross CRITICAL threshold |
steps_to_degradation |
int |
Steps until predicted degradation |
PredictiveSupervisor¶
PredictiveSupervisor(
n_oscillators: int,
dt: float,
horizon: int = 10, # prediction steps ahead
divergence_threshold: float = 0.3, # OA model trust threshold
)
Methods:
predict(phases, omegas, knm, alpha) → Prediction— runs OA forward model forhorizonsteps, returns trajectorydecide(phases, omegas, knm, alpha, upde_state, boundary_state) → list[ControlAction]— predicts then acts if degradation imminent
phases, omegas, knm, and alpha are finite real-valued arrays. Boolean
aliases and complex/object-complex payloads are rejected before OA prediction so
the forward model cannot silently reinterpret non-physical inputs as real
oscillator states, frequencies, coupling, or phase-lag matrices.
Safety fallback¶
When |R_predicted - R_measured| > divergence_threshold, the MPC
discards its prediction and falls back to reactive control. This
prevents acting on a forward model that has lost accuracy.
Computational advantage¶
The OA reduction is O(1) per step (single complex ODE) versus O(N) for the full Kuramoto model. For N=1000 oscillators with horizon=10, MPC prediction costs ~10 ODE steps versus 10000 Euler steps.
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
¶
Return True when observed coherence exceeds the target.
Returns¶
bool Return True when observed coherence exceeds the target.
Methods:¶
to_audit_record ¶
Return a serialisable audit payload.
Returns¶
dict[str, float] Return a serialisable audit payload.
Source code in src/scpn_phase_orchestrator/supervisor/predictive.py
FEPHierarchyChildAssessment
dataclass
¶
FEPHierarchyChildAssessment(
name: str,
assessment: FEPPredictionAssessment,
actions: tuple[ControlAction, ...],
)
Assessment for one child node in a hierarchical FEP supervisor.
Methods:¶
to_audit_record ¶
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
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 ¶
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
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
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
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
257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 | |
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
Attributes¶
target_R
property
¶
Target order parameter used by the free-energy controller.
Returns¶
float Target order parameter used by the free-energy controller.
last_assessment
property
¶
Most recent free-energy assessment, if assess has run.
Returns¶
FEPPredictionAssessment | None
Most recent free-energy assessment, if assess has run.
Methods:¶
assess ¶
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
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
reset ¶
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
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 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 | |
FEP Predictive Supervisor¶
FEPPredictiveSupervisor is the first Python supervisor mode that uses the
existing VariationalPredictor as an auditable free-energy signal. It observes
the current phase vector, updates the variational predictor, and emits bounded
zeta / Psi actions only when free energy, prediction error, or stability
proxy thresholds indicate a pre-emptive correction is needed.
from scpn_phase_orchestrator.supervisor import (
FEPPredictiveSupervisor,
assess_fep_hierarchy,
)
fep = FEPPredictiveSupervisor(
n_oscillators=len(phases),
dt=0.01,
target_R=0.8,
free_energy_threshold=1.0,
)
assessment = fep.assess(phases, omegas)
actions = fep.decide(phases, omegas, upde_state, boundary_state)
audit_payload = assessment.to_audit_record()
FEPPredictionAssessment records free energy, complexity, mean absolute
prediction error, precision statistics, observed and predicted order
parameters, target R, and a scalar surprise proxy. This keeps the FEP path
reviewable in the same audit trail as policy, causal, STL, and topology
decisions.
This slice is intentionally conservative: it is a FEP-Kuramoto correspondence controller over the existing variational predictor, not a claim of a complete biological active-inference agent.
assess_fep_hierarchy() is the reusable hierarchy primitive. It runs one child
FEPPredictiveSupervisor per named child observation, reduces each child's
observed coherence into a parent phase vector, then runs a parent
FEPPredictiveSupervisor over the reduced child state. The returned
FEPHierarchyAssessment records child assessments, child actions, parent
assessment, parent actions, child R values, and parent phase encoding.
Child phase and frequency observations use the same finite real-valued boundary
contract as the single-supervisor path.
hierarchy = assess_fep_hierarchy(
{
"generation_area": (generation_phases, generation_omegas),
"demand_area": (demand_phases, demand_omegas),
},
dt=0.01,
parent_dt=0.1,
)
audit_hierarchy = hierarchy.to_audit_record()
Domainpack hierarchy proofs:
domainpacks/power_grid/fep_hierarchy_demo.pyruns generation and demand/renewable child regions into a parent grid supervisor.domainpacks/cardiac_rhythm/fep_hierarchy_demo.pyruns pacemaker/atrial and ventricular/recovery child axes into a parent cardiac supervisor.
Performance summary¶
| Operation | Budget | Notes |
|---|---|---|
RegimeManager.evaluate() |
< 10 μs | Pure Python comparison |
SupervisorPolicy.decide() |
< 50 μs | Rule evaluation + action construction |
PetriNet.enabled() |
< 10 μs | Guard evaluation |
PredictiveSupervisor.predict() |
< 1 ms | OA mean-field (10 complex ODE steps) |
EventBus.post() |
< 5 μs | Synchronous dispatch |
Active Inference Agent¶
The ActiveInferenceAgent provides a predictive candidate-generation
framework based on a Variational Free Energy objective. It is an optional
Rust-backed research surface; its outputs still require the normal policy,
projection, audit, and operator-review boundaries.
Mathematical Model¶
The agent maintains a low-dimensional internal state \(x\) and minimises a Variational Free Energy objective \(F\) between its prediction \(\hat{R}\) and the observed coherence \(R_{\mathrm{obs}}\):
The agent proposes a forcing strength \(\zeta\) and reference phase \(\Psi\) for a configured target coherence \(R_{\mathrm{target}}\). “Optimal” is relative to the implemented local objective and does not establish domain-level optimality or safe actuation.
Features¶
- Candidate suppression: can propose anti-phase driving (\(\Psi = \psi + \pi\)) against a configured coherence objective.
- Rust implementation: available through the optional
spo_kernelFFI; no portable latency or hard-real-time guarantee is claimed. - Prediction-error adaptation: updates its internal state in response to observed divergence; robustness to domain drift requires separate evidence.
Rust-only module
ActiveInferenceAgent is implemented in spo-kernel (Rust crate spo-supervisor::active_inference).
Python access via spo_kernel.PyActiveInferenceAgent when the FFI is installed.
Evolutionary Review Surfaces¶
Offline-only evolutionary search, grammar, policy DSL, topology mutation, and example builders used for non-actuating supervisor review workflows.
evolutionary_examples ¶
Deterministic examples for offline evolutionary supervisor policy search.
Functions:¶
build_evolutionary_supervisor_search_examples ¶
Return deterministic offline-search example inputs for reference gates.
Returns¶
tuple[dict[str, object], ...] Return deterministic offline-search example inputs for reference gates.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_examples.py
build_evolutionary_supervisor_search_examples_from_worker_a_api ¶
build_evolutionary_supervisor_search_examples_from_worker_a_api() -> (
tuple[dict[str, object], ...]
)
Return examples enriched with core offline-search report counts.
Returns¶
tuple[dict[str, object], ...] Return examples enriched with core offline-search report counts.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_examples.py
evolutionary_petri_grammar ¶
Review-only offline evolutionary mutation grammar for Petri-net topologies.
This module produces deterministic mutation candidates and plans from a simple, normalised net descriptor. It performs no execution, no actuation, and never commits graph changes itself.
Classes¶
EvolutionaryPetriMutationConfig
dataclass
¶
EvolutionaryPetriMutationConfig(
generation_count: int = 2,
candidates_per_generation: int = 6,
mutation_step: float = 0.1,
max_arc_weight: int = 4,
max_token_bound: int = 128,
)
Mutation generation configuration for the offline Petri grammar.
EvolutionaryPetriMutationCandidate
dataclass
¶
EvolutionaryPetriMutationCandidate(
candidate_id: str,
generation: int,
mutation_type: MutationType,
mutation_target: str,
mutation_kind: str,
blocked_reasons: tuple[str, ...],
before: dict[str, object],
after: dict[str, object],
mutation_delta: float,
candidate_hash: str,
operator_review_required: bool = True,
execution_disabled: bool = True,
live_merge_permitted: bool = False,
hot_patch_permitted: bool = False,
actuation_permitted: bool = False,
)
One offline-only Petri-net mutation candidate.
Attributes¶
accepted
property
¶
Return whether this candidate is accepted for review.
Returns¶
bool Return whether this candidate is accepted for review.
status
property
¶
Return the review status label for this candidate.
Returns¶
str Return the review status label for this candidate.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_petri_grammar.py
EvolutionaryPetriMutationPlan
dataclass
¶
EvolutionaryPetriMutationPlan(
schema_name: str,
schema_version: str,
config: EvolutionaryPetriMutationConfig,
source_net_hash: str,
candidate_count: int,
accepted_count: int,
rejected_count: int,
candidates: tuple[
EvolutionaryPetriMutationCandidate, ...
],
best_candidate_id: str | None,
source_net: dict[str, object],
operator_review_required: bool,
execution_disabled: bool,
live_merge_permitted: bool,
hot_patch_permitted: bool,
actuation_permitted: bool,
non_actuating: bool,
plan_hash: str,
)
Deterministic, offline review plan for Petri-net grammar search.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_petri_grammar.py
Functions:¶
run_offline_evolutionary_petri_mutation_grammar ¶
run_offline_evolutionary_petri_mutation_grammar(
net_like: Mapping[str, object] | Sequence[object],
*,
generation_count: int = 2,
candidates_per_generation: int = 6,
mutation_step: float = 0.1,
max_arc_weight: int = 4,
max_token_bound: int = 128,
) -> EvolutionaryPetriMutationPlan
Build a deterministic review-only mutation plan from a net-like payload.
Parameters¶
net_like : Mapping[str, object] | Sequence[object] A net-like payload describing places, transitions, and arcs. generation_count : int Number of search generations. candidates_per_generation : int Number of candidates evaluated per generation. mutation_step : float Mutation step size applied per generation. max_arc_weight : int Maximum arc weight allowed in a mutated net. max_token_bound : int Maximum token count allowed per place.
Returns¶
EvolutionaryPetriMutationPlan The review-only Petri mutation plan.
Raises¶
ValueError If the net-like payload or bounds are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_petri_grammar.py
304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 | |
evolutionary_policy_dsl ¶
Policy DSL mutation helpers for offline evolutionary supervisor review.
Classes¶
PolicyCondition
dataclass
¶
PolicyAction
dataclass
¶
PolicyRule
dataclass
¶
Policy rule composed from conditions and actions.
Methods:¶
to_dsl ¶
Return the deterministic policy DSL representation.
Returns¶
str Return the deterministic policy DSL representation.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_policy_dsl.py
PolicyMutationSearchConfig
dataclass
¶
PolicyMutationSearchConfig(
generation_count: int = 2,
population_size: int = 6,
mutation_step: float = 0.05,
)
Configuration for deterministic policy mutation search.
PolicyMutationPlan
dataclass
¶
PolicyMutationPlan(
rule_name: str,
component: str,
component_index: int,
operator: str,
original_value: float,
mutated_value: float,
mutation_delta: float,
)
Planned mutation candidate for policy DSL review.
PolicyMutationCandidate
dataclass
¶
PolicyMutationCandidate(
candidate_id: str,
generation: int,
mutation_index: int,
source_rule_name: str,
source_rule_text: str,
mutated_rule_text: str,
candidate_policy_dsl: str,
mutation_plan: PolicyMutationPlan,
blocked_reasons: tuple[str, ...],
candidate_hash: str,
operator_review_required: bool = True,
execution_disabled: bool = True,
live_merge_permitted: bool = False,
hot_patch_permitted: bool = False,
actuation_permitted: bool = False,
)
One non-actuating policy mutation candidate.
Attributes¶
accepted
property
¶
Return whether this candidate is accepted for review.
Returns¶
bool Return whether this candidate is accepted for review.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, Any] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_policy_dsl.py
PolicyMutationSearchReport
dataclass
¶
PolicyMutationSearchReport(
schema_name: str,
schema_version: str,
config: PolicyMutationSearchConfig,
source_policy_dsl: str,
source_policy_hash: str,
candidate_count: int,
accepted_count: int,
rejected_count: int,
candidates: tuple[PolicyMutationCandidate, ...],
execution_disabled: bool,
hot_patch_permitted: bool,
live_merge_permitted: bool,
actuation_permitted: bool,
operator_review_required: bool,
non_actuating: bool,
report_hash: str,
)
Aggregate report for policy mutation search.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, Any] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_policy_dsl.py
Functions:¶
parse_policy_dsl ¶
Parse immutable rule objects from a compact policy DSL string.
Parameters¶
policy_dsl : str A compact policy-DSL source string.
Returns¶
tuple[PolicyRule, ...] The immutable policy rules parsed from the DSL.
Raises¶
ValueError If the DSL string is malformed.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_policy_dsl.py
run_offline_evolutionary_policy_dsl_search ¶
run_offline_evolutionary_policy_dsl_search(
policy_dsl: str,
*,
generation_count: int = 2,
population_size: int = 6,
mutation_step: float = 0.05,
) -> PolicyMutationSearchReport
Generate deterministic offline policy-DSl mutation candidates for review.
Parameters¶
policy_dsl : str A compact policy-DSL source string. generation_count : int Number of search generations. population_size : int Number of candidates per generation. mutation_step : float Mutation step size applied per generation.
Returns¶
PolicyMutationSearchReport The offline policy-DSL mutation search report.
Raises¶
ValueError If the DSL string or search parameters are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_policy_dsl.py
333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 | |
evolutionary_search ¶
Deterministic offline evolutionary supervisor policy search.
Classes¶
EvolutionarySearchConfig
dataclass
¶
EvolutionarySearchConfig(
generation_count: int = 2,
population_size: int = 8,
mutation_step: float = 0.05,
minimum_replay_reward: float = 0.0,
minimum_safety_margin: float = 0.0,
)
Configuration for deterministic offline candidate evolution.
EvolutionaryCandidate
dataclass
¶
EvolutionaryCandidate(
candidate_id: str,
generation: int,
knob: str,
parent_value: float,
candidate_value: float,
mutation_delta: float,
genome: tuple[tuple[str, float], ...],
replay_fitness: float,
stl_robustness: float,
stl_satisfied: bool,
replay_violation_count: int,
blocked_reasons: tuple[str, ...],
candidate_hash: str,
review_required: bool = True,
live_merge_permitted: bool = False,
hot_patch_permitted: bool = False,
actuation_permitted: bool = False,
)
One offline candidate snapshot from a deterministic mutation step.
Attributes¶
accepted
property
¶
Whether the candidate passed all guard and replay gates.
Returns¶
bool Whether the candidate passed all guard and replay gates.
status
property
¶
Methods:¶
to_audit_record ¶
Return JSON-safe candidate evidence for audit transport.
Returns¶
dict[str, object] Return JSON-safe candidate evidence for audit transport.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_search.py
EvolutionarySearchReport
dataclass
¶
EvolutionarySearchReport(
schema_name: str,
schema_version: str,
config: EvolutionarySearchConfig,
parent_policy_hash: str,
replay_summary: _ReplaySummary,
stl_spec: str,
stl_monitoring: dict[str, object],
candidate_count: int,
accepted_count: int,
rejected_count: int,
candidates: tuple[EvolutionaryCandidate, ...],
best_candidate: EvolutionaryCandidate | None,
claim_boundary: str,
non_actuating: bool,
execution_disabled: bool,
hot_patch_permitted: bool,
live_merge_permitted: bool,
operator_review_required: bool,
report_hash: str,
)
Deterministic, offline-only audit report for evolutionary search.
Methods:¶
to_audit_record ¶
Return a JSON-safe audit record for review tooling.
Returns¶
dict[str, object] Return a JSON-safe audit record for review tooling.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_search.py
Functions:¶
run_offline_evolutionary_supervisor_search ¶
run_offline_evolutionary_supervisor_search(
parent_policy: Mapping[str, object],
audit_replays: Sequence[Mapping[str, object]],
*,
stl_spec: str,
trace: Mapping[str, Sequence[object]],
generation_count: int = 2,
population_size: int = 8,
mutation_step: float = 0.05,
minimum_replay_reward: float = 0.0,
minimum_safety_margin: float = 0.0,
) -> EvolutionarySearchReport
Run deterministic offline evolutionary policy mutation search.
Returns review-only candidates plus guards that block any live merge/hot patch.
Parameters¶
parent_policy : Mapping[str, object] The parent policy genome. audit_replays : Sequence[Mapping[str, object]] Audit replay records used to score candidates. stl_spec : str An STL specification string used as a safety gate. trace : Mapping[str, Sequence[object]] Signal trace keyed by variable name, each a sequence of floats. generation_count : int Number of search generations. population_size : int Number of candidates per generation. mutation_step : float Mutation step size applied per generation. minimum_replay_reward : float Minimum replay reward a candidate must reach. minimum_safety_margin : float Minimum safety margin a candidate must preserve.
Returns¶
EvolutionarySearchReport The offline evolutionary search report.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_search.py
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 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 | |
evolutionary_topology_grammar ¶
Offline review-only topology mutation grammar.
This module provides deterministic candidates for topology mutation operations. All generated candidates are review-only and include stable audit records.
Classes¶
TopologyMutationConfig
dataclass
¶
TopologyMutationConfig(
generation_count: int = 2,
population_size: int = 8,
mutation_step: float = 0.05,
min_edge_weight: float = 0.0,
max_edge_weight: float = 10.0,
edge_add_base_weight: float = 0.4,
max_add_candidates: int = 16,
)
Knobs used to shape deterministic grammar expansion.
TopologyMutationNode
dataclass
¶
TopologyMutationEdge
dataclass
¶
Normalised pairwise edge record.
Attributes¶
pair
property
¶
Return the canonical undirected edge pair.
Returns¶
tuple[int, int] Return the canonical undirected edge pair.
Methods:¶
TopologyMutationPlan
dataclass
¶
TopologyMutationPlan(
operation: str,
node_a: int,
node_b: int,
source_weight: float,
candidate_weight: float | None,
mutation_delta: float,
source_communities: tuple[str | None, str | None],
)
One planned grammar mutation.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_topology_grammar.py
TopologyMutationCandidate
dataclass
¶
TopologyMutationCandidate(
candidate_id: str,
generation: int,
mutation_index: int,
source_topology_hash: str,
plan: TopologyMutationPlan,
source_edge_count: int,
candidate_edges: tuple[TopologyMutationEdge, ...],
blocked_reasons: tuple[str, ...],
candidate_hash: str,
operator_review_required: bool = True,
execution_disabled: bool = True,
live_merge_permitted: bool = False,
hot_patch_permitted: bool = False,
actuation_permitted: bool = False,
)
One review-only topology candidate.
Attributes¶
accepted
property
¶
Return whether this candidate is accepted for review.
Returns¶
bool Return whether this candidate is accepted for review.
status
property
¶
Return the review status label for this candidate.
Returns¶
str Return the review status label for this candidate.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_topology_grammar.py
TopologyMutationReport
dataclass
¶
TopologyMutationReport(
schema_name: str,
schema_version: str,
config: TopologyMutationConfig,
source_topology_hash: str,
node_records: tuple[TopologyMutationNode, ...],
edge_records: tuple[TopologyMutationEdge, ...],
candidate_count: int,
accepted_count: int,
rejected_count: int,
candidates: tuple[TopologyMutationCandidate, ...],
claim_boundary: str,
operator_review_required: bool,
non_actuating: bool,
execution_disabled: bool,
hot_patch_permitted: bool,
live_merge_permitted: bool,
actuation_permitted: bool,
report_hash: str,
)
Offline-only topology mutation audit report.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_topology_grammar.py
Functions:¶
run_offline_evolutionary_topology_mutation_search ¶
run_offline_evolutionary_topology_mutation_search(
node_records: Sequence[Mapping[str, object]],
edge_records: Sequence[Mapping[str, object]],
*,
generation_count: int = 2,
population_size: int = 8,
mutation_step: float = 0.05,
min_edge_weight: float = 0.0,
max_edge_weight: float = 10.0,
edge_add_base_weight: float = 0.4,
max_add_candidates: int = 16,
) -> TopologyMutationReport
Generate deterministic offline topology mutation candidates.
Parameters¶
node_records : Sequence[Mapping[str, object]] Topology node records. edge_records : Sequence[Mapping[str, object]] Topology edge records. generation_count : int Number of search generations. population_size : int Number of candidates per generation. mutation_step : float Mutation step size applied per generation. min_edge_weight : float Minimum retained edge weight. max_edge_weight : float Maximum allowed edge weight. edge_add_base_weight : float Base weight assigned to newly added edges. max_add_candidates : int Maximum number of edge-addition candidates.
Returns¶
TopologyMutationReport The offline topology mutation report.
Raises¶
ValueError If the node/edge records or bounds are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/evolutionary_topology_grammar.py
279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 | |
Federated Review Surfaces¶
Federated orchestration, differential-privacy noise service, secure aggregation, and transport manifests. These APIs produce audit material and deployment preflight evidence without exporting raw local data.
federated ¶
Review-only federated policy-gradient aggregation manifests.
Classes¶
FederatedAggregationConfig
dataclass
¶
FederatedAggregationConfig(
clipping_norm: float = 1.0,
noise_multiplier: float = 1.0,
epsilon: float = 3.0,
delta: float = 1e-06,
min_node_count: int = 3,
)
Privacy and acceptance bounds for one offline aggregation review.
FederatedNodeUpdate
dataclass
¶
FederatedNodeUpdate(
node_id: str,
policy_delta: tuple[tuple[str, float], ...],
sample_count: int,
local_loss: float,
previous_audit_hash: str,
privacy_epsilon_spent: float,
clipped_l2_norm: float,
clip_scale: float,
accepted: bool,
rejection_reasons: tuple[str, ...],
update_hash: str,
)
Validated node-local policy-gradient update without raw time-series.
Methods:¶
to_audit_record ¶
Return JSON-safe node update evidence.
Returns¶
dict[str, object] Return JSON-safe node update evidence.
Source code in src/scpn_phase_orchestrator/supervisor/federated.py
FederatedPolicyAggregationReport
dataclass
¶
FederatedPolicyAggregationReport(
schema_name: str,
schema_version: str,
config: FederatedAggregationConfig,
required_policy_keys: tuple[str, ...],
node_updates: tuple[FederatedNodeUpdate, ...],
accepted_node_count: int,
rejected_node_count: int,
total_sample_count: int,
aggregate_delta: tuple[tuple[str, float], ...],
aggregate_hash: str,
privacy_budget_spent: float,
privacy_budget_remaining: float,
raw_time_series_received: bool,
claim_boundary: str,
operator_review_required: bool,
non_actuating: bool,
execution_disabled: bool,
live_transport_permitted: bool,
raw_data_export_permitted: bool,
actuation_permitted: bool,
report_hash: str,
)
Offline federated aggregation report with explicit safety boundaries.
Methods:¶
to_audit_record ¶
Return JSON-safe aggregate evidence.
Returns¶
dict[str, object] Return JSON-safe aggregate evidence.
Source code in src/scpn_phase_orchestrator/supervisor/federated.py
Functions:¶
build_federated_meta_orchestrator_manifest ¶
build_federated_meta_orchestrator_manifest(
node_updates: Sequence[Mapping[str, object]],
*,
required_policy_keys: Sequence[str] | None = None,
clipping_norm: float = 1.0,
noise_multiplier: float = 1.0,
epsilon: float = 3.0,
delta: float = 1e-06,
min_node_count: int = 3,
) -> FederatedPolicyAggregationReport
Build a deterministic review manifest for federated policy aggregation.
Parameters¶
node_updates : Sequence[Mapping[str, object]]
Federated node update records.
required_policy_keys : Sequence[str] | None
Policy keys every node update must carry, or None.
clipping_norm : float
L2 clipping norm applied to each node update.
noise_multiplier : float
Gaussian noise multiplier for differential privacy.
epsilon : float
Differential-privacy ε budget.
delta : float
Differential-privacy δ budget.
min_node_count : int
Minimum number of participating nodes required.
Returns¶
FederatedPolicyAggregationReport The federated policy aggregation review manifest.
Raises¶
ValueError If the node updates or privacy parameters are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/federated.py
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 | |
federated_dp_noise_service ¶
Offline differential-privacy noise service manifests for review-only use.
Classes¶
DpNoiseServiceReadiness
dataclass
¶
DpNoiseNodePrivacyBudget
dataclass
¶
DpNoiseServiceRequestManifest
dataclass
¶
DpNoiseServiceRequestManifest(
epsilon: float,
delta: float,
sensitivity: float,
noise_multiplier: float,
node_count: int,
seed_hash: str,
policy_keys: tuple[str, ...],
node_budgets: tuple[DpNoiseNodePrivacyBudget, ...],
schema_name: str = "federated_dp_noise_service",
schema_version: str = "1.0.0",
)
Validated request for offline DP-noise boundary review.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/federated_dp_noise_service.py
DpNoiseServiceResponseManifest
dataclass
¶
DpNoiseServiceResponseManifest(
schema_name: str,
schema_version: str,
request_hash: str,
service_readiness: DpNoiseServiceReadiness,
epsilon: float,
delta: float,
sensitivity: float,
noise_multiplier: float,
privacy_budget_spent: float,
privacy_budget_remaining: float,
node_count: int,
policy_keys: tuple[str, ...],
policy_noise_audit_vector: tuple[
tuple[str, float], ...
],
service_execution_permitted: bool,
raw_data_export_permitted: bool,
operator_review_required: bool,
non_actuating: bool,
node_budgets: tuple[DpNoiseNodePrivacyBudget, ...],
audit_record_hash: str,
)
Offline review manifest returned by the audit boundary.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/federated_dp_noise_service.py
DpNoiseServiceDeploymentPreflightManifest
dataclass
¶
DpNoiseServiceDeploymentPreflightManifest(
schema_name: str,
schema_version: str,
mechanism_label: str,
privacy_accountant_owner: str,
seed_custody_label: str,
budget_issuer_label: str,
service_endpoint_label: str,
operator_approved: bool,
request_hash: str,
response_hash: str,
epsilon: float,
delta: float,
deployment_readiness: DpNoiseServiceReadiness,
service_execution_permitted: bool,
raw_data_export_permitted: bool,
operator_review_required: bool,
non_actuating: bool,
audit_record_hash: str,
)
Deterministic review-only deployment preflight manifest.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/federated_dp_noise_service.py
Functions:¶
build_dp_noise_service_manifest ¶
build_dp_noise_service_manifest(
request: DpNoiseServiceRequestManifest,
) -> DpNoiseServiceResponseManifest
Build a deterministic, dependency-free offline DP-noise review manifest.
Parameters¶
request : DpNoiseServiceRequestManifest The DP-noise service request manifest.
Returns¶
DpNoiseServiceResponseManifest The offline DP-noise review response manifest.
Source code in src/scpn_phase_orchestrator/supervisor/federated_dp_noise_service.py
build_dp_noise_service_deployment_preflight_manifest ¶
build_dp_noise_service_deployment_preflight_manifest(
request_manifest: DpNoiseServiceRequestManifest,
response_manifest: DpNoiseServiceResponseManifest,
*,
mechanism_label: str,
privacy_accountant_owner: str,
seed_custody_label: str,
budget_issuer_label: str,
service_endpoint_label: str,
operator_approved: bool,
) -> DpNoiseServiceDeploymentPreflightManifest
Build a deterministic DP-noise deployment preflight manifest.
Parameters¶
request_manifest : DpNoiseServiceRequestManifest The DP-noise service request manifest. response_manifest : DpNoiseServiceResponseManifest The DP-noise service response manifest. mechanism_label : str Label of the privacy mechanism. privacy_accountant_owner : str Owner of the privacy accountant. seed_custody_label : str Label describing seed custody. budget_issuer_label : str Label of the privacy-budget issuer. service_endpoint_label : str Label of the service endpoint. operator_approved : bool Whether a human operator has approved the deployment.
Returns¶
DpNoiseServiceDeploymentPreflightManifest The DP-noise deployment preflight manifest.
Raises¶
ValueError If the request and response manifests are inconsistent.
Source code in src/scpn_phase_orchestrator/supervisor/federated_dp_noise_service.py
395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 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 | |
federated_secure_aggregation ¶
Offline secure aggregation manifests for federated supervisor review.
Classes¶
SecureAggregationConfig
dataclass
¶
SecureAggregationConfig(
clipping_norm: float = 1.0,
min_node_count: int = 3,
epsilon: float = 3.0,
delta: float = 1e-06,
)
Offline policy for deterministic manifest-only secure aggregation.
SecureNodeCommitment
dataclass
¶
SecureNodeCommitment(
node_id: str,
masked_policy_delta: tuple[tuple[str, float], ...],
sample_count: int,
share_commitment: str,
share_commitment_hash: str,
share_hash: str,
masked_delta_hash: str,
accepted: bool,
rejection_reasons: tuple[str, ...],
update_hash: str,
)
Validated masked node commitment used in secure aggregation review.
Methods:¶
to_audit_record ¶
Return a JSON-safe audit record for this node commitment.
Returns¶
dict[str, object] Return a JSON-safe audit record for this node commitment.
Source code in src/scpn_phase_orchestrator/supervisor/federated_secure_aggregation.py
SecureAggregationQuorumEvidence
dataclass
¶
SecureNodeCustodyRecord
dataclass
¶
SecureNodeCustodyRecord(
node_id: str,
key_custody_label: str,
share_custody_label: str,
previous_key_custody_label: str,
previous_share_custody_label: str,
key_custody_continuity_hash: str,
share_custody_continuity_hash: str,
)
Custody metadata for key and share labels used in review.
Methods:¶
to_audit_record ¶
Return a JSON-safe node-custody audit record.
Returns¶
dict[str, object] Return a JSON-safe node-custody audit record.
Source code in src/scpn_phase_orchestrator/supervisor/federated_secure_aggregation.py
FederatedSecureAggregationManifest
dataclass
¶
FederatedSecureAggregationManifest(
schema_name: str,
schema_version: str,
config: SecureAggregationConfig,
required_policy_keys: tuple[str, ...],
node_commitments: tuple[SecureNodeCommitment, ...],
accepted_node_count: int,
rejected_node_count: int,
total_sample_count: int,
aggregate_masked_delta: tuple[tuple[str, float], ...],
aggregate_masked_delta_hash: str,
secure_aggregation_execution_permitted: bool,
raw_data_export_permitted: bool,
operator_review_required: bool,
non_actuating: bool,
quorum_met: bool,
claim_boundary: str,
report_hash: str,
)
Manifest for deterministic offline secure aggregation review only.
Methods:¶
to_audit_record ¶
Return a JSON-safe aggregate audit record.
Returns¶
dict[str, object] Return a JSON-safe aggregate audit record.
Source code in src/scpn_phase_orchestrator/supervisor/federated_secure_aggregation.py
FederatedSecureAggregationPreflightManifest
dataclass
¶
FederatedSecureAggregationPreflightManifest(
schema_name: str,
schema_version: str,
secure_aggregation_schema_name: str,
secure_aggregation_schema_version: str,
secure_aggregation_report_hash: str,
accepted_node_threshold: int,
accepted_node_count: int,
quorum_evidence: tuple[
SecureAggregationQuorumEvidence, ...
],
custody_rotation_policy: str,
custody_records: tuple[SecureNodeCustodyRecord, ...],
operator_approved: bool,
operator_id: str,
service_owner: str,
secure_aggregation_execution_permitted: bool,
raw_data_export_permitted: bool,
operator_review_required: bool,
non_actuating: bool,
report_hash: str,
)
Review-only deployment preflight envelope for manifest execution.
Methods:¶
to_audit_record ¶
Return a JSON-safe preflight audit record.
Returns¶
dict[str, object] Return a JSON-safe preflight audit record.
Source code in src/scpn_phase_orchestrator/supervisor/federated_secure_aggregation.py
Functions:¶
build_federated_secure_aggregation_manifest ¶
build_federated_secure_aggregation_manifest(
node_commitments: Sequence[Mapping[str, object]],
*,
required_policy_keys: Sequence[str] | None = None,
clipping_norm: float = 1.0,
min_node_count: int = 3,
epsilon: float = 3.0,
delta: float = 1e-06,
) -> FederatedSecureAggregationManifest
Build a deterministic secure aggregation manifest.
Parameters¶
node_commitments : Sequence[Mapping[str, object]]
Secure-aggregation node commitment records.
required_policy_keys : Sequence[str] | None
Policy keys every node update must carry, or None.
clipping_norm : float
L2 clipping norm applied to each node update.
min_node_count : int
Minimum number of participating nodes required.
epsilon : float
Differential-privacy ε budget.
delta : float
Differential-privacy δ budget.
Returns¶
FederatedSecureAggregationManifest The secure-aggregation manifest.
Raises¶
ValueError If the node commitments or privacy parameters are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/federated_secure_aggregation.py
256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 | |
build_federated_secure_aggregation_preflight_manifest ¶
build_federated_secure_aggregation_preflight_manifest(
secure_aggregation_manifest: FederatedSecureAggregationManifest,
*,
quorum_evidence: Sequence[Mapping[str, object]],
custody_rotation_policy: str,
custody_records: Sequence[Mapping[str, object]],
accepted_node_threshold: int,
operator_approved: bool,
operator_id: str,
service_owner: str,
) -> FederatedSecureAggregationPreflightManifest
Build a deterministic review-only deployment preflight manifest.
Parameters¶
secure_aggregation_manifest : FederatedSecureAggregationManifest The secure-aggregation manifest to preflight. quorum_evidence : Sequence[Mapping[str, object]] Per-node quorum evidence records. custody_rotation_policy : str Key-custody rotation policy label. custody_records : Sequence[Mapping[str, object]] Node custody records. accepted_node_threshold : int Minimum number of accepted nodes required. operator_approved : bool Whether a human operator has approved the deployment. operator_id : str Identifier of the approving operator. service_owner : str Owner of the aggregation service.
Returns¶
FederatedSecureAggregationPreflightManifest The secure-aggregation deployment preflight manifest.
Raises¶
TypeError If an argument has the wrong type. ValueError If the manifest or quorum evidence is invalid.
Source code in src/scpn_phase_orchestrator/supervisor/federated_secure_aggregation.py
376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 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 | |
federated_transport ¶
Federated transport envelope and replay validation helpers.
Classes¶
FederatedTransportEnvelope
dataclass
¶
FederatedTransportEnvelope(
schema_name: str,
schema_version: str,
batch_id: str,
sequence_position: int,
node_id: str,
node_sequence: int,
envelope_id: str,
parent_envelope_hash: str,
node_update_audit_record: tuple[
tuple[str, object], ...
],
node_update_audit_hash: str,
envelope_signature: str,
envelope_hash: str,
transport_execution_permitted: bool,
raw_data_export_permitted: bool,
operator_review_required: bool,
)
Signed/hash-linked transport envelope around one node audit update.
Methods:¶
to_audit_record ¶
Return JSON-safe audit evidence for this transport envelope.
Returns¶
dict[str, object] Return JSON-safe audit evidence for this transport envelope.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
FederatedTransportReplayLedger
dataclass
¶
FederatedTransportReplayLedger(
schema_name: str,
schema_version: str,
batch_id: str,
envelope_count: int,
envelope_ids: tuple[str, ...],
node_last_sequences: tuple[tuple[str, int], ...],
replay_hash: str,
)
Replay result for an ordered transport batch.
Methods:¶
to_audit_record ¶
Return JSON-safe replay evidence.
Returns¶
dict[str, object] Return JSON-safe replay evidence.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
FederatedTransportDeploymentPreflightManifest
dataclass
¶
FederatedTransportDeploymentPreflightManifest(
schema_name: str,
schema_version: str,
batch_id: str,
preflight_id: str,
transport: str,
transport_endpoint: str,
transport_audit_record: tuple[tuple[str, object], ...],
transport_audit_hash: str,
replay_ledger_hash: str,
transport_execution_permitted: bool,
raw_data_export_permitted: bool,
operator_review_required: bool,
non_actuating: bool,
preflight_signature: str,
preflight_hash: str,
)
Deterministic, review-only preflight manifest for transport deployment.
Methods:¶
to_audit_record ¶
Return JSON-safe preflight audit evidence.
Returns¶
dict[str, object] Return JSON-safe preflight audit evidence.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
Functions:¶
build_signed_transport_envelopes ¶
build_signed_transport_envelopes(
node_update_audit_records: Sequence[
Mapping[str, object]
],
*,
schema_name: str = _DEFAULT_SCHEMA_NAME,
schema_version: str = _DEFAULT_SCHEMA_VERSION,
batch_id: str | None = None,
) -> tuple[FederatedTransportEnvelope, ...]
Build deterministic hash-linked envelopes from node audit records.
Parameters¶
node_update_audit_records : Sequence[Mapping[str, object]]
Federated node update audit records.
schema_name : str
Transport schema name.
schema_version : str
Transport schema version.
batch_id : str | None
Identifier of the transport batch, or None.
Returns¶
tuple[FederatedTransportEnvelope, ...] The hash-linked signed transport envelopes.
Raises¶
ValueError If the node audit records are malformed.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
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 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 | |
validate_federated_transport_batch ¶
validate_federated_transport_batch(
envelopes: Sequence[FederatedTransportEnvelope],
) -> tuple[FederatedTransportEnvelope, ...]
Validate deterministic hash-links and ordering for an ordered transport batch.
Parameters¶
envelopes : Sequence[FederatedTransportEnvelope] The ordered transport envelopes.
Returns¶
tuple[FederatedTransportEnvelope, ...] The validated transport batch.
Raises¶
ValueError If the hash-links or ordering are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 | |
replay_federated_transport_batch ¶
replay_federated_transport_batch(
envelopes: Sequence[FederatedTransportEnvelope],
) -> FederatedTransportReplayLedger
Replay and materialise a deterministic digest for an ordered transport batch.
Parameters¶
envelopes : Sequence[FederatedTransportEnvelope] The ordered transport envelopes.
Returns¶
FederatedTransportReplayLedger The replay ledger for the transport batch.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
build_transport_deployment_preflight_manifest ¶
build_transport_deployment_preflight_manifest(
transport_declaration: Mapping[str, object],
*,
replay_ledger: FederatedTransportReplayLedger,
schema_name: str = _DEFAULT_SCHEMA_NAME,
schema_version: str = _DEFAULT_SCHEMA_VERSION,
batch_id: str | None = None,
) -> FederatedTransportDeploymentPreflightManifest
Build deterministic transport preflight evidence.
Parameters¶
transport_declaration : Mapping[str, object]
The transport declaration to preflight.
replay_ledger : FederatedTransportReplayLedger
The replay ledger for the transport batch.
schema_name : str
Transport schema name.
schema_version : str
Transport schema version.
batch_id : str | None
Identifier of the transport batch, or None.
Returns¶
FederatedTransportDeploymentPreflightManifest The transport deployment preflight manifest.
Raises¶
ValueError If the transport declaration or ledger is invalid.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
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 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 | |
validate_transport_deployment_preflight_manifest ¶
validate_transport_deployment_preflight_manifest(
manifest: FederatedTransportDeploymentPreflightManifest,
) -> FederatedTransportDeploymentPreflightManifest
Validate deterministic transport preflight manifest content and hashes.
Parameters¶
manifest : FederatedTransportDeploymentPreflightManifest The manifest to validate.
Returns¶
FederatedTransportDeploymentPreflightManifest The validated transport preflight manifest.
Raises¶
ValueError If the manifest content or hashes are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/federated_transport.py
653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 | |
Information Geometry and Lineage¶
Information-geometric control proposals, static scenario examples, and
autopoietic lineage inheritance helpers for review-only policy evolution.
The information-geometry primitive keeps NumPy as the default audit-stable
backend and exposes explicit backend="jax" acceleration with reference-gated
parity for Fisher-Rao distance, Wasserstein distance, curvature proxy, and
natural-gradient proposals. Both paths remain non-actuating review surfaces.
The lineage sandbox generates deterministic child-policy candidates from a
parent policy and replay corpus, records accepted/rejected evidence, hashes the
lineage and replay corpus, and keeps live merge, hot patching, execution, and
actuation disabled. The curated replay corpus spans power-grid recovery,
cardiac-rhythm pacing recovery, traffic-flow platooning, and cyber-industrial
recontainment so operators can compare policy diffs across domains before any
separate inheritance-review workflow.
Intergenerational inheritance then signs accepted child-policy records,
materialises inherited genomes, records multi-objective replay fitness, and can
package deterministic history rows for operator review. The history package
links lineage hashes, inheritance hashes, HMAC signature metadata, replay
domains, and fitness ranges while keeping direct hot patching and actuation
disabled.
information_geometry ¶
Deterministic information-geometry control proposals for geometry-aware control.
Classes¶
InformationGeometryState
dataclass
¶
InformationGeometryState(
simplex_coordinates: FloatArray,
target_coordinates: FloatArray,
metric_tensor: FloatArray,
tangent_vector: FloatArray,
curvature_proxy: float,
geodesic_length: float,
)
Internal geometry state on simplex coordinates.
Attributes are designed to be deterministic and JSON-safe after conversion.
InformationGeometryControlProposal
dataclass
¶
InformationGeometryControlProposal(
action_proposals: tuple[ControlAction, ...],
fisher_rao_distance: float,
wasserstein_distance: float,
natural_gradient_norm: float,
curvature_proxy: float,
backend: str,
claim_boundary: str,
non_actuating: bool,
execution_disabled: bool,
proposal_hash: str,
state: InformationGeometryState,
)
Review-only control proposal derived from information-geometry metrics.
Methods:¶
to_audit_record ¶
Return a JSON-safe audit payload for the proposal.
Returns¶
dict[str, object] Return a JSON-safe audit payload for the proposal.
Source code in src/scpn_phase_orchestrator/supervisor/information_geometry.py
Functions:¶
propose_information_geometry_control ¶
propose_information_geometry_control(
current_distribution: FloatArray
| list[float]
| tuple[float, ...],
target_distribution: FloatArray
| list[float]
| tuple[float, ...],
coupling_gradient: FloatArray
| list[float]
| tuple[float, ...]
| None = None,
*,
max_step: float,
knob: str = _DEFAULT_KNOB,
scope: str = _DEFAULT_SCOPE,
backend: str = "numpy",
) -> InformationGeometryControlProposal
Compute a finite, deterministic information-geometry control proposal.
Parameters are validated eagerly and no mutation of caller arrays is performed.
The default NumPy backend preserves historical audit hashes; passing
backend="jax" uses a JAX-native vectorised metric path and converts the
resulting proposal back to JSON-safe NumPy scalars and arrays.
Parameters¶
current_distribution : FloatArray | list[float] | tuple[float, ...]
The current probability distribution.
target_distribution : FloatArray | list[float] | tuple[float, ...]
The target probability distribution.
coupling_gradient : FloatArray | list[float] | tuple[float, ...] | None
Gradient of coherence with respect to the coupling, or None.
max_step : float
Maximum control step magnitude.
knob : str
Name of the control knob to adjust.
scope : str
Scope label for the proposed control.
backend : str
Name of the compute backend to use.
Returns¶
InformationGeometryControlProposal The deterministic information-geometry control proposal.
Raises¶
ValueError If the distributions or step are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/information_geometry.py
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 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 | |
information_geometry_examples ¶
Information geometry control scenarios for non-actuating review.
Classes¶
DistributionPair
dataclass
¶
Discrete distribution pair describing review-only geometry transfer targets.
Attributes¶
current_summary
property
¶
Return summary statistics for the current distribution.
Returns¶
dict[str, float] Return summary statistics for the current distribution.
target_summary
property
¶
Return summary statistics for the target distribution.
Returns¶
dict[str, float] Return summary statistics for the target distribution.
Methods:¶
to_record ¶
Return a deterministic JSON-safe record.
Returns¶
dict[str, list[float]] Return a deterministic JSON-safe record.
Source code in src/scpn_phase_orchestrator/supervisor/information_geometry_examples.py
InformationGeometryScenario
dataclass
¶
InformationGeometryScenario(
domain: str,
scenario_id: str,
distributions: DistributionPair,
objective_labels: tuple[str, ...],
control_gradient: tuple[tuple[str, float], ...],
max_step: float,
knob_hints: tuple[str, ...] = (),
non_actuating: bool = True,
execution_disabled: bool = True,
claim_boundary: str = InformationGeometryBoundary,
)
Deterministic control scenario for information-geometry review fixtures.
Methods:¶
scenario_hash ¶
Return the deterministic scenario digest.
Returns¶
str Return the deterministic scenario digest.
Source code in src/scpn_phase_orchestrator/supervisor/information_geometry_examples.py
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/information_geometry_examples.py
Functions:¶
build_information_geometry_control_scenarios ¶
Build deterministic information-geometry control scenarios.
Returns¶
tuple[dict[str, object], ...] Build deterministic information-geometry control scenarios.
Source code in src/scpn_phase_orchestrator/supervisor/information_geometry_examples.py
lineage ¶
Review-only child-policy lineage manifests for replay sandboxes.
Functions:¶
build_autopoietic_lineage_replay_corpus ¶
Return a deterministic multi-domain replay corpus for lineage review.
The corpus is intentionally offline and compact. It gives the lineage sandbox domain-diverse replay evidence without loading partner data, contacting services, or enabling any live merge path.
Returns¶
tuple[dict[str, object], ...] Return a deterministic multi-domain replay corpus for lineage review.
Source code in src/scpn_phase_orchestrator/supervisor/lineage.py
build_autopoietic_lineage_sandbox ¶
build_autopoietic_lineage_sandbox(
parent_policy: Mapping[str, object],
audit_replays: Sequence[Mapping[str, object]],
*,
child_budget: int = 3,
mutation_step: float = 0.02,
minimum_replay_reward: float = 0.0,
minimum_safety_margin: float = 0.0,
) -> dict[str, object]
Build a deterministic offline child-policy lineage review manifest.
The sandbox mutates a numeric parent-policy mapping into a bounded set of child candidates, evaluates each candidate only against supplied replay summaries, and emits reviewable policy diffs. It never permits live merge, hot patching, or actuation.
Parameters¶
parent_policy : Mapping[str, object] The parent policy genome. audit_replays : Sequence[Mapping[str, object]] Audit replay records used to score child candidates. child_budget : int Maximum number of child candidates to evaluate. mutation_step : float Mutation step size applied to the parent genome. minimum_replay_reward : float Minimum replay reward a child must reach to be accepted. minimum_safety_margin : float Minimum safety margin a child must preserve.
Returns¶
dict[str, object] The offline child-policy lineage review manifest.
Raises¶
ValueError If the parent policy or replay inputs are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/lineage.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 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 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 | |
build_intergenerational_policy_inheritance ¶
build_intergenerational_policy_inheritance(
lineage_manifest: Mapping[str, object],
child_candidate: Mapping[str, object],
*,
signer_id: str,
signing_key: str,
objective_weights: Mapping[str, object] | None = None,
) -> dict[str, object]
Build signed review metadata for inherited child-policy genomes.
The resulting manifest materialises the inherited policy genome from a reviewed child diff, records replay-fitness components, and signs metadata for operator review. It does not permit direct hot patches or actuation.
Parameters¶
lineage_manifest : Mapping[str, object]
The lineage review manifest.
child_candidate : Mapping[str, object]
The candidate child policy genome.
signer_id : str
Identifier of the signing authority.
signing_key : str
HMAC signing key for the record.
objective_weights : Mapping[str, object] | None
Per-objective weights, or None for defaults.
Returns¶
dict[str, object] The signed inherited child-policy review metadata.
Source code in src/scpn_phase_orchestrator/supervisor/lineage.py
build_intergenerational_policy_inheritance_history ¶
build_intergenerational_policy_inheritance_history(
lineage_manifest: Mapping[str, object],
inheritance_manifests: Sequence[Mapping[str, object]],
) -> dict[str, object]
Build deterministic review history for signed inherited-policy records.
The history package joins one lineage sandbox manifest with signed inheritance manifests derived from it. It is evidence for operator review: the package is deterministic, validates disabled direct hot patching and actuation, and does not execute or merge inherited policies.
Parameters¶
lineage_manifest : Mapping[str, object] The lineage review manifest. inheritance_manifests : Sequence[Mapping[str, object]] Signed inheritance manifests to fold into the history.
Returns¶
dict[str, object] The review history of signed inherited-policy records.
Raises¶
ValueError If an inheritance manifest is inconsistent.
Source code in src/scpn_phase_orchestrator/supervisor/lineage.py
Multiverse and Topos Review¶
Counterfactual branch simulation, example manifests, branch-risk gates, and
categorical policy composition checks. The multiverse simulator keeps NumPy as
the default deterministic audit backend and exposes explicit backend="jax"
acceleration for larger branch corpora where JAX can place the vectorized
rollout on an available accelerator. Reference benchmarks gate JAX output
against NumPy branch hashes, topology metrics, order-parameter trajectories, and
final phase angles while preserving the non-actuating, execution-disabled review
boundary. Multiverse branch rollouts preserve the Kuramoto graph contract by
requiring zero diagonal baseline coupling, phase-lag, and topology-mask matrices;
matrix branch actions are projected back onto the off-diagonal graph before
simulation. Domain scenario fixtures now cover power-grid, cardiac-rhythm,
cyber-industrial, traffic-flow, manufacturing process-control, and plasma-control
use cases with simulator-compatible K, alpha, zeta, and Psi candidate
controls. Studio packages rollout manifests and branch-risk reports through
the public
scpn_phase_orchestrator.studio.build_multiverse_counterfactual_studio_panel()
facade, which preserves the non-actuating claim boundaries, joins branch hashes,
renders approval/rejection evidence, and never emits executable actions.
multiverse ¶
Deterministic counterfactual branch rollouts over branch topologies.
The implementation runs vectorised NumPy or optional JAX trajectories for multiple branch interventions in one pass and keeps a strict non-actuation boundary. It is an upstream-safe simulation surface for research, policy gating, and audit review.
Classes¶
MultiverseBranchSpec
dataclass
¶
MultiverseBranchSpec(
branch_id: str,
actions: tuple[ControlAction, ...],
topology_mask: FloatArray | None = None,
)
Declarative counterfactual branch intervention specification.
Methods:¶
to_audit_record ¶
Return a JSON-safe branch specification record.
Returns¶
dict[str, object] Return a JSON-safe branch specification record.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse.py
MultiverseBranchRecord
dataclass
¶
MultiverseBranchRecord(
branch_id: str,
branch_hash: str,
action_count: int,
action_labels: tuple[str, ...],
topology_edge_count: int,
topology_scale: float,
final_R: float,
mean_R: float,
min_R: float,
max_R: float,
final_psi: float,
)
Audit record for one counterfactual branch rollout.
Methods:¶
to_audit_record ¶
Return a JSON-safe branch rollout record.
Returns¶
dict[str, object] Return a JSON-safe branch rollout record.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse.py
MultiverseCounterfactualManifest
dataclass
¶
MultiverseCounterfactualManifest(
schema_name: str,
schema_version: str,
branch_records: tuple[MultiverseBranchRecord, ...],
branch_count: int,
horizon: int,
backend: str,
non_actuating: bool,
execution_disabled: bool,
claim_boundary: str,
manifest_hash: str,
)
Audit manifest for a full multiverse counterfactual rollout.
Methods:¶
to_audit_record ¶
Return a JSON-safe multiverse rollout manifest.
Returns¶
dict[str, object] Return a JSON-safe multiverse rollout manifest.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse.py
Functions:¶
simulate_multiverse_counterfactual_branches ¶
simulate_multiverse_counterfactual_branches(
phases: NDArray[float64],
omegas: NDArray[float64],
baseline_k: NDArray[float64],
baseline_alpha: NDArray[float64],
branch_specs: tuple[MultiverseBranchSpec, ...] = (),
*,
branch_action_sets: tuple[Sequence[ControlAction], ...]
| None = None,
topology_masks: tuple[FloatArray, ...] | None = None,
baseline_zeta: float = 0.0,
baseline_psi: float = 0.0,
horizon: int = 20,
dt: float = 0.01,
method: str = "rk4",
backend: str = "numpy",
) -> MultiverseCounterfactualManifest
Run deterministic branch counterfactual rollouts without actuation.
Parameters¶
phases : NDArray[np.float64]
Oscillator phases in radians, shape (N,).
omegas : NDArray[np.float64]
Natural frequencies in rad/s, shape (N,).
baseline_k : NDArray[np.float64]
Baseline coupling matrix K_nm, shape (N, N).
baseline_alpha : NDArray[np.float64]
Baseline phase-lag matrix, shape (N, N).
branch_specs : tuple[MultiverseBranchSpec, ...]
Specifications of the counterfactual branches.
branch_action_sets : tuple[Sequence[ControlAction], ...] | None
Per-branch control-action sequences, or None.
topology_masks : tuple[FloatArray, ...] | None
Per-branch topology masks, or None.
baseline_zeta : float
Baseline external drive strength ζ.
baseline_psi : float
Baseline external drive reference phase Ψ in radians.
horizon : int
Rollout horizon in steps.
dt : float
Integration step size.
method : str
Integration method (euler, rk4, or rk45).
backend : str
Name of the compute backend to use.
Returns¶
MultiverseCounterfactualManifest The multiverse counterfactual rollout manifest.
Raises¶
ValueError If the branch specs or rollout inputs are invalid.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse.py
638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 | |
multiverse_examples ¶
Multiverse counterfactual scenario examples for branch review.
Classes¶
BranchCandidate
dataclass
¶
BranchCandidate(
candidate_id: str,
knob_variations: tuple[tuple[str, float], ...],
topology_variations: tuple[str, ...],
objective_labels: tuple[str, ...],
non_actuating: bool = True,
execution_disabled: bool = True,
claim_boundary: str = CounterfactualBoundary,
)
Single deterministic branch candidate for a counterfactual rollout scenario.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_examples.py
DomainScenario
dataclass
¶
DomainScenario(
domain: str,
scenario_id: str,
initial_phases: NDArray[float64],
initial_omegas: NDArray[float64],
branch_candidates: tuple[BranchCandidate, ...],
objective_labels: tuple[str, ...],
non_actuating: bool = True,
execution_disabled: bool = True,
claim_boundary: str = CounterfactualBoundary,
)
Deterministic scenario definition for aggregate multiverse benchmarks.
Methods:¶
scenario_hash ¶
Return the deterministic scenario digest.
Returns¶
str Return the deterministic scenario digest.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_examples.py
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_examples.py
Functions:¶
build_multiverse_domain_scenarios ¶
Build deterministic multiverse domain scenario records.
Returns¶
tuple[dict[str, object], ...] Build deterministic multiverse domain scenario records.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_examples.py
multiverse_risk ¶
Fail-closed review gate over precomputed branch rollout manifests.
Classes¶
MultiverseRiskThresholds
dataclass
¶
MultiverseRiskThresholds(
min_mean_R: float = 0.0,
min_final_R: float = 0.0,
max_action_count: int = 64,
max_topology_edge_count: int | None = None,
max_topology_scale: float | None = None,
)
Guard thresholds for branch review in the multiverse gate.
BranchRiskDecision
dataclass
¶
BranchRiskDecision(
branch_id: str,
branch_hash: str,
final_R: float,
mean_R: float,
min_R: float,
max_R: float,
action_count: int,
topology_edge_count: int | None,
topology_scale: float | None,
approved: bool,
rejection_reasons: tuple[str, ...],
)
Outcome for one branch in a manifest.
Methods:¶
to_audit_record ¶
Return a JSON-safe branch decision record.
Returns¶
dict[str, object] Return a JSON-safe branch decision record.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_risk.py
MultiverseRiskReport
dataclass
¶
MultiverseRiskReport(
schema_name: str,
schema_version: str,
branch_decisions: tuple[BranchRiskDecision, ...],
approved_count: int,
rejected_count: int,
safest_branch_id: str | None,
safest_branch_hash: str | None,
rejection_reasons: tuple[str, ...],
claim_boundary: str,
non_actuating: bool,
execution_disabled: bool,
report_hash: str,
)
JSON-safe aggregate of the branch-risk review decision.
Methods:¶
to_audit_record ¶
Return a JSON-safe multiverse risk gate audit record.
Returns¶
dict[str, object] Return a JSON-safe multiverse risk gate audit record.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_risk.py
Functions:¶
evaluate_multiverse_branch_risk ¶
evaluate_multiverse_branch_risk(
manifest: Mapping[str, object],
thresholds: MultiverseRiskThresholds | None = None,
) -> MultiverseRiskReport
Evaluate branch risk decisions for a branch manifest without actuation.
Parameters¶
manifest : Mapping[str, object]
The branch rollout manifest to evaluate.
thresholds : MultiverseRiskThresholds | None
Risk thresholds, or None for defaults.
Returns¶
MultiverseRiskReport The multiverse risk-gate report.
Source code in src/scpn_phase_orchestrator/supervisor/multiverse_risk.py
topos_policy ¶
Deterministic audit/proof-obligation validation for policy composition.
Classes¶
PolicyCompositionObject
dataclass
¶
PolicyCompositionMorphism
dataclass
¶
PolicyCompositionObligation
dataclass
¶
PolicyCompositionValidationReport
dataclass
¶
PolicyCompositionValidationReport(
schema_name: str,
schema_version: str,
object_count: int,
morphism_count: int,
obligation_records: tuple[
PolicyCompositionObligation, ...
],
objects: tuple[PolicyCompositionObject, ...],
morphisms: tuple[PolicyCompositionMorphism, ...],
passed: bool,
report_hash: str,
proof_boundary: str,
non_actuating: bool = True,
)
JSON-safe deterministic validation report for policy composition.
Methods:¶
to_audit_record ¶
Return a deterministic JSON-safe audit record.
Returns¶
dict[str, object] Return a deterministic JSON-safe audit record.
Source code in src/scpn_phase_orchestrator/supervisor/topos_policy.py
Functions:¶
validate_policy_composition_category ¶
validate_policy_composition_category(
rules: tuple[PolicyRule, ...] | list[PolicyRule],
) -> PolicyCompositionValidationReport
Validate PolicyRule collections as a categorical composition proof boundary.
Parameters¶
rules : tuple[PolicyRule, ...] | list[PolicyRule] The policy rules to export or validate.
Returns¶
PolicyCompositionValidationReport The categorical composition validation report.
Raises¶
ValueError If the rules violate the composition proof boundary.
Source code in src/scpn_phase_orchestrator/supervisor/topos_policy.py
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 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 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 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 | |