Core & Exceptions

Foundation types shared across all SPO subsystems.

Purpose for production teams

This page is the reliability seam for the entire runtime. The shared error family and compatibility controls are what allow subsystems to fail with bounded, actionable behaviour instead of cascading into undefined control actions.

When reading this page:

• use the exception section to align operational alerting and fallback policy,
• use the compatibility section to understand optional dependency fallbacks,
• use the observability entry points to verify what metrics and telemetry remain available when optional stacks are missing.

The intent is to make runtime limits and optional paths explicit before deployment planning begins.

Exception Hierarchy

SPO defines a hierarchy of domain-specific exceptions rooted at SPOError. Each subsystem raises its own subclass so that callers can catch errors at the appropriate granularity:

SPOError
├── BindingError          # Invalid binding specification or missing fields
├── ValidationError       # Schema or constraint violation
├── ExtractorError        # Phase extraction failure (bad signal, wrong sample rate)
├── EngineError           # UPDE integration failure (NaN, divergence, dt violation)
├── PolicyError           # Policy rule evaluation failure
└── AuditError            # Audit chain integrity violation

All exceptions carry a descriptive message and preserve the original traceback when wrapping lower-level errors.

Catching by Granularity

from scpn_phase_orchestrator.exceptions import SPOError, EngineError

try:
    engine.step(phases, omegas, knm, zeta, psi, alpha)
except EngineError as e:
    # Handle integration failure specifically
    logger.warning(f"Engine diverged: {e}")
    supervisor.force_transition(Regime.CRITICAL)
except SPOError as e:
    # Catch-all for any SPO error
    logger.error(f"SPO error: {e}")

exceptions

Exception hierarchy for scpn-phase-orchestrator.

Classes

SPOError

Bases: Exception

Base exception for all scpn-phase-orchestrator errors.

BindingError

Bases: SPOError, ValueError

Failed to load or parse a binding spec.

ValidationError

Bases: SPOError, ValueError

Binding spec failed validation.

ExtractorError

Bases: SPOError, RuntimeError

Phase extractor encountered an unrecoverable signal condition.

EngineError

Bases: SPOError, RuntimeError

UPDE or Stuart-Landau integrator diverged or hit a numerical fault.

PolicyError

Bases: SPOError, ValueError

Malformed policy rule or unresolvable action.

AuditError

Bases: SPOError, RuntimeError

Audit log integrity violation (hash chain mismatch, missing header).

Compatibility

Internal compatibility module providing shared constants and conditional imports for optional dependencies:

• TWO_PI — \(2\pi\) as float64 (avoids repeated computation)
• Conditional imports for jax, equinox, redis, opentelemetry that fall back gracefully when the dependency is not installed

_compat

Shared compatibility constants for Python and optional Rust execution paths.

The module exposes TWO_PI and a process-local HAS_RUST capability flag computed from the presence of the optional spo_kernel package. Importing it does not load the Rust extension or change backend selection by itself; concrete modules decide when to dispatch to accelerated kernels.

CLI entry point

The public Click command tree lives in scpn_phase_orchestrator.runtime.cli. Use the CLI Reference for command-oriented examples; this API section keeps the callable entry points visible to mkdocstrings.

cli

Command-line entry point for validation, replay, export, and review workflows.

The CLI wraps public SPO APIs behind explicit commands for binding validation, inspection, auto-binding proposals, coupling estimation, formal export, replay, plugin catalogs, scaffolding, and selected runtime utilities. Commands validate local inputs and emit text or JSON review artifacts; they do not push commits, start network services, or perform live actuation unless an explicit subcommand is invoked for that runtime path.

Functions:

main

main() -> None

SCPN Phase Orchestrator CLI.

Source code in src/scpn_phase_orchestrator/runtime/cli/_app.py

@click.group()
@click.version_option(
    package_name="scpn-phase-orchestrator", prog_name="scpn-phase-orchestrator"
)
def main() -> None:
    """SCPN Phase Orchestrator CLI."""

meta_transfer_manifest

meta_transfer_manifest(
    audit_paths: tuple[str, ...],
    audit_directory: str | None,
    pattern: str,
    min_records: int,
    package_name: str,
    import_target: str,
    console_script: str,
    output: str | None,
) -> None

Emit a review-only meta-transfer package manifest from audit history.

Parameters

audit_paths : tuple[str, ...] Audit-log paths to package. audit_directory : str | None Directory of audit logs, or None. pattern : str Glob pattern for audit-log discovery. min_records : int Minimum number of records required. package_name : str Name for the emitted package. import_target : str Import target for the generated package. console_script : str Console-script entry-point name. output : str | None Destination path, or None for stdout.

Raises

ClickException If the inputs are invalid or the operation fails.

Source code in src/scpn_phase_orchestrator/runtime/cli/meta.py

@main.command("meta-transfer-manifest")
@click.argument(
    "audit_paths",
    nargs=-1,
    type=click.Path(exists=True, dir_okay=False),
)
@click.option(
    "--audit-directory",
    default=None,
    type=click.Path(exists=True, file_okay=False),
    help="Nested audit-history directory to discover with --pattern.",
)
@click.option(
    "--pattern",
    default="**/*.jsonl",
    show_default=True,
    help="Glob pattern used with --audit-directory.",
)
@click.option("--min-records", default=1, show_default=True, type=int)
@click.option("--package-name", default="scpn-meta", show_default=True)
@click.option(
    "--import-target",
    default="scpn_phase_orchestrator.meta",
    show_default=True,
)
@click.option("--console-script", default="scpn-meta", show_default=True)
@click.option(
    "--output",
    "-o",
    default=None,
    type=click.Path(),
    help="Write manifest JSON to a file instead of stdout.",
)
def meta_transfer_manifest(
    audit_paths: tuple[str, ...],
    audit_directory: str | None,
    pattern: str,
    min_records: int,
    package_name: str,
    import_target: str,
    console_script: str,
    output: str | None,
) -> None:
    """Emit a review-only meta-transfer package manifest from audit history.

    Parameters
    ----------
    audit_paths : tuple[str, ...]
        Audit-log paths to package.
    audit_directory : str | None
        Directory of audit logs, or ``None``.
    pattern : str
        Glob pattern for audit-log discovery.
    min_records : int
        Minimum number of records required.
    package_name : str
        Name for the emitted package.
    import_target : str
        Import target for the generated package.
    console_script : str
        Console-script entry-point name.
    output : str | None
        Destination path, or ``None`` for stdout.

    Raises
    ------
    ClickException
        If the inputs are invalid or the operation fails.
    """
    if min_records < 1:
        raise click.ClickException("--min-records must be at least 1")
    if audit_directory is None and not audit_paths:
        raise click.ClickException(
            "provide one or more audit JSONL files or --audit-directory"
        )
    if audit_directory is not None and audit_paths:
        raise click.ClickException(
            "audit JSONL files and --audit-directory are mutually exclusive"
        )
    try:
        if audit_directory is not None:
            model = CrossDomainMetaTransfer.fit_audit_directory(
                audit_directory,
                pattern=pattern,
                min_records=min_records,
            )
        else:
            model = CrossDomainMetaTransfer.fit_audit_history(
                audit_paths,
                min_records=min_records,
            )
        manifest = model.to_package_manifest(
            package_name=package_name,
            import_target=import_target,
            console_script=console_script,
        )
    except (
        OSError,
        TypeError,
        ValueError,
        json.JSONDecodeError,
        UnicodeDecodeError,
    ) as exc:
        raise click.ClickException(str(exc)) from exc

    text = json.dumps(manifest.to_audit_record(), indent=2, sort_keys=True) + "\n"
    if output is None:
        click.echo(text, nl=False)
        return
    Path(output).write_text(text, encoding="utf-8")
    click.echo(f"Meta-transfer package manifest written: {output}")

simulate

simulate(
    spec: BindingSpec,
    *,
    steps: int = 100,
    seed: int = 42,
    policy_enabled: bool = True,
    control_mode: SimulationControlMode = "supervisor_policy",
    audit_logger: AuditLogger | None = None,
    strict_audit_integrity: bool = True,
    binding_spec_path: Path | None = None,
    scenario_hook: ScenarioCallback | None = None,
    conformal_gate: TwinConformalGate | None = None,
    twin_confidence_source: TwinConfidenceSource
    | None = None,
) -> SimulationResult

Advance a binding spec for steps and return the simulation outcome.

Parameters

spec : BindingSpec A validated binding spec. steps : int Number of integration steps. seed : int RNG seed for the initial phases. policy_enabled : bool Closed-loop control feedback on (True) or open-loop baseline (False). control_mode : {"supervisor_policy"} The live control surface used by the generic binding-spec simulator. Koopman MPC is intentionally not a selectable simulate mode; it remains a review-only/offline proposal surface in :mod:scpn_phase_orchestrator.runtime.dvoc_oscillation_damping. audit_logger : AuditLogger | None Optional logger; when given, the header, per-step records, and events are written. The caller owns its lifecycle (close). strict_audit_integrity : bool Fail closed on a broken audit chain. When True (default) and the audit logger owns a protobuf event stream whose close-time integrity check fails, the run raises :class:~scpn_phase_orchestrator.exceptions.AuditError instead of returning a result with a tamper-evident but unenforced audit_event_stream_integrity field. Set False to attach the failed result for inspection without failing the run. binding_spec_path : Path | None Optional path to the spec, used only to locate an adjacent policy.yaml. When None, no domainpack policy rules are loaded (the supervisor policy still runs when policy_enabled). scenario_hook : ScenarioCallback | None Optional deterministic per-step perturbation hook for benchmark and case-study scenarios. The hook can mutate phases, frequencies, coupling, zeta, or psi_target in a validated :class:SimulationScenarioContext; it cannot perform actuation. conformal_gate : TwinConformalGate | None Optional calibrated conformal twin-confidence gate. When supplied with twin_confidence_source, the gate scores every live policy tick before actions are applied. twin_confidence_source : TwinConfidenceSource | None Callable that returns a twin-confidence score for the current policy tick. It must be supplied together with conformal_gate.

Returns

SimulationResult A :class:SimulationResult.

Raises

ValueError If the spec declares no oscillators, an unsupported control mode, or an incomplete conformal-admission configuration. AuditError If strict_audit_integrity and the audit event-stream integrity check fails at close time (the recorded run cannot be trusted).

Source code in src/scpn_phase_orchestrator/runtime/simulation.py

def simulate(
    spec: BindingSpec,
    *,
    steps: int = 100,
    seed: int = 42,
    policy_enabled: bool = True,
    control_mode: SimulationControlMode = "supervisor_policy",
    audit_logger: AuditLogger | None = None,
    strict_audit_integrity: bool = True,
    binding_spec_path: Path | None = None,
    scenario_hook: ScenarioCallback | None = None,
    conformal_gate: TwinConformalGate | None = None,
    twin_confidence_source: TwinConfidenceSource | None = None,
) -> SimulationResult:
    """Advance a binding spec for ``steps`` and return the simulation outcome.

    Parameters
    ----------
    spec : BindingSpec
        A validated binding spec.
    steps : int
        Number of integration steps.
    seed : int
        RNG seed for the initial phases.
    policy_enabled : bool
        Closed-loop control feedback on (``True``) or open-loop baseline (``False``).
    control_mode : {"supervisor_policy"}
        The live control surface used by the generic binding-spec simulator.
        Koopman MPC is intentionally not a selectable ``simulate`` mode; it
        remains a review-only/offline proposal surface in
        :mod:`scpn_phase_orchestrator.runtime.dvoc_oscillation_damping`.
    audit_logger : AuditLogger | None
        Optional logger; when given, the header, per-step records, and events are
        written. The caller owns its lifecycle (close).
    strict_audit_integrity : bool
        Fail closed on a broken audit chain. When ``True`` (default) and the audit
        logger owns a protobuf event stream whose close-time integrity check fails,
        the run raises :class:`~scpn_phase_orchestrator.exceptions.AuditError`
        instead of returning a result with a tamper-evident but unenforced
        ``audit_event_stream_integrity`` field. Set ``False`` to attach the failed
        result for inspection without failing the run.
    binding_spec_path : Path | None
        Optional path to the spec, used only to locate an adjacent ``policy.yaml``. When
        ``None``, no domainpack policy rules are loaded (the supervisor policy still
        runs when ``policy_enabled``).
    scenario_hook : ScenarioCallback | None
        Optional deterministic per-step perturbation hook for benchmark and case-study
        scenarios. The hook can mutate phases, frequencies, coupling, ``zeta``, or
        ``psi_target`` in a validated :class:`SimulationScenarioContext`; it cannot
        perform actuation.
    conformal_gate : TwinConformalGate | None
        Optional calibrated conformal twin-confidence gate. When supplied with
        ``twin_confidence_source``, the gate scores every live policy tick before
        actions are applied.
    twin_confidence_source : TwinConfidenceSource | None
        Callable that returns a twin-confidence score for the current policy tick.
        It must be supplied together with ``conformal_gate``.

    Returns
    -------
    SimulationResult
        A :class:`SimulationResult`.

    Raises
    ------
    ValueError
        If the spec declares no oscillators, an unsupported control mode, or an
        incomplete conformal-admission configuration.
    AuditError
        If ``strict_audit_integrity`` and the audit event-stream integrity check
        fails at close time (the recorded run cannot be trusted).
    """
    if control_mode != "supervisor_policy":
        raise ValueError(
            "simulate control_mode must be 'supervisor_policy'; Koopman MPC is "
            "offline/review-only via runtime.dvoc_oscillation_damping"
        )
    if (conformal_gate is None) != (twin_confidence_source is None):
        raise ValueError(
            "conformal_gate and twin_confidence_source must be supplied together"
        )

    n_osc = sum(len(layer.oscillator_ids) for layer in spec.layers)
    if n_osc == 0:
        raise ValueError("no oscillators defined in layers")

    binding_summary = resolved_binding_config(spec)
    builder = CouplingBuilder()
    amplitude_mode = spec.amplitude is not None
    sl_engine: StuartLandauEngine | None = None
    upde_engine: UPDEEngine | None = None
    mu: FloatArray | None = None

    if amplitude_mode:
        amp = spec.amplitude
        assert amp is not None  # nosec B101
        coupling = builder.build_with_amplitude(
            n_osc,
            spec.coupling.base_strength,
            spec.coupling.decay_alpha,
            amp.amp_coupling_strength,
            amp.amp_coupling_decay,
        )
        sl_engine = StuartLandauEngine(n_osc, dt=spec.sample_period_s)
        mu = np.full(n_osc, amp.mu)
    else:
        coupling = builder.build(
            n_osc,
            spec.coupling.base_strength,
            spec.coupling.decay_alpha,
        )
        upde_engine = UPDEEngine(n_osc, dt=spec.sample_period_s)

    event_bus = EventBus()
    boundary_observer = BoundaryObserver(spec.boundaries)
    boundary_observer.set_event_bus(event_bus)
    regime_manager = RegimeManager(event_bus=event_bus)
    channel_runtime = ChannelRuntimeExecutor.from_spec(spec)

    petri_adapter: PetriNetAdapter | None = None
    if spec.protocol_net is not None:
        net, marking = petri_net_from_protocol(spec.protocol_net)
        petri_adapter = PetriNetAdapter(
            net,
            marking,
            spec.protocol_net.place_regime,
            event_bus=event_bus,
        )

    supervisor = SupervisorPolicy(regime_manager, petri_adapter=petri_adapter)
    projector = ActionProjector.from_actuator_mappings(spec.actuators)
    prev_values: dict[str, float] = {"K": 0.0, "zeta": 0.0, "alpha": 0.0, "Psi": 0.0}

    policy_engine: PolicyEngine | None = None
    if binding_spec_path is not None:
        policy_path = binding_spec_path.parent / "policy.yaml"
        if policy_path.exists():
            rules = load_policy_rules(policy_path)
            if rules:
                policy_engine = PolicyEngine(rules)

    imprint_model: ImprintModel | None = None
    imprint_state: ImprintState | None = None
    if spec.imprint_model is not None:
        imprint_model = ImprintModel(
            spec.imprint_model.decay_rate, spec.imprint_model.saturation
        )
        imprint_state = ImprintState(m_k=np.zeros(n_osc), last_update=0.0)

    geo_constraints: list[GeometryConstraint] = []
    if spec.geometry_prior is not None:
        ct = spec.geometry_prior.constraint_type.lower()
        if "symmetric" in ct:
            geo_constraints.append(SymmetryConstraint())
        if "non_negative" in ct or "nonneg" in ct:
            geo_constraints.append(NonNegativeConstraint())

    rng = np.random.default_rng(seed)
    phases = rng.uniform(0, 2 * np.pi, n_osc)
    omegas = np.array(spec.get_omegas(), dtype=np.float64)

    amplitudes = np.array([], dtype=np.float64)
    input_amplitudes = np.array([], dtype=np.float64)
    sl_state = np.array([], dtype=np.float64)
    phases_history: list[FloatArray] = []
    amps_history: list[FloatArray] = []
    if amplitude_mode and mu is not None:
        r_init = np.sqrt(np.maximum(mu, 0.0))
        sl_state = np.concatenate([phases, r_init])

    layer_osc_ranges: dict[int, list[int]] = {}
    osc_idx = 0
    for layer in spec.layers:
        n_layer = len(layer.oscillator_ids)
        layer_osc_ranges[layer.index] = list(range(osc_idx, osc_idx + n_layer))
        osc_idx += n_layer

    zeta = max(
        (cfg.get("zeta", 0.0) for cfg in spec.drivers.all_channel_configs().values()),
        default=0.0,
    )
    zeta_ttl = 0
    psi_target = spec.drivers.physical.get("psi", 0.0)

    psi_driver: PhysicalDriver | InformationalDriver | SymbolicDriver | None = None
    if "frequency" in spec.drivers.physical:
        psi_driver = PhysicalDriver(
            frequency=spec.drivers.physical["frequency"],
            amplitude=spec.drivers.physical.get("amplitude", 1.0),
        )
    elif "cadence_hz" in spec.drivers.informational:
        psi_driver = InformationalDriver(
            cadence_hz=spec.drivers.informational["cadence_hz"],
        )
    elif "sequence" in spec.drivers.symbolic:
        psi_driver = SymbolicDriver(
            sequence=spec.drivers.symbolic["sequence"],
        )

    control_interval = max(1, round(spec.control_period_s / spec.sample_period_s))

    if audit_logger is not None:
        audit_logger.log_header(
            n_oscillators=n_osc,
            dt=spec.sample_period_s,
            seed=seed,
            amplitude_mode=amplitude_mode,
            binding_config=binding_summary,
            control_mode=control_mode,
        )

    r_good_history: list[float] = []
    r_bad_history: list[float] = []
    boundary_violation_total = 0
    action_total = 0
    conformal_decisions: list[tuple[int, ConformalDecision]] = []
    eff_mu = mu

    for step_idx in range(steps):
        if zeta_ttl > 0:
            zeta_ttl -= 1
            if zeta_ttl == 0:
                zeta = 0.0

        if psi_driver is not None:
            t = step_idx * spec.sample_period_s
            if isinstance(psi_driver, SymbolicDriver):
                psi_target = psi_driver.compute(step_idx)
            else:
                psi_target = psi_driver.compute(t)

        if scenario_hook is not None:
            context = SimulationScenarioContext(
                spec_name=spec.name,
                step=step_idx,
                sample_period_s=spec.sample_period_s,
                phases=phases,
                omegas=omegas,
                coupling=coupling,
                zeta=zeta,
                psi_target=psi_target,
                layer_osc_ranges=layer_osc_ranges,
                rng=rng,
            )
            scenario_hook(context)
            phases, omegas, coupling, zeta, psi_target = _apply_scenario_context(
                context,
                n_osc=n_osc,
            )

        eff_knm = coupling.knm
        eff_alpha = coupling.alpha
        if imprint_model is not None and imprint_state is not None:
            eff_knm = imprint_model.modulate_coupling(eff_knm, imprint_state)
            eff_alpha = imprint_model.modulate_lag(eff_alpha, imprint_state)
        if geo_constraints:
            eff_knm = project_knm(eff_knm, geo_constraints)

        input_phases = phases.copy()
        logged_zeta = zeta
        logged_psi = psi_target
        if amplitude_mode and sl_engine is not None and mu is not None:
            assert coupling.knm_r is not None  # nosec B101
            eff_mu = mu
            if imprint_model is not None and imprint_state is not None:
                eff_mu = imprint_model.modulate_mu(mu, imprint_state)
            # Capture the pre-step amplitudes so the audit record pairs them with
            # the pre-step phases. Logging post-step amplitudes alongside pre-step
            # phases yields a mixed state that no single step produced, which is
            # what broke `spo replay --verify` for amplitude-driven specs.
            input_amplitudes = sl_state[n_osc:].copy()
            sl_state = sl_engine.step(
                sl_state,
                omegas,
                eff_mu,
                eff_knm,
                coupling.knm_r,
                zeta,
                psi_target,
                eff_alpha,
                epsilon=spec.amplitude.epsilon,  # type: ignore[union-attr]  # amplitude_mode guard guarantees spec.amplitude is not None
            )
            phases = sl_state[:n_osc]
            amplitudes = sl_state[n_osc:]
            phases_history.append(phases.copy())
            amps_history.append(amplitudes.copy())
        else:
            assert upde_engine is not None  # nosec B101
            phases = upde_engine.step(
                phases, omegas, eff_knm, zeta, psi_target, eff_alpha
            )

        layer_states = []
        for layer in spec.layers:
            osc_ids = layer_osc_ranges[layer.index]
            if osc_ids:
                r, psi_l = compute_order_parameter(phases[osc_ids])
            else:
                r, psi_l = 0.0, 0.0
            ls_kwargs: dict[str, Any] = {"R": r, "psi": psi_l}
            if amplitude_mode:
                layer_r = amplitudes[osc_ids] if osc_ids else np.array([])
                if layer_r.size > 0:
                    ls_kwargs["mean_amplitude"] = float(np.mean(layer_r))
                    mean_r = float(np.mean(layer_r))
                    if mean_r > 0:
                        ls_kwargs["amplitude_spread"] = float(np.std(layer_r) / mean_r)
            layer_states.append(LayerState(**ls_kwargs))

        n_layers = len(spec.layers)
        cla = np.zeros((n_layers, n_layers))
        for li in range(n_layers):
            for lj in range(li + 1, n_layers):
                ids_i = layer_osc_ranges[spec.layers[li].index]
                ids_j = layer_osc_ranges[spec.layers[lj].index]
                if ids_i and ids_j:
                    pi, pj = phases[ids_i], phases[ids_j]
                    min_len = min(len(pi), len(pj))
                    plv = compute_plv(pi[:min_len], pj[:min_len])
                    cla[li, lj] = plv
                    cla[lj, li] = plv

        runtime_execution = channel_runtime.execute(layer_states)
        executed_layer_states = list(runtime_execution.layers)

        mean_r_val = (
            float(np.mean([ls.R for ls in executed_layer_states]))
            if executed_layer_states
            else 0.0
        )
        state_kwargs: dict[str, Any] = {
            "layers": executed_layer_states,
            "cross_layer_alignment": cla,
            "stability_proxy": mean_r_val,
            "regime_id": regime_manager.current_regime.value,
        }
        if amplitude_mode:
            state_kwargs["mean_amplitude"] = float(np.mean(amplitudes))
            sub_count = int(np.sum(amplitudes < 0.1))
            state_kwargs["subcritical_fraction"] = (
                sub_count / n_osc if n_osc > 0 else 0.0
            )
            if len(phases_history) >= 20:
                recent_ph = np.array(phases_history[-20:])
                recent_am = np.array(amps_history[-20:])
                pac_vals = [
                    modulation_index(recent_ph[:, i], recent_am[:, i])
                    for i in range(n_osc)
                ]
                state_kwargs["pac_max"] = float(max(pac_vals))

        if imprint_state is not None:
            state_kwargs["imprint_mean"] = float(np.mean(imprint_state.m_k))

        obs_values: dict[str, float] = {"R": state_kwargs["stability_proxy"]}
        if amplitude_mode:
            obs_values["mean_amplitude"] = state_kwargs.get("mean_amplitude", 0.0)
            obs_values["pac_max"] = state_kwargs.get("pac_max", 0.0)
            obs_values["subcritical_fraction"] = state_kwargs.get(
                "subcritical_fraction", 0.0
            )
        for i, ls in enumerate(executed_layer_states):
            obs_values[f"R_{i}"] = ls.R
        boundary_state = boundary_observer.observe(obs_values, step=step_idx)
        state_kwargs["boundary_violation_count"] = len(boundary_state.violations)
        boundary_violation_total += len(boundary_state.violations)
        upde_state = UPDEState(**state_kwargs)

        actions: list[Any] = []
        if policy_enabled and step_idx % control_interval == 0:
            actions = supervisor.decide(
                upde_state, boundary_state, petri_ctx=obs_values
            )
            if policy_engine is not None:
                actions.extend(
                    policy_engine.evaluate(
                        regime_manager.current_regime,
                        upde_state,
                        spec.objectives.good_layers,
                        spec.objectives.bad_layers,
                    )
                )
            actions = [
                projector.project(a, prev_values.get(a.knob, 0.0)) for a in actions
            ]
            if conformal_gate is not None and twin_confidence_source is not None:
                confidence_context = SimulationTwinConfidenceContext(
                    spec_name=spec.name,
                    step=step_idx,
                    sample_period_s=spec.sample_period_s,
                    model_phases=phases.copy(),
                    r_good=_objective_r(
                        phases,
                        spec.objectives.good_layers,
                        layer_osc_ranges,
                    ),
                    r_bad=_objective_r(
                        phases,
                        spec.objectives.bad_layers,
                        layer_osc_ranges,
                    ),
                    regime=regime_manager.current_regime.value,
                    boundary_violation_count=len(boundary_state.violations),
                    proposed_action_count=len(actions),
                    layer_order_parameters=tuple(
                        float(layer_state.R) for layer_state in executed_layer_states
                    ),
                )
                confidence_score = twin_confidence_source(confidence_context)
                if not isinstance(confidence_score, TwinConfidenceScore):
                    raise ValueError(
                        "twin_confidence_source must return TwinConfidenceScore"
                    )
                decision = conformal_gate.update(
                    confidence_nonconformity(confidence_score),
                    regime=confidence_context.regime,
                )
                conformal_decisions.append((step_idx, decision))
                if not decision.admitted:
                    actions = []
        action_total += len(actions)

        for act in actions:
            if act.knob == "zeta":
                zeta = max(0.0, min(zeta + act.value, 0.5))
                zeta_ttl = int(act.ttl_s / spec.sample_period_s)
            elif act.knob == "K":
                if act.scope == "global":
                    coupling = CouplingState(
                        knm=coupling.knm * (1.0 + act.value),
                        alpha=coupling.alpha,
                        active_template=coupling.active_template,
                        knm_r=coupling.knm_r,
                    )
                elif act.scope.startswith("layer_"):
                    idx = int(act.scope.split("_", 1)[1])
                    new_knm = coupling.knm.copy()
                    new_knm[idx, :] *= 1.0 + act.value
                    new_knm[:, idx] *= 1.0 + act.value
                    new_knm[idx, idx] = 0.0
                    coupling = CouplingState(
                        knm=new_knm,
                        alpha=coupling.alpha,
                        active_template=coupling.active_template,
                        knm_r=coupling.knm_r,
                    )
            elif act.knob == "Psi":
                psi_target = act.value
            prev_values[act.knob] = act.value

        if imprint_model is not None and imprint_state is not None:
            exposure = np.array(
                [
                    layer_states[i].R
                    for i, layer in enumerate(spec.layers)
                    for _ in layer.oscillator_ids
                ]
            )
            imprint_state = imprint_model.update(
                imprint_state, exposure, spec.sample_period_s
            )

        if audit_logger is not None:
            log_kwargs: dict[str, Any] = {
                "phases": input_phases,
                "omegas": omegas,
                "knm": eff_knm,
                "alpha": eff_alpha,
                "zeta": logged_zeta,
                "psi_drive": logged_psi,
                "channel_runtime": runtime_execution.to_audit_record(),
            }
            if amplitude_mode:
                log_kwargs["amplitudes"] = input_amplitudes
                log_kwargs["mu"] = eff_mu
                log_kwargs["knm_r"] = coupling.knm_r
                log_kwargs["epsilon"] = spec.amplitude.epsilon  # type: ignore[union-attr]  # amplitude_mode guard guarantees spec.amplitude is not None
            audit_logger.log_step(step_idx, upde_state, actions, **log_kwargs)

        r_good_history.append(
            _objective_r(phases, spec.objectives.good_layers, layer_osc_ranges)
        )
        r_bad_history.append(
            _objective_r(phases, spec.objectives.bad_layers, layer_osc_ranges)
        )

    if audit_logger is not None:
        for step_idx, decision in conformal_decisions:
            audit_logger.log_event(
                "conformal_admission",
                {"step": step_idx, "decision": decision.to_audit_record()},
            )
        for evt in event_bus.history:
            audit_logger.log_event(evt.kind, {"step": evt.step, "detail": evt.detail})
        audit_event_stream_integrity = audit_logger.verify_event_stream_integrity()
        if (
            strict_audit_integrity
            and audit_event_stream_integrity is not None
            and not audit_event_stream_integrity.ok
        ):
            raise AuditError(
                "audit event-stream integrity verification failed for "
                f"{audit_event_stream_integrity.event_stream_path!r} after "
                f"{audit_event_stream_integrity.verified_events} verified event(s); "
                "the recorded run cannot be trusted. Pass "
                "strict_audit_integrity=False to attach the failed result instead "
                "of failing closed."
            )
    else:
        audit_event_stream_integrity = None

    r_good = r_good_history[-1] if r_good_history else 0.0
    r_bad = r_bad_history[-1] if r_bad_history else 0.0
    mean_amplitude = (
        float(np.mean(amplitudes)) if amplitude_mode and amplitudes.size > 0 else None
    )

    return SimulationResult(
        spec_name=spec.name,
        steps=steps,
        policy_enabled=policy_enabled,
        control_mode=control_mode,
        amplitude_mode=amplitude_mode,
        final_phases=phases,
        final_amplitudes=amplitudes if amplitude_mode else None,
        r_good=r_good,
        r_bad=r_bad,
        separation=r_good - r_bad,
        final_regime=regime_manager.current_regime.value,
        mean_amplitude=mean_amplitude,
        r_good_history=tuple(r_good_history),
        r_bad_history=tuple(r_bad_history),
        boundary_violation_total=boundary_violation_total,
        action_total=action_total,
        conformal_admission_total=len(conformal_decisions),
        conformal_admission_rejections=sum(
            1 for _, decision in conformal_decisions if not decision.admitted
        ),
        last_conformal_admission=(
            conformal_decisions[-1][1] if conformal_decisions else None
        ),
        audit_event_stream_integrity=audit_event_stream_integrity,
    )

evolutionary_grammar

CLI commands for the review-only evolutionary supervisor grammar family.

Three operator commands wrap the deterministic offline mutation-search grammars in supervisor.evolutionary_policy_dsl, supervisor.evolutionary_petri_grammar, and supervisor.evolutionary_topology_grammar. Each reads a local source artefact, runs the offline search, and emits one deterministic review bundle. None of the commands actuate, merge, hot-patch, or execute any mutated candidate; they only generate operator-review evidence.

Functions:

evolutionary_policy_dsl_search

evolutionary_policy_dsl_search(
    policy_dsl_file: Path,
    generations: int,
    population: int,
    mutation_step: float,
    output: Path | None,
) -> None

Emit review evidence for offline policy-DSL mutation candidates.

Parameters

policy_dsl_file : Path Text file containing the compact policy-DSL source. generations : int Number of search generations. population : int Number of candidates per generation. mutation_step : float Mutation step size applied per generation. output : Path | None Optional path to write the emitted bundle JSON.

Raises

ClickException If the DSL source or search parameters are invalid.

Source code in src/scpn_phase_orchestrator/runtime/cli/evolutionary_grammar.py

@main.command("evolutionary-policy-dsl-search")
@click.argument(
    "policy_dsl_file",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
    "--generations", type=int, default=2, help="Number of search generations."
)
@click.option("--population", type=int, default=6, help="Candidates per generation.")
@click.option(
    "--mutation-step",
    type=float,
    default=0.05,
    help="Mutation step size per generation.",
)
@_OUTPUT_OPTION
def evolutionary_policy_dsl_search(
    policy_dsl_file: Path,
    generations: int,
    population: int,
    mutation_step: float,
    output: Path | None,
) -> None:
    """Emit review evidence for offline policy-DSL mutation candidates.

    Parameters
    ----------
    policy_dsl_file : Path
        Text file containing the compact policy-DSL source.
    generations : int
        Number of search generations.
    population : int
        Number of candidates per generation.
    mutation_step : float
        Mutation step size applied per generation.
    output : Path | None
        Optional path to write the emitted bundle JSON.

    Raises
    ------
    ClickException
        If the DSL source or search parameters are invalid.
    """
    policy_dsl = _read_text(policy_dsl_file, artifact="policy DSL")
    try:
        report = run_offline_evolutionary_policy_dsl_search(
            policy_dsl,
            generation_count=generations,
            population_size=population,
            mutation_step=mutation_step,
        )
    except ValueError as exc:
        raise click.ClickException(
            f"evolutionary policy-DSL search failed: {exc}"
        ) from exc
    _emit(_grammar_bundle("policy-dsl", report.to_audit_record()), output)

evolutionary_petri_mutation

evolutionary_petri_mutation(
    net_json: Path,
    generations: int,
    candidates_per_generation: int,
    mutation_step: float,
    max_arc_weight: int,
    max_token_bound: int,
    output: Path | None,
) -> None

Emit review evidence for offline Petri-net mutation candidates.

Parameters

net_json : Path JSON file with a net-like payload of places, transitions, and arcs. generations : int Number of search generations. candidates_per_generation : int 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. output : Path | None Optional path to write the emitted bundle JSON.

Raises

ClickException If the net payload or bounds are invalid.

Source code in src/scpn_phase_orchestrator/runtime/cli/evolutionary_grammar.py

@main.command("evolutionary-petri-mutation")
@click.argument(
    "net_json",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
    "--generations", type=int, default=2, help="Number of search generations."
)
@click.option(
    "--candidates-per-generation",
    type=int,
    default=6,
    help="Candidates evaluated per generation.",
)
@click.option(
    "--mutation-step",
    type=float,
    default=0.1,
    help="Mutation step size per generation.",
)
@click.option(
    "--max-arc-weight", type=int, default=4, help="Maximum mutated arc weight."
)
@click.option(
    "--max-token-bound", type=int, default=128, help="Maximum token count per place."
)
@_OUTPUT_OPTION
def evolutionary_petri_mutation(
    net_json: Path,
    generations: int,
    candidates_per_generation: int,
    mutation_step: float,
    max_arc_weight: int,
    max_token_bound: int,
    output: Path | None,
) -> None:
    """Emit review evidence for offline Petri-net mutation candidates.

    Parameters
    ----------
    net_json : Path
        JSON file with a net-like payload of places, transitions, and arcs.
    generations : int
        Number of search generations.
    candidates_per_generation : int
        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.
    output : Path | None
        Optional path to write the emitted bundle JSON.

    Raises
    ------
    ClickException
        If the net payload or bounds are invalid.
    """
    net_like = _load_net_like(net_json)
    try:
        plan = run_offline_evolutionary_petri_mutation_grammar(
            net_like,
            generation_count=generations,
            candidates_per_generation=candidates_per_generation,
            mutation_step=mutation_step,
            max_arc_weight=max_arc_weight,
            max_token_bound=max_token_bound,
        )
    except ValueError as exc:
        raise click.ClickException(
            f"evolutionary Petri mutation failed: {exc}"
        ) from exc
    _emit(_grammar_bundle("petri", plan.to_audit_record()), output)

evolutionary_topology_mutation

evolutionary_topology_mutation(
    topology_json: Path,
    generations: int,
    population: int,
    mutation_step: float,
    min_edge_weight: float,
    max_edge_weight: float,
    edge_add_base_weight: float,
    max_add_candidates: int,
    output: Path | None,
) -> None

Emit review evidence for offline topology mutation candidates.

Parameters

topology_json : Path JSON file with a nodes array and an edges array. generations : int Number of search generations. population : 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. output : Path | None Optional path to write the emitted bundle JSON.

Raises

ClickException If the topology records or bounds are invalid.

Source code in src/scpn_phase_orchestrator/runtime/cli/evolutionary_grammar.py

@main.command("evolutionary-topology-mutation")
@click.argument(
    "topology_json",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
    "--generations", type=int, default=2, help="Number of search generations."
)
@click.option("--population", type=int, default=8, help="Candidates per generation.")
@click.option(
    "--mutation-step",
    type=float,
    default=0.05,
    help="Mutation step size per generation.",
)
@click.option(
    "--min-edge-weight", type=float, default=0.0, help="Minimum retained edge weight."
)
@click.option(
    "--max-edge-weight", type=float, default=10.0, help="Maximum allowed edge weight."
)
@click.option(
    "--edge-add-base-weight",
    type=float,
    default=0.4,
    help="Base weight assigned to added edges.",
)
@click.option(
    "--max-add-candidates",
    type=int,
    default=16,
    help="Maximum number of edge-addition candidates.",
)
@_OUTPUT_OPTION
def evolutionary_topology_mutation(
    topology_json: Path,
    generations: int,
    population: int,
    mutation_step: float,
    min_edge_weight: float,
    max_edge_weight: float,
    edge_add_base_weight: float,
    max_add_candidates: int,
    output: Path | None,
) -> None:
    """Emit review evidence for offline topology mutation candidates.

    Parameters
    ----------
    topology_json : Path
        JSON file with a ``nodes`` array and an ``edges`` array.
    generations : int
        Number of search generations.
    population : 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.
    output : Path | None
        Optional path to write the emitted bundle JSON.

    Raises
    ------
    ClickException
        If the topology records or bounds are invalid.
    """
    payload = _load_json_file(topology_json, artifact="topology")
    nodes = _json_array_of_objects(payload, "nodes")
    edges = _json_array_of_objects(payload, "edges")
    try:
        report = run_offline_evolutionary_topology_mutation_search(
            nodes,
            edges,
            generation_count=generations,
            population_size=population,
            mutation_step=mutation_step,
            min_edge_weight=min_edge_weight,
            max_edge_weight=max_edge_weight,
            edge_add_base_weight=edge_add_base_weight,
            max_add_candidates=max_add_candidates,
        )
    except ValueError as exc:
        raise click.ClickException(
            f"evolutionary topology mutation failed: {exc}"
        ) from exc
    _emit(_grammar_bundle("topology", report.to_audit_record()), output)

federated_dp_noise_service

CLI command for review-only federated DP noise-service preflight evidence.

The command consumes a DP-noise request declaration plus a deployment declaration, builds the deterministic DP-noise request and response manifests, then a review-only deployment preflight manifest, and emits a non-actuating preflight bundle. It validates the same production supervisor surface as supervisor.federated_dp_noise_service and never opens sockets, generates live noise, or permits live DP noise-service execution. Missing deployment prerequisites are reported as a not-ready readiness verdict rather than an error; only malformed inputs fail closed.

Classes

Functions:

federated_dp_noise_service_preflight

federated_dp_noise_service_preflight(
    request_json: Path,
    deployment_json: Path,
    output: Path | None,
) -> None

Emit deterministic review evidence for a DP noise-service deployment.

Parameters

request_json : Path JSON file describing the DP-noise request (privacy parameters, seed commitment, policy keys, and per-node privacy budgets). deployment_json : Path JSON file describing the deployment preflight inputs (mechanism, custody, accountant, budget issuer, service endpoint, and operator approval). output : Path | None Optional path to write the emitted bundle JSON.

Raises

ClickException If inputs are malformed or the DP noise-service preflight fails closed. A merely not-ready deployment is reported in the bundle, not raised.

Source code in src/scpn_phase_orchestrator/runtime/cli/federated_dp_noise_service.py

@main.command("federated-dp-noise-service-preflight")
@click.argument(
    "request_json",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.argument(
    "deployment_json",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
    "--output",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help="Optional path for the deterministic preflight bundle JSON.",
)
def federated_dp_noise_service_preflight(
    request_json: Path,
    deployment_json: Path,
    output: Path | None,
) -> None:
    """Emit deterministic review evidence for a DP noise-service deployment.

    Parameters
    ----------
    request_json : Path
        JSON file describing the DP-noise request (privacy parameters, seed
        commitment, policy keys, and per-node privacy budgets).
    deployment_json : Path
        JSON file describing the deployment preflight inputs (mechanism, custody,
        accountant, budget issuer, service endpoint, and operator approval).
    output : Path | None
        Optional path to write the emitted bundle JSON.

    Raises
    ------
    ClickException
        If inputs are malformed or the DP noise-service preflight fails closed.
        A merely not-ready deployment is reported in the bundle, not raised.
    """
    request_payload = _load_json_file(
        request_json,
        artifact="federated DP noise-service request",
    )
    deployment = _load_json_file(
        deployment_json,
        artifact="federated DP noise-service deployment",
    )
    try:
        request = _build_request(request_payload)
        response = build_dp_noise_service_manifest(request)
        preflight = build_dp_noise_service_deployment_preflight_manifest(
            request,
            response,
            mechanism_label=_string_field(deployment, "mechanism_label"),
            privacy_accountant_owner=_string_field(
                deployment, "privacy_accountant_owner"
            ),
            seed_custody_label=_string_field(deployment, "seed_custody_label"),
            budget_issuer_label=_string_field(deployment, "budget_issuer_label"),
            service_endpoint_label=_string_field(deployment, "service_endpoint_label"),
            operator_approved=_bool_field(deployment, "operator_approved"),
        )
    except (TypeError, ValueError) as exc:
        raise click.ClickException(
            f"federated DP noise-service preflight failed: {exc}"
        ) from exc

    bundle = _preflight_bundle_payload(
        deployment_ready=preflight.deployment_readiness.ready,
        deployment_reason=preflight.deployment_readiness.reason,
        request_hash=preflight.request_hash,
        response_hash=preflight.response_hash,
        request=request.to_audit_record(),
        response=response.to_audit_record(),
        preflight=preflight.to_audit_record(),
    )
    rendered = json.dumps(bundle, indent=2, sort_keys=True)
    if output is not None:
        try:
            output.write_text(rendered + "\n", encoding="utf-8")
        except OSError as exc:
            raise click.ClickException(
                "cannot write federated DP noise-service preflight bundle "
                f"{output!s}: {exc}"
            ) from exc
    click.echo(rendered)

federated_secure_aggregation

CLI command for review-only federated secure-aggregation preflight evidence.

The command consumes newline-delimited secure-aggregation node commitment records plus a deployment declaration, builds the deterministic secure-aggregation manifest, then a review-only deployment preflight manifest, and emits a non-actuating preflight bundle. It validates the same production supervisor surface as supervisor.federated_secure_aggregation and never opens sockets, runs aggregation, or permits live secure-aggregation execution.

Classes

Functions:

federated_secure_aggregation_preflight

federated_secure_aggregation_preflight(
    node_commitments_jsonl: Path,
    deployment_json: Path,
    output: Path | None,
) -> None

Emit deterministic review evidence for a secure-aggregation deployment.

Parameters

node_commitments_jsonl : Path JSONL file containing secure-aggregation node commitment records. deployment_json : Path JSON file describing the aggregation policy and deployment preflight inputs (quorum evidence, custody records, operator approval). output : Path | None Optional path to write the emitted bundle JSON.

Raises

ClickException If inputs are malformed or the secure-aggregation preflight fails closed.

Source code in src/scpn_phase_orchestrator/runtime/cli/federated_secure_aggregation.py

@main.command("federated-secure-aggregation-preflight")
@click.argument(
    "node_commitments_jsonl",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.argument(
    "deployment_json",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
    "--output",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help="Optional path for the deterministic preflight bundle JSON.",
)
def federated_secure_aggregation_preflight(
    node_commitments_jsonl: Path,
    deployment_json: Path,
    output: Path | None,
) -> None:
    """Emit deterministic review evidence for a secure-aggregation deployment.

    Parameters
    ----------
    node_commitments_jsonl : Path
        JSONL file containing secure-aggregation node commitment records.
    deployment_json : Path
        JSON file describing the aggregation policy and deployment preflight
        inputs (quorum evidence, custody records, operator approval).
    output : Path | None
        Optional path to write the emitted bundle JSON.

    Raises
    ------
    ClickException
        If inputs are malformed or the secure-aggregation preflight fails closed.
    """
    commitments = _load_node_commitment_jsonl(node_commitments_jsonl)
    deployment = _load_json_file(
        deployment_json,
        artifact="federated secure aggregation deployment",
    )
    try:
        manifest = _build_manifest(commitments, deployment)
        preflight = _build_preflight(manifest, deployment)
    except (TypeError, ValueError) as exc:
        raise click.ClickException(
            f"federated secure aggregation preflight failed: {exc}"
        ) from exc

    bundle = _preflight_bundle_payload(
        manifest=manifest.to_audit_record(),
        preflight=preflight.to_audit_record(),
    )
    rendered = json.dumps(bundle, indent=2, sort_keys=True)
    if output is not None:
        try:
            output.write_text(rendered + "\n", encoding="utf-8")
        except OSError as exc:
            raise click.ClickException(
                "cannot write federated secure aggregation preflight bundle "
                f"{output!s}: {exc}"
            ) from exc
    click.echo(rendered)

federated_transport

CLI command for review-only federated transport preflight evidence.

The command consumes newline-delimited federated node update audit records plus a transport declaration, then builds signed/hash-linked envelopes, replays the batch, and emits a deterministic non-actuating preflight bundle. It validates the same production supervisor transport surface as supervisor.federated_transport and never opens sockets or permits live transport execution.

Functions:

federated_transport_preflight

federated_transport_preflight(
    node_updates_jsonl: Path,
    transport_declaration_json: Path,
    output: Path | None,
) -> None

Emit deterministic review evidence for a federated transport batch.

Parameters

node_updates_jsonl : Path JSONL file containing node update audit records. transport_declaration_json : Path JSON file describing the intended transport boundary. output : Path | None Optional path to write the emitted bundle JSON.

Raises

ClickException If inputs are malformed or the transport preflight fails closed.

Source code in src/scpn_phase_orchestrator/runtime/cli/federated_transport.py

@main.command("federated-transport-preflight")
@click.argument(
    "node_updates_jsonl",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.argument(
    "transport_declaration_json",
    type=click.Path(exists=True, dir_okay=False, path_type=Path),
)
@click.option(
    "--output",
    type=click.Path(dir_okay=False, path_type=Path),
    default=None,
    help="Optional path for the deterministic preflight bundle JSON.",
)
def federated_transport_preflight(
    node_updates_jsonl: Path,
    transport_declaration_json: Path,
    output: Path | None,
) -> None:
    """Emit deterministic review evidence for a federated transport batch.

    Parameters
    ----------
    node_updates_jsonl : Path
        JSONL file containing node update audit records.
    transport_declaration_json : Path
        JSON file describing the intended transport boundary.
    output : Path | None
        Optional path to write the emitted bundle JSON.

    Raises
    ------
    ClickException
        If inputs are malformed or the transport preflight fails closed.
    """
    updates = _load_node_update_jsonl(node_updates_jsonl)
    declaration = _load_json_file(
        transport_declaration_json,
        artifact="federated transport declaration",
    )
    try:
        envelopes = build_signed_transport_envelopes(updates)
        replay_ledger = replay_federated_transport_batch(envelopes)
        preflight_manifest = build_transport_deployment_preflight_manifest(
            declaration,
            replay_ledger=replay_ledger,
        )
    except ValueError as exc:
        raise click.ClickException(
            f"federated transport preflight failed: {exc}"
        ) from exc

    bundle = _preflight_bundle_payload(
        envelopes=[envelope.to_audit_record() for envelope in envelopes],
        replay_ledger=replay_ledger.to_audit_record(),
        preflight_manifest=preflight_manifest.to_audit_record(),
    )
    rendered = json.dumps(bundle, indent=2, sort_keys=True)
    if output is not None:
        try:
            output.write_text(rendered + "\n", encoding="utf-8")
        except OSError as exc:
            raise click.ClickException(
                f"cannot write federated transport preflight bundle {output!s}: {exc}"
            ) from exc
    click.echo(rendered)

Simulation core

runtime.simulation is the single non-actuating closed/open-loop core that backs both spo run and the public evaluate_binding_spec facade, so a binding spec is advanced by exactly one implementation. policy_enabled is the open/closed-loop switch: with it on, the supervisor and domainpack policy feed bounded actions back into the dynamics; with it off, the same drivers and intrinsic plasticity run without control feedback, giving the baseline for measuring orchestration uplift on a fixed seed.

The live control mode is deliberately explicit and narrow: control_mode="supervisor_policy" is the only accepted simulate() mode. Koopman MPC is not a hidden or partially wired binding-spec simulator mode; it is kept in the offline/review-only dVOC damping pipeline where the fitted predictor, plant model, and PRC evidence boundary are explicit.

simulate() also accepts an optional scenario_hook for deterministic non-actuating perturbation schedules. The hook receives a SimulationScenarioContext before each integration step and may adjust phases, natural frequencies, coupling state, zeta, or psi_target; the core validates finite vector shapes, scalar types, and CouplingState identity immediately after the hook returns. This is the supported path for benchmark and case-study scenario evidence, not a hardware or live-actuation path.

For deployments that have an observed-twin confidence stream, simulate() can also receive a calibrated TwinConformalGate plus a twin_confidence_source. The source receives SimulationTwinConfidenceContext on each live policy tick and returns a TwinConfidenceScore; the gate scores that value through confidence_nonconformity. A rejected conformal decision fails closed for that tick by suppressing all proposed policy actions, and the result records the admission count, rejection count, and last decision. Audit-enabled runs emit a conformal_admission event for every scored policy tick. The default CLI run does not synthesize an observed twin feed, so the gate is opt-in rather than implicitly active for every binding spec.

simulation

Single-source closed/open-loop simulation core for binding specs.

The CLI spo run command and the public Orchestrator.evaluate API both call :func:simulate, so a binding spec is advanced by exactly one implementation — there is no second loop to drift out of fidelity.

The core is non-actuating: it never writes to hardware, opens a network connection, or enforces a safety tier (tier enforcement is a caller policy, kept in the CLI). It supports both the Kuramoto (UPDE) and amplitude (Stuart-Landau) engines, optional Hebbian imprint plasticity, geometry-prior projection, exogenous physical/informational/symbolic drivers, Petri-net protocol gating, boundary observation, and the supervisor + domainpack policy control loop.

policy_enabled is the open/closed-loop switch:

• True (closed loop) — the supervisor and domainpack policy evaluate the state every control interval and their bounded, projected actions feed back into coupling, lag, damping, and drive. This is what spo run uses.
• False (open loop) — the same exogenous drivers and intrinsic plasticity still run, but no control feedback is applied. This is the baseline against which the closed-loop orchestration uplift is measured on the same seed.

control_mode is intentionally narrower than the actuation catalogue. The generic binding-spec loop accepts only "supervisor_policy"; Koopman MPC stays in the specialized offline/review-only dVOC damping pipeline where its fitted predictor, plant model, and evidence boundaries are explicit.

Classes

SimulationResult dataclass

SimulationResult(
    spec_name: str,
    steps: int,
    policy_enabled: bool,
    control_mode: str,
    amplitude_mode: bool,
    final_phases: FloatArray,
    final_amplitudes: FloatArray | None,
    r_good: float,
    r_bad: float,
    separation: float,
    final_regime: str,
    mean_amplitude: float | None,
    r_good_history: tuple[float, ...],
    r_bad_history: tuple[float, ...],
    boundary_violation_total: int,
    action_total: int,
    conformal_admission_total: int = 0,
    conformal_admission_rejections: int = 0,
    last_conformal_admission: ConformalDecision
    | None = None,
    audit_event_stream_integrity: AuditStreamIntegrityResult
    | None = None,
)

Outcome of one :func:simulate run.

Attributes

spec_name: Binding-spec name.
steps: Number of steps advanced.
policy_enabled: Whether the closed-loop control feedback was active.
control_mode: Live control surface used by the simulation core.
amplitude_mode: Whether the Stuart-Landau (amplitude) engine was used.
final_phases: Final oscillator phases, shape ``(n,)``.
final_amplitudes: Final amplitudes for amplitude mode, else ``None``.
r_good: Final order parameter over the objective good layers.
r_bad: Final order parameter over the objective bad layers.
separation: ``r_good - r_bad`` (the coherence objective margin).
final_regime: Final regime label.
mean_amplitude: Final mean amplitude for amplitude mode, else ``None``.
r_good_history / r_bad_history: Per-step good/bad order parameters.
boundary_violation_total: Summed boundary violations across steps.
action_total: Total projected control actions applied (0 when open loop).
conformal_admission_total: Number of live policy ticks scored by the
    conformal twin-confidence admission gate.
conformal_admission_rejections: Number of scored policy ticks rejected by
    the conformal admission gate. Rejected ticks suppress all proposed
    actions for that control interval.
last_conformal_admission: Last conformal admission decision, when the
    optional gate was active.
audit_event_stream_integrity: Close-time protobuf audit-stream integrity
    summary when an event stream was written, else ``None``. Under the
    default ``strict_audit_integrity`` this is always ``ok`` (a failed
    check fails the run closed); it can carry a failed summary only when
    :func:`simulate` was called with ``strict_audit_integrity=False``.

Methods:

to_record

to_record() -> dict[str, object]

Return a deterministic JSON-serialisable summary (history omitted).

Returns

dict[str, object] Return a deterministic JSON-serialisable summary (history omitted).

Source code in src/scpn_phase_orchestrator/runtime/simulation.py

def to_record(self) -> dict[str, object]:
    """Return a deterministic JSON-serialisable summary (history omitted).

    Returns
    -------
    dict[str, object]
        Return a deterministic JSON-serialisable summary (history omitted).
    """
    return {
        "spec_name": self.spec_name,
        "steps": self.steps,
        "policy_enabled": self.policy_enabled,
        "control_mode": self.control_mode,
        "amplitude_mode": self.amplitude_mode,
        "r_good": self.r_good,
        "r_bad": self.r_bad,
        "separation": self.separation,
        "final_regime": self.final_regime,
        "mean_amplitude": self.mean_amplitude,
        "boundary_violation_total": self.boundary_violation_total,
        "action_total": self.action_total,
        "conformal_admission_total": self.conformal_admission_total,
        "conformal_admission_rejections": self.conformal_admission_rejections,
        "last_conformal_admission": (
            self.last_conformal_admission.to_audit_record()
            if self.last_conformal_admission is not None
            else None
        ),
        "audit_event_stream_integrity": (
            self.audit_event_stream_integrity.to_audit_record()
            if self.audit_event_stream_integrity is not None
            else None
        ),
    }

SimulationScenarioContext dataclass

SimulationScenarioContext(
    spec_name: str,
    step: int,
    sample_period_s: float,
    phases: FloatArray,
    omegas: FloatArray,
    coupling: CouplingState,
    zeta: float,
    psi_target: float,
    layer_osc_ranges: dict[int, list[int]],
    rng: Generator,
)

Mutable, validated per-step scenario hook context.

Scenario hooks are non-actuating perturbation fixtures for benchmarks and case-study replay. They may mutate phase/frequency arrays in place or assign updated coupling, zeta, and psi_target values. The simulation core validates those fields immediately after the hook returns.

SimulationTwinConfidenceContext dataclass

SimulationTwinConfidenceContext(
    spec_name: str,
    step: int,
    sample_period_s: float,
    model_phases: FloatArray,
    r_good: float,
    r_bad: float,
    regime: str,
    boundary_violation_count: int,
    proposed_action_count: int,
    layer_order_parameters: tuple[float, ...],
)

Read-only live state handed to a twin-confidence source.

The simulation core does not know how a deployment acquires observed twin data. A configured source receives this context, computes or retrieves a :class:~scpn_phase_orchestrator.monitor.twin_confidence.TwinConfidenceScore, and the conformal gate uses that score to admit or reject the current policy tick.

Attributes

spec_name : str Binding-spec name being simulated. step : int Zero-based simulation step for the policy tick. sample_period_s : float Integration sample period in seconds. model_phases : FloatArray Current model phase vector after the integration step. r_good, r_bad : float Objective-layer order parameters for the current model state. regime : str Current supervisor regime label. boundary_violation_count : int Number of boundary violations on the current step. proposed_action_count : int Number of projected actions proposed before conformal admission. layer_order_parameters : tuple[float, ...] Per-layer order parameters for the current model state.

Functions:

petri_net_from_protocol

petri_net_from_protocol(
    protocol: ProtocolNetSpec,
) -> tuple[PetriNet, Marking]

Build a Petri net and initial marking from a protocol-net spec.

Parameters

protocol : ProtocolNetSpec The protocol-net specification.

Returns

tuple[PetriNet, Marking] The Petri net and its initial marking.

Source code in src/scpn_phase_orchestrator/runtime/simulation.py

def petri_net_from_protocol(protocol: ProtocolNetSpec) -> tuple[PetriNet, Marking]:
    """Build a Petri net and initial marking from a protocol-net spec.

    Parameters
    ----------
    protocol : ProtocolNetSpec
        The protocol-net specification.

    Returns
    -------
    tuple[PetriNet, Marking]
        The Petri net and its initial marking.
    """
    places = [Place(name) for name in protocol.places]
    transitions = []
    for ts in protocol.transitions:
        guard = parse_guard(ts.guard) if ts.guard else None
        transitions.append(
            Transition(
                name=ts.name,
                inputs=[Arc(a["place"], a.get("weight", 1)) for a in ts.inputs],
                outputs=[Arc(a["place"], a.get("weight", 1)) for a in ts.outputs],
                guard=guard,
            )
        )
    return PetriNet(places, transitions), Marking(tokens=dict(protocol.initial))

simulate

simulate(
    spec: BindingSpec,
    *,
    steps: int = 100,
    seed: int = 42,
    policy_enabled: bool = True,
    control_mode: SimulationControlMode = "supervisor_policy",
    audit_logger: AuditLogger | None = None,
    strict_audit_integrity: bool = True,
    binding_spec_path: Path | None = None,
    scenario_hook: ScenarioCallback | None = None,
    conformal_gate: TwinConformalGate | None = None,
    twin_confidence_source: TwinConfidenceSource
    | None = None,
) -> SimulationResult

Advance a binding spec for steps and return the simulation outcome.

Parameters

spec : BindingSpec A validated binding spec. steps : int Number of integration steps. seed : int RNG seed for the initial phases. policy_enabled : bool Closed-loop control feedback on (True) or open-loop baseline (False). control_mode : {"supervisor_policy"} The live control surface used by the generic binding-spec simulator. Koopman MPC is intentionally not a selectable simulate mode; it remains a review-only/offline proposal surface in :mod:scpn_phase_orchestrator.runtime.dvoc_oscillation_damping. audit_logger : AuditLogger | None Optional logger; when given, the header, per-step records, and events are written. The caller owns its lifecycle (close). strict_audit_integrity : bool Fail closed on a broken audit chain. When True (default) and the audit logger owns a protobuf event stream whose close-time integrity check fails, the run raises :class:~scpn_phase_orchestrator.exceptions.AuditError instead of returning a result with a tamper-evident but unenforced audit_event_stream_integrity field. Set False to attach the failed result for inspection without failing the run. binding_spec_path : Path | None Optional path to the spec, used only to locate an adjacent policy.yaml. When None, no domainpack policy rules are loaded (the supervisor policy still runs when policy_enabled). scenario_hook : ScenarioCallback | None Optional deterministic per-step perturbation hook for benchmark and case-study scenarios. The hook can mutate phases, frequencies, coupling, zeta, or psi_target in a validated :class:SimulationScenarioContext; it cannot perform actuation. conformal_gate : TwinConformalGate | None Optional calibrated conformal twin-confidence gate. When supplied with twin_confidence_source, the gate scores every live policy tick before actions are applied. twin_confidence_source : TwinConfidenceSource | None Callable that returns a twin-confidence score for the current policy tick. It must be supplied together with conformal_gate.

Returns

SimulationResult A :class:SimulationResult.

Raises

ValueError If the spec declares no oscillators, an unsupported control mode, or an incomplete conformal-admission configuration. AuditError If strict_audit_integrity and the audit event-stream integrity check fails at close time (the recorded run cannot be trusted).

Source code in src/scpn_phase_orchestrator/runtime/simulation.py

def simulate(
    spec: BindingSpec,
    *,
    steps: int = 100,
    seed: int = 42,
    policy_enabled: bool = True,
    control_mode: SimulationControlMode = "supervisor_policy",
    audit_logger: AuditLogger | None = None,
    strict_audit_integrity: bool = True,
    binding_spec_path: Path | None = None,
    scenario_hook: ScenarioCallback | None = None,
    conformal_gate: TwinConformalGate | None = None,
    twin_confidence_source: TwinConfidenceSource | None = None,
) -> SimulationResult:
    """Advance a binding spec for ``steps`` and return the simulation outcome.

    Parameters
    ----------
    spec : BindingSpec
        A validated binding spec.
    steps : int
        Number of integration steps.
    seed : int
        RNG seed for the initial phases.
    policy_enabled : bool
        Closed-loop control feedback on (``True``) or open-loop baseline (``False``).
    control_mode : {"supervisor_policy"}
        The live control surface used by the generic binding-spec simulator.
        Koopman MPC is intentionally not a selectable ``simulate`` mode; it
        remains a review-only/offline proposal surface in
        :mod:`scpn_phase_orchestrator.runtime.dvoc_oscillation_damping`.
    audit_logger : AuditLogger | None
        Optional logger; when given, the header, per-step records, and events are
        written. The caller owns its lifecycle (close).
    strict_audit_integrity : bool
        Fail closed on a broken audit chain. When ``True`` (default) and the audit
        logger owns a protobuf event stream whose close-time integrity check fails,
        the run raises :class:`~scpn_phase_orchestrator.exceptions.AuditError`
        instead of returning a result with a tamper-evident but unenforced
        ``audit_event_stream_integrity`` field. Set ``False`` to attach the failed
        result for inspection without failing the run.
    binding_spec_path : Path | None
        Optional path to the spec, used only to locate an adjacent ``policy.yaml``. When
        ``None``, no domainpack policy rules are loaded (the supervisor policy still
        runs when ``policy_enabled``).
    scenario_hook : ScenarioCallback | None
        Optional deterministic per-step perturbation hook for benchmark and case-study
        scenarios. The hook can mutate phases, frequencies, coupling, ``zeta``, or
        ``psi_target`` in a validated :class:`SimulationScenarioContext`; it cannot
        perform actuation.
    conformal_gate : TwinConformalGate | None
        Optional calibrated conformal twin-confidence gate. When supplied with
        ``twin_confidence_source``, the gate scores every live policy tick before
        actions are applied.
    twin_confidence_source : TwinConfidenceSource | None
        Callable that returns a twin-confidence score for the current policy tick.
        It must be supplied together with ``conformal_gate``.

    Returns
    -------
    SimulationResult
        A :class:`SimulationResult`.

    Raises
    ------
    ValueError
        If the spec declares no oscillators, an unsupported control mode, or an
        incomplete conformal-admission configuration.
    AuditError
        If ``strict_audit_integrity`` and the audit event-stream integrity check
        fails at close time (the recorded run cannot be trusted).
    """
    if control_mode != "supervisor_policy":
        raise ValueError(
            "simulate control_mode must be 'supervisor_policy'; Koopman MPC is "
            "offline/review-only via runtime.dvoc_oscillation_damping"
        )
    if (conformal_gate is None) != (twin_confidence_source is None):
        raise ValueError(
            "conformal_gate and twin_confidence_source must be supplied together"
        )

    n_osc = sum(len(layer.oscillator_ids) for layer in spec.layers)
    if n_osc == 0:
        raise ValueError("no oscillators defined in layers")

    binding_summary = resolved_binding_config(spec)
    builder = CouplingBuilder()
    amplitude_mode = spec.amplitude is not None
    sl_engine: StuartLandauEngine | None = None
    upde_engine: UPDEEngine | None = None
    mu: FloatArray | None = None

    if amplitude_mode:
        amp = spec.amplitude
        assert amp is not None  # nosec B101
        coupling = builder.build_with_amplitude(
            n_osc,
            spec.coupling.base_strength,
            spec.coupling.decay_alpha,
            amp.amp_coupling_strength,
            amp.amp_coupling_decay,
        )
        sl_engine = StuartLandauEngine(n_osc, dt=spec.sample_period_s)
        mu = np.full(n_osc, amp.mu)
    else:
        coupling = builder.build(
            n_osc,
            spec.coupling.base_strength,
            spec.coupling.decay_alpha,
        )
        upde_engine = UPDEEngine(n_osc, dt=spec.sample_period_s)

    event_bus = EventBus()
    boundary_observer = BoundaryObserver(spec.boundaries)
    boundary_observer.set_event_bus(event_bus)
    regime_manager = RegimeManager(event_bus=event_bus)
    channel_runtime = ChannelRuntimeExecutor.from_spec(spec)

    petri_adapter: PetriNetAdapter | None = None
    if spec.protocol_net is not None:
        net, marking = petri_net_from_protocol(spec.protocol_net)
        petri_adapter = PetriNetAdapter(
            net,
            marking,
            spec.protocol_net.place_regime,
            event_bus=event_bus,
        )

    supervisor = SupervisorPolicy(regime_manager, petri_adapter=petri_adapter)
    projector = ActionProjector.from_actuator_mappings(spec.actuators)
    prev_values: dict[str, float] = {"K": 0.0, "zeta": 0.0, "alpha": 0.0, "Psi": 0.0}

    policy_engine: PolicyEngine | None = None
    if binding_spec_path is not None:
        policy_path = binding_spec_path.parent / "policy.yaml"
        if policy_path.exists():
            rules = load_policy_rules(policy_path)
            if rules:
                policy_engine = PolicyEngine(rules)

    imprint_model: ImprintModel | None = None
    imprint_state: ImprintState | None = None
    if spec.imprint_model is not None:
        imprint_model = ImprintModel(
            spec.imprint_model.decay_rate, spec.imprint_model.saturation
        )
        imprint_state = ImprintState(m_k=np.zeros(n_osc), last_update=0.0)

    geo_constraints: list[GeometryConstraint] = []
    if spec.geometry_prior is not None:
        ct = spec.geometry_prior.constraint_type.lower()
        if "symmetric" in ct:
            geo_constraints.append(SymmetryConstraint())
        if "non_negative" in ct or "nonneg" in ct:
            geo_constraints.append(NonNegativeConstraint())

    rng = np.random.default_rng(seed)
    phases = rng.uniform(0, 2 * np.pi, n_osc)
    omegas = np.array(spec.get_omegas(), dtype=np.float64)

    amplitudes = np.array([], dtype=np.float64)
    input_amplitudes = np.array([], dtype=np.float64)
    sl_state = np.array([], dtype=np.float64)
    phases_history: list[FloatArray] = []
    amps_history: list[FloatArray] = []
    if amplitude_mode and mu is not None:
        r_init = np.sqrt(np.maximum(mu, 0.0))
        sl_state = np.concatenate([phases, r_init])

    layer_osc_ranges: dict[int, list[int]] = {}
    osc_idx = 0
    for layer in spec.layers:
        n_layer = len(layer.oscillator_ids)
        layer_osc_ranges[layer.index] = list(range(osc_idx, osc_idx + n_layer))
        osc_idx += n_layer

    zeta = max(
        (cfg.get("zeta", 0.0) for cfg in spec.drivers.all_channel_configs().values()),
        default=0.0,
    )
    zeta_ttl = 0
    psi_target = spec.drivers.physical.get("psi", 0.0)

    psi_driver: PhysicalDriver | InformationalDriver | SymbolicDriver | None = None
    if "frequency" in spec.drivers.physical:
        psi_driver = PhysicalDriver(
            frequency=spec.drivers.physical["frequency"],
            amplitude=spec.drivers.physical.get("amplitude", 1.0),
        )
    elif "cadence_hz" in spec.drivers.informational:
        psi_driver = InformationalDriver(
            cadence_hz=spec.drivers.informational["cadence_hz"],
        )
    elif "sequence" in spec.drivers.symbolic:
        psi_driver = SymbolicDriver(
            sequence=spec.drivers.symbolic["sequence"],
        )

    control_interval = max(1, round(spec.control_period_s / spec.sample_period_s))

    if audit_logger is not None:
        audit_logger.log_header(
            n_oscillators=n_osc,
            dt=spec.sample_period_s,
            seed=seed,
            amplitude_mode=amplitude_mode,
            binding_config=binding_summary,
            control_mode=control_mode,
        )

    r_good_history: list[float] = []
    r_bad_history: list[float] = []
    boundary_violation_total = 0
    action_total = 0
    conformal_decisions: list[tuple[int, ConformalDecision]] = []
    eff_mu = mu

    for step_idx in range(steps):
        if zeta_ttl > 0:
            zeta_ttl -= 1
            if zeta_ttl == 0:
                zeta = 0.0

        if psi_driver is not None:
            t = step_idx * spec.sample_period_s
            if isinstance(psi_driver, SymbolicDriver):
                psi_target = psi_driver.compute(step_idx)
            else:
                psi_target = psi_driver.compute(t)

        if scenario_hook is not None:
            context = SimulationScenarioContext(
                spec_name=spec.name,
                step=step_idx,
                sample_period_s=spec.sample_period_s,
                phases=phases,
                omegas=omegas,
                coupling=coupling,
                zeta=zeta,
                psi_target=psi_target,
                layer_osc_ranges=layer_osc_ranges,
                rng=rng,
            )
            scenario_hook(context)
            phases, omegas, coupling, zeta, psi_target = _apply_scenario_context(
                context,
                n_osc=n_osc,
            )

        eff_knm = coupling.knm
        eff_alpha = coupling.alpha
        if imprint_model is not None and imprint_state is not None:
            eff_knm = imprint_model.modulate_coupling(eff_knm, imprint_state)
            eff_alpha = imprint_model.modulate_lag(eff_alpha, imprint_state)
        if geo_constraints:
            eff_knm = project_knm(eff_knm, geo_constraints)

        input_phases = phases.copy()
        logged_zeta = zeta
        logged_psi = psi_target
        if amplitude_mode and sl_engine is not None and mu is not None:
            assert coupling.knm_r is not None  # nosec B101
            eff_mu = mu
            if imprint_model is not None and imprint_state is not None:
                eff_mu = imprint_model.modulate_mu(mu, imprint_state)
            # Capture the pre-step amplitudes so the audit record pairs them with
            # the pre-step phases. Logging post-step amplitudes alongside pre-step
            # phases yields a mixed state that no single step produced, which is
            # what broke `spo replay --verify` for amplitude-driven specs.
            input_amplitudes = sl_state[n_osc:].copy()
            sl_state = sl_engine.step(
                sl_state,
                omegas,
                eff_mu,
                eff_knm,
                coupling.knm_r,
                zeta,
                psi_target,
                eff_alpha,
                epsilon=spec.amplitude.epsilon,  # type: ignore[union-attr]  # amplitude_mode guard guarantees spec.amplitude is not None
            )
            phases = sl_state[:n_osc]
            amplitudes = sl_state[n_osc:]
            phases_history.append(phases.copy())
            amps_history.append(amplitudes.copy())
        else:
            assert upde_engine is not None  # nosec B101
            phases = upde_engine.step(
                phases, omegas, eff_knm, zeta, psi_target, eff_alpha
            )

        layer_states = []
        for layer in spec.layers:
            osc_ids = layer_osc_ranges[layer.index]
            if osc_ids:
                r, psi_l = compute_order_parameter(phases[osc_ids])
            else:
                r, psi_l = 0.0, 0.0
            ls_kwargs: dict[str, Any] = {"R": r, "psi": psi_l}
            if amplitude_mode:
                layer_r = amplitudes[osc_ids] if osc_ids else np.array([])
                if layer_r.size > 0:
                    ls_kwargs["mean_amplitude"] = float(np.mean(layer_r))
                    mean_r = float(np.mean(layer_r))
                    if mean_r > 0:
                        ls_kwargs["amplitude_spread"] = float(np.std(layer_r) / mean_r)
            layer_states.append(LayerState(**ls_kwargs))

        n_layers = len(spec.layers)
        cla = np.zeros((n_layers, n_layers))
        for li in range(n_layers):
            for lj in range(li + 1, n_layers):
                ids_i = layer_osc_ranges[spec.layers[li].index]
                ids_j = layer_osc_ranges[spec.layers[lj].index]
                if ids_i and ids_j:
                    pi, pj = phases[ids_i], phases[ids_j]
                    min_len = min(len(pi), len(pj))
                    plv = compute_plv(pi[:min_len], pj[:min_len])
                    cla[li, lj] = plv
                    cla[lj, li] = plv

        runtime_execution = channel_runtime.execute(layer_states)
        executed_layer_states = list(runtime_execution.layers)

        mean_r_val = (
            float(np.mean([ls.R for ls in executed_layer_states]))
            if executed_layer_states
            else 0.0
        )
        state_kwargs: dict[str, Any] = {
            "layers": executed_layer_states,
            "cross_layer_alignment": cla,
            "stability_proxy": mean_r_val,
            "regime_id": regime_manager.current_regime.value,
        }
        if amplitude_mode:
            state_kwargs["mean_amplitude"] = float(np.mean(amplitudes))
            sub_count = int(np.sum(amplitudes < 0.1))
            state_kwargs["subcritical_fraction"] = (
                sub_count / n_osc if n_osc > 0 else 0.0
            )
            if len(phases_history) >= 20:
                recent_ph = np.array(phases_history[-20:])
                recent_am = np.array(amps_history[-20:])
                pac_vals = [
                    modulation_index(recent_ph[:, i], recent_am[:, i])
                    for i in range(n_osc)
                ]
                state_kwargs["pac_max"] = float(max(pac_vals))

        if imprint_state is not None:
            state_kwargs["imprint_mean"] = float(np.mean(imprint_state.m_k))

        obs_values: dict[str, float] = {"R": state_kwargs["stability_proxy"]}
        if amplitude_mode:
            obs_values["mean_amplitude"] = state_kwargs.get("mean_amplitude", 0.0)
            obs_values["pac_max"] = state_kwargs.get("pac_max", 0.0)
            obs_values["subcritical_fraction"] = state_kwargs.get(
                "subcritical_fraction", 0.0
            )
        for i, ls in enumerate(executed_layer_states):
            obs_values[f"R_{i}"] = ls.R
        boundary_state = boundary_observer.observe(obs_values, step=step_idx)
        state_kwargs["boundary_violation_count"] = len(boundary_state.violations)
        boundary_violation_total += len(boundary_state.violations)
        upde_state = UPDEState(**state_kwargs)

        actions: list[Any] = []
        if policy_enabled and step_idx % control_interval == 0:
            actions = supervisor.decide(
                upde_state, boundary_state, petri_ctx=obs_values
            )
            if policy_engine is not None:
                actions.extend(
                    policy_engine.evaluate(
                        regime_manager.current_regime,
                        upde_state,
                        spec.objectives.good_layers,
                        spec.objectives.bad_layers,
                    )
                )
            actions = [
                projector.project(a, prev_values.get(a.knob, 0.0)) for a in actions
            ]
            if conformal_gate is not None and twin_confidence_source is not None:
                confidence_context = SimulationTwinConfidenceContext(
                    spec_name=spec.name,
                    step=step_idx,
                    sample_period_s=spec.sample_period_s,
                    model_phases=phases.copy(),
                    r_good=_objective_r(
                        phases,
                        spec.objectives.good_layers,
                        layer_osc_ranges,
                    ),
                    r_bad=_objective_r(
                        phases,
                        spec.objectives.bad_layers,
                        layer_osc_ranges,
                    ),
                    regime=regime_manager.current_regime.value,
                    boundary_violation_count=len(boundary_state.violations),
                    proposed_action_count=len(actions),
                    layer_order_parameters=tuple(
                        float(layer_state.R) for layer_state in executed_layer_states
                    ),
                )
                confidence_score = twin_confidence_source(confidence_context)
                if not isinstance(confidence_score, TwinConfidenceScore):
                    raise ValueError(
                        "twin_confidence_source must return TwinConfidenceScore"
                    )
                decision = conformal_gate.update(
                    confidence_nonconformity(confidence_score),
                    regime=confidence_context.regime,
                )
                conformal_decisions.append((step_idx, decision))
                if not decision.admitted:
                    actions = []
        action_total += len(actions)

        for act in actions:
            if act.knob == "zeta":
                zeta = max(0.0, min(zeta + act.value, 0.5))
                zeta_ttl = int(act.ttl_s / spec.sample_period_s)
            elif act.knob == "K":
                if act.scope == "global":
                    coupling = CouplingState(
                        knm=coupling.knm * (1.0 + act.value),
                        alpha=coupling.alpha,
                        active_template=coupling.active_template,
                        knm_r=coupling.knm_r,
                    )
                elif act.scope.startswith("layer_"):
                    idx = int(act.scope.split("_", 1)[1])
                    new_knm = coupling.knm.copy()
                    new_knm[idx, :] *= 1.0 + act.value
                    new_knm[:, idx] *= 1.0 + act.value
                    new_knm[idx, idx] = 0.0
                    coupling = CouplingState(
                        knm=new_knm,
                        alpha=coupling.alpha,
                        active_template=coupling.active_template,
                        knm_r=coupling.knm_r,
                    )
            elif act.knob == "Psi":
                psi_target = act.value
            prev_values[act.knob] = act.value

        if imprint_model is not None and imprint_state is not None:
            exposure = np.array(
                [
                    layer_states[i].R
                    for i, layer in enumerate(spec.layers)
                    for _ in layer.oscillator_ids
                ]
            )
            imprint_state = imprint_model.update(
                imprint_state, exposure, spec.sample_period_s
            )

        if audit_logger is not None:
            log_kwargs: dict[str, Any] = {
                "phases": input_phases,
                "omegas": omegas,
                "knm": eff_knm,
                "alpha": eff_alpha,
                "zeta": logged_zeta,
                "psi_drive": logged_psi,
                "channel_runtime": runtime_execution.to_audit_record(),
            }
            if amplitude_mode:
                log_kwargs["amplitudes"] = input_amplitudes
                log_kwargs["mu"] = eff_mu
                log_kwargs["knm_r"] = coupling.knm_r
                log_kwargs["epsilon"] = spec.amplitude.epsilon  # type: ignore[union-attr]  # amplitude_mode guard guarantees spec.amplitude is not None
            audit_logger.log_step(step_idx, upde_state, actions, **log_kwargs)

        r_good_history.append(
            _objective_r(phases, spec.objectives.good_layers, layer_osc_ranges)
        )
        r_bad_history.append(
            _objective_r(phases, spec.objectives.bad_layers, layer_osc_ranges)
        )

    if audit_logger is not None:
        for step_idx, decision in conformal_decisions:
            audit_logger.log_event(
                "conformal_admission",
                {"step": step_idx, "decision": decision.to_audit_record()},
            )
        for evt in event_bus.history:
            audit_logger.log_event(evt.kind, {"step": evt.step, "detail": evt.detail})
        audit_event_stream_integrity = audit_logger.verify_event_stream_integrity()
        if (
            strict_audit_integrity
            and audit_event_stream_integrity is not None
            and not audit_event_stream_integrity.ok
        ):
            raise AuditError(
                "audit event-stream integrity verification failed for "
                f"{audit_event_stream_integrity.event_stream_path!r} after "
                f"{audit_event_stream_integrity.verified_events} verified event(s); "
                "the recorded run cannot be trusted. Pass "
                "strict_audit_integrity=False to attach the failed result instead "
                "of failing closed."
            )
    else:
        audit_event_stream_integrity = None

    r_good = r_good_history[-1] if r_good_history else 0.0
    r_bad = r_bad_history[-1] if r_bad_history else 0.0
    mean_amplitude = (
        float(np.mean(amplitudes)) if amplitude_mode and amplitudes.size > 0 else None
    )

    return SimulationResult(
        spec_name=spec.name,
        steps=steps,
        policy_enabled=policy_enabled,
        control_mode=control_mode,
        amplitude_mode=amplitude_mode,
        final_phases=phases,
        final_amplitudes=amplitudes if amplitude_mode else None,
        r_good=r_good,
        r_bad=r_bad,
        separation=r_good - r_bad,
        final_regime=regime_manager.current_regime.value,
        mean_amplitude=mean_amplitude,
        r_good_history=tuple(r_good_history),
        r_bad_history=tuple(r_bad_history),
        boundary_violation_total=boundary_violation_total,
        action_total=action_total,
        conformal_admission_total=len(conformal_decisions),
        conformal_admission_rejections=sum(
            1 for _, decision in conformal_decisions if not decision.admitted
        ),
        last_conformal_admission=(
            conformal_decisions[-1][1] if conformal_decisions else None
        ),
        audit_event_stream_integrity=audit_event_stream_integrity,
    )

Network security helpers

Shared helpers for production-mode detection, environment integer parsing, and per-identity fixed-window rate limiting.

network_security

Small network-service security helpers shared by optional HTTP surfaces.

The module provides production-mode environment detection, validated non-negative integer environment parsing, and a thread-safe per-identity token-bucket rate limiter. Helpers are intentionally local and dependency-free: they do not configure servers, store credentials, or perform authentication by themselves.

Classes

TokenBucketRateLimiter

TokenBucketRateLimiter(
    limit_per_minute: int,
    *,
    burst_capacity: int | None = None,
)

Thread-safe per-identity token-bucket rate limiter.

Source code in src/scpn_phase_orchestrator/runtime/network_security.py

def __init__(
    self,
    limit_per_minute: int,
    *,
    burst_capacity: int | None = None,
) -> None:
    if not isinstance(limit_per_minute, int) or isinstance(limit_per_minute, bool):
        raise ValueError("limit_per_minute must be an integer >= 1")
    if limit_per_minute < 1:
        raise ValueError("limit_per_minute must be >= 1")
    if burst_capacity is None:
        burst = max(1, min(limit_per_minute, ceil(limit_per_minute / 10)))
    else:
        if not isinstance(burst_capacity, int) or isinstance(burst_capacity, bool):
            raise ValueError("burst_capacity must be an integer >= 1")
        if burst_capacity < 1:
            raise ValueError("burst_capacity must be >= 1")
        if burst_capacity > limit_per_minute:
            raise ValueError("burst_capacity must be <= limit_per_minute")
        burst = burst_capacity
    self._limit = limit_per_minute
    self._capacity = burst
    self._refill_per_second = limit_per_minute / 60.0
    self._lock = threading.Lock()
    self._buckets: dict[str, tuple[float, float]] = {}

Methods:

allow

allow(identity: str, now: float | None = None) -> bool

Return True if identity has at least one available token.

Parameters

identity : str Caller identity for rate limiting. now : float | None Current time in seconds, or None.

Returns

bool True when the identity has an available token.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/runtime/network_security.py

def allow(self, identity: str, now: float | None = None) -> bool:
    """Return True if *identity* has at least one available token.

    Parameters
    ----------
    identity : str
        Caller identity for rate limiting.
    now : float | None
        Current time in seconds, or ``None``.

    Returns
    -------
    bool
        ``True`` when the identity has an available token.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    key = _validated_identifier(identity, "identity")
    if now is None:
        timestamp = time.time()
    else:
        if (
            not isinstance(now, Real)
            or isinstance(now, bool)
            or not isfinite(float(now))
        ):
            raise ValueError("now must be a finite real timestamp")
        timestamp = float(now)
    with self._lock:
        tokens, updated_at = self._buckets.get(
            key, (float(self._capacity), timestamp)
        )
        elapsed = max(0.0, timestamp - updated_at)
        tokens = min(
            float(self._capacity), tokens + elapsed * self._refill_per_second
        )
        if tokens < 1.0:
            self._buckets[key] = (tokens, timestamp)
            return False
        self._buckets[key] = (tokens - 1.0, timestamp)
        return True

FixedWindowRateLimiter

FixedWindowRateLimiter(
    limit_per_minute: int,
    *,
    burst_capacity: int | None = None,
)

Bases: TokenBucketRateLimiter

Backward-compatible name for the production token-bucket limiter.

Source code in src/scpn_phase_orchestrator/runtime/network_security.py

def __init__(
    self,
    limit_per_minute: int,
    *,
    burst_capacity: int | None = None,
) -> None:
    if not isinstance(limit_per_minute, int) or isinstance(limit_per_minute, bool):
        raise ValueError("limit_per_minute must be an integer >= 1")
    if limit_per_minute < 1:
        raise ValueError("limit_per_minute must be >= 1")
    if burst_capacity is None:
        burst = max(1, min(limit_per_minute, ceil(limit_per_minute / 10)))
    else:
        if not isinstance(burst_capacity, int) or isinstance(burst_capacity, bool):
            raise ValueError("burst_capacity must be an integer >= 1")
        if burst_capacity < 1:
            raise ValueError("burst_capacity must be >= 1")
        if burst_capacity > limit_per_minute:
            raise ValueError("burst_capacity must be <= limit_per_minute")
        burst = burst_capacity
    self._limit = limit_per_minute
    self._capacity = burst
    self._refill_per_second = limit_per_minute / 60.0
    self._lock = threading.Lock()
    self._buckets: dict[str, tuple[float, float]] = {}

Functions:

is_production_mode

is_production_mode(prefix: str) -> bool

Return True when a service-specific or generic env profile is production.

Parameters

prefix : str Service-specific environment-variable prefix.

Returns

bool True when the environment profile is production.

Source code in src/scpn_phase_orchestrator/runtime/network_security.py

def is_production_mode(prefix: str) -> bool:
    """Return True when a service-specific or generic env profile is production.

    Parameters
    ----------
    prefix : str
        Service-specific environment-variable prefix.

    Returns
    -------
    bool
        ``True`` when the environment profile is production.
    """
    prefix = _validated_identifier(prefix, "prefix")
    for key in (f"{prefix}_ENV", f"{prefix}_PROFILE", "SPO_ENV", "SPO_PROFILE"):
        if os.environ.get(key, "").strip().lower() == "production":
            return True
    return False

env_int

env_int(name: str, default: int) -> int

Read a non-negative integer from the environment.

Parameters

name : str The span or resource name. default : int Default value when the variable is unset.

Returns

int The non-negative integer read from the environment.

Raises

ValueError If the inputs are invalid or inconsistent.

Source code in src/scpn_phase_orchestrator/runtime/network_security.py

def env_int(name: str, default: int) -> int:
    """Read a non-negative integer from the environment.

    Parameters
    ----------
    name : str
        The span or resource name.
    default : int
        Default value when the variable is unset.

    Returns
    -------
    int
        The non-negative integer read from the environment.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    name = _validated_identifier(name, "name")
    if not isinstance(default, int) or isinstance(default, bool) or default < 0:
        raise ValueError(f"{name} default must be a non-negative integer")
    raw = os.environ.get(name)
    if raw is None or raw == "":
        return default
    try:
        value = int(raw)
    except ValueError as exc:
        raise ValueError(f"{name} must be an integer") from exc
    if value < 0:
        raise ValueError(f"{name} must be non-negative")
    return value

Runtime observability

Prometheus text metrics are a default Runtime/Serving surface, not an optional external adapter. OpenTelemetry remains an optional backend; absent OTel packages produce validated no-op spans while /api/metrics and local Prometheus text export stay active.

observability

Default runtime observability for SPO serving surfaces.

The runtime layer always exposes deterministic Prometheus text metrics and a validated OpenTelemetry-compatible export surface. OpenTelemetry remains an optional backend dependency; when it is absent, traces and metrics become validated no-ops rather than disabling runtime observability.

Classes

RuntimeMetricSnapshot dataclass

RuntimeMetricSnapshot(
    upde_state: UPDEState,
    regime: str,
    latency_ms: float,
    step_idx: int | None = None,
)

Validated runtime metrics emitted by HTTP, gRPC, and CLI surfaces.

PrometheusEvidenceSource

Protocol-like base for audit-record evidence exported as metrics.

Methods:

to_audit_record

to_audit_record() -> Mapping[str, object]

Return JSON-safe evidence fields for Prometheus exposition.

Returns

Mapping[str, object] Return JSON-safe evidence fields for Prometheus exposition.

Raises

NotImplementedError If the operation is not implemented.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def to_audit_record(self) -> Mapping[str, object]:
    """Return JSON-safe evidence fields for Prometheus exposition.

    Returns
    -------
    Mapping[str, object]
        Return JSON-safe evidence fields for Prometheus exposition.

    Raises
    ------
    NotImplementedError
        If the operation is not implemented.
    """
    raise NotImplementedError

MetricsExporter

MetricsExporter(prefix: str = 'spo')

Format UPDE state, regime, and latency as Prometheus text exposition.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def __init__(self, prefix: str = "spo") -> None:
    if not isinstance(prefix, str) or not _PROMETHEUS_PREFIX_RE.fullmatch(prefix):
        raise ValueError(
            "prefix must be a valid Prometheus metric prefix "
            "([A-Za-z_:][A-Za-z0-9_:]*)"
        )
    self._prefix = prefix

Methods:

exposition_lines

exposition_lines(
    upde_state: UPDEState,
    regime: str,
    latency_ms: float,
    *,
    step_idx: int | None = None,
) -> list[str]

Build individual metric lines in Prometheus text format.

Parameters

upde_state : UPDEState The UPDE state to record or export. regime : str The current control regime label. latency_ms : float Step latency in milliseconds. step_idx : int | None Zero-based simulation step index, or None.

Returns

list[str] The individual Prometheus metric lines.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def exposition_lines(
    self,
    upde_state: UPDEState,
    regime: str,
    latency_ms: float,
    *,
    step_idx: int | None = None,
) -> list[str]:
    """Build individual metric lines in Prometheus text format.

    Parameters
    ----------
    upde_state : UPDEState
        The UPDE state to record or export.
    regime : str
        The current control regime label.
    latency_ms : float
        Step latency in milliseconds.
    step_idx : int | None
        Zero-based simulation step index, or ``None``.

    Returns
    -------
    list[str]
        The individual Prometheus metric lines.
    """
    p = self._prefix
    regime_label = _escape_label_value(
        _validated_regime(regime, allow_control=True)
    )
    latency_ms = _validated_latency_ms(latency_ms)
    lines: list[str] = []

    r_values = [
        _validated_finite_metric(layer.R, field=f"layer {idx} R")
        for idx, layer in enumerate(upde_state.layers)
    ]
    r_global = sum(r_values) / len(r_values) if r_values else 0.0
    stability_proxy = _validated_finite_metric(
        upde_state.stability_proxy,
        field="stability_proxy",
    )
    pac_max = _validated_finite_metric(upde_state.pac_max, field="pac_max")

    lines.append(f"# HELP {p}_r_global Global Kuramoto order parameter R")
    lines.append(f"# TYPE {p}_r_global gauge")
    lines.append(f'{p}_r_global{{regime="{regime_label}"}} {r_global:.6f}')

    lines.append(f"# HELP {p}_stability_proxy Mean R across layers")
    lines.append(f"# TYPE {p}_stability_proxy gauge")
    lines.append(
        f'{p}_stability_proxy{{regime="{regime_label}"}} {stability_proxy:.6f}'
    )

    lines.append(f"# HELP {p}_pac_max Maximum phase-amplitude coupling")
    lines.append(f"# TYPE {p}_pac_max gauge")
    lines.append(f'{p}_pac_max{{regime="{regime_label}"}} {pac_max:.6f}')

    lines.append(f"# HELP {p}_latency_ms UPDE step latency in milliseconds")
    lines.append(f"# TYPE {p}_latency_ms gauge")
    lines.append(f'{p}_latency_ms{{regime="{regime_label}"}} {latency_ms:.3f}')

    lines.append(f"# HELP {p}_layer_count Number of active UPDE layers")
    lines.append(f"# TYPE {p}_layer_count gauge")
    lines.append(f"{p}_layer_count {len(upde_state.layers)}")

    if step_idx is not None:
        step_idx = _validated_step_idx(step_idx)
        lines.append(f"# HELP {p}_step Current runtime step index")
        lines.append(f"# TYPE {p}_step gauge")
        lines.append(f"{p}_step {step_idx}")

    for i, r_value in enumerate(r_values):
        lines.append(
            f'{p}_layer_r{{layer="{i}",regime="{regime_label}"}} {r_value:.6f}'
        )

    return lines

export

export(
    upde_state: UPDEState,
    regime: str,
    latency_ms: float,
    *,
    step_idx: int | None = None,
) -> str

Return full Prometheus text exposition as a single string.

Parameters

upde_state : UPDEState The UPDE state to record or export. regime : str The current control regime label. latency_ms : float Step latency in milliseconds. step_idx : int | None Zero-based simulation step index, or None.

Returns

str The full Prometheus text exposition.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def export(
    self,
    upde_state: UPDEState,
    regime: str,
    latency_ms: float,
    *,
    step_idx: int | None = None,
) -> str:
    """Return full Prometheus text exposition as a single string.

    Parameters
    ----------
    upde_state : UPDEState
        The UPDE state to record or export.
    regime : str
        The current control regime label.
    latency_ms : float
        Step latency in milliseconds.
    step_idx : int | None
        Zero-based simulation step index, or ``None``.

    Returns
    -------
    str
        The full Prometheus text exposition.
    """
    return (
        "\n".join(
            self.exposition_lines(
                upde_state,
                regime,
                latency_ms,
                step_idx=step_idx,
            )
        )
        + "\n"
    )

digital_twin_operator_evidence_lines

digital_twin_operator_evidence_lines(
    evidence: Mapping[str, object]
    | PrometheusEvidenceSource,
) -> list[str]

Build Prometheus lines for live or replayed digital-twin evidence.

Parameters

evidence : Mapping[str, object] | PrometheusEvidenceSource Live or replayed digital-twin operator evidence.

Returns

list[str] The Prometheus lines for the digital-twin evidence.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def digital_twin_operator_evidence_lines(
    self,
    evidence: Mapping[str, object] | PrometheusEvidenceSource,
) -> list[str]:
    """Build Prometheus lines for live or replayed digital-twin evidence.

    Parameters
    ----------
    evidence : Mapping[str, object] | PrometheusEvidenceSource
        Live or replayed digital-twin operator evidence.

    Returns
    -------
    list[str]
        The Prometheus lines for the digital-twin evidence.
    """
    record = _evidence_record(evidence)
    p = self._prefix
    contract_hash = _escape_label_value(
        _validated_contract_hash(record.get("contract_hash"))
    )
    accepted_count = _validated_non_negative_count(
        record.get("accepted_count"),
        field="accepted_count",
    )
    rejected_count = _validated_non_negative_count(
        record.get("rejected_count"),
        field="rejected_count",
    )
    adapter_count = _validated_non_negative_count(
        record.get("adapter_count"),
        field="adapter_count",
    )
    unhealthy_adapter_count = _validated_non_negative_count(
        record.get("unhealthy_adapter_count"),
        field="unhealthy_adapter_count",
    )
    latest_sequence = _validated_optional_sequence(
        record.get("latest_sequence"),
        field="latest_sequence",
    )
    max_abs_twin_residual = _validated_optional_residual(
        record.get("max_abs_twin_residual"),
        field="max_abs_twin_residual",
    )
    status = _validated_operator_status(record.get("status"))
    capability_counts = _validated_label_mapping(
        record.get("capability_counts"),
        field="capability_counts",
    )
    direction_counts = _validated_label_mapping(
        record.get("direction_counts"),
        field="direction_counts",
    )
    mismatch_reason_counts = _validated_reason_counts(
        record.get("mismatch_reasons"),
        field="mismatch_reasons",
    )

    contract_label = f'contract_hash="{contract_hash}"'
    lines = [
        (
            f"# HELP {p}_digital_twin_sync_accepted_total "
            "Accepted digital-twin sync validations"
        ),
        f"# TYPE {p}_digital_twin_sync_accepted_total counter",
        (
            f"{p}_digital_twin_sync_accepted_total{{{contract_label}}} "
            f"{accepted_count}"
        ),
        (
            f"# HELP {p}_digital_twin_sync_rejected_total "
            "Rejected digital-twin sync validations"
        ),
        f"# TYPE {p}_digital_twin_sync_rejected_total counter",
        (
            f"{p}_digital_twin_sync_rejected_total{{{contract_label}}} "
            f"{rejected_count}"
        ),
        f"# HELP {p}_digital_twin_adapter_count Digital-twin adapter count",
        f"# TYPE {p}_digital_twin_adapter_count gauge",
        f"{p}_digital_twin_adapter_count{{{contract_label}}} {adapter_count}",
        (
            f"# HELP {p}_digital_twin_unhealthy_adapter_count "
            "Digital-twin adapters failing compatibility or health review"
        ),
        f"# TYPE {p}_digital_twin_unhealthy_adapter_count gauge",
        (
            f"{p}_digital_twin_unhealthy_adapter_count{{{contract_label}}} "
            f"{unhealthy_adapter_count}"
        ),
    ]
    if latest_sequence is not None:
        lines.extend(
            [
                (
                    f"# HELP {p}_digital_twin_latest_sequence "
                    "Latest accepted digital-twin sequence"
                ),
                f"# TYPE {p}_digital_twin_latest_sequence gauge",
                (
                    f"{p}_digital_twin_latest_sequence{{{contract_label}}} "
                    f"{latest_sequence}"
                ),
            ]
        )
    if max_abs_twin_residual is not None:
        lines.extend(
            [
                (
                    f"# HELP {p}_digital_twin_max_abs_residual "
                    "Maximum absolute digital-twin residual"
                ),
                f"# TYPE {p}_digital_twin_max_abs_residual gauge",
                (
                    f"{p}_digital_twin_max_abs_residual{{{contract_label}}} "
                    f"{max_abs_twin_residual:.6f}"
                ),
            ]
        )
    lines.extend(
        [
            (
                f"# HELP {p}_digital_twin_status "
                "One-hot digital-twin operator status"
            ),
            f"# TYPE {p}_digital_twin_status gauge",
        ]
    )
    for status_value in _OPERATOR_STATUS_VALUES:
        status_label = _escape_label_value(status_value)
        lines.append(
            f"{p}_digital_twin_status{{{contract_label},"
            f'status="{status_label}"}} {int(status == status_value)}'
        )
    for capability, count in sorted(capability_counts.items()):
        capability_label = _escape_label_value(capability)
        lines.append(
            f"{p}_digital_twin_capability_count{{{contract_label},"
            f'capability="{capability_label}"}} {count}'
        )
    for direction, count in sorted(direction_counts.items()):
        direction_label = _escape_label_value(direction)
        lines.append(
            f"{p}_digital_twin_direction_count{{{contract_label},"
            f'direction="{direction_label}"}} {count}'
        )
    for reason, count in sorted(mismatch_reason_counts.items()):
        reason_label = _escape_label_value(reason)
        lines.append(
            f"{p}_digital_twin_mismatch_reason_count{{{contract_label},"
            f'reason="{reason_label}"}} {count}'
        )
    return lines

export_digital_twin_operator_evidence

export_digital_twin_operator_evidence(
    evidence: Mapping[str, object]
    | PrometheusEvidenceSource,
) -> str

Return Prometheus text for digital-twin operator evidence.

Parameters

evidence : Mapping[str, object] | PrometheusEvidenceSource Live or replayed digital-twin operator evidence.

Returns

str The Prometheus text for the digital-twin operator evidence.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def export_digital_twin_operator_evidence(
    self,
    evidence: Mapping[str, object] | PrometheusEvidenceSource,
) -> str:
    """Return Prometheus text for digital-twin operator evidence.

    Parameters
    ----------
    evidence : Mapping[str, object] | PrometheusEvidenceSource
        Live or replayed digital-twin operator evidence.

    Returns
    -------
    str
        The Prometheus text for the digital-twin operator evidence.
    """
    return "\n".join(self.digital_twin_operator_evidence_lines(evidence)) + "\n"

export_twin_confidence

export_twin_confidence(
    summary: TwinConfidenceSummary,
) -> str

Return Prometheus text for a twin-confidence operator summary.

Parameters

summary : TwinConfidenceSummary The operator-facing aggregate over scored twin-confidence ticks.

Returns

str Prometheus exposition text using this exporter's metric prefix.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def export_twin_confidence(self, summary: TwinConfidenceSummary) -> str:
    """Return Prometheus text for a twin-confidence operator summary.

    Parameters
    ----------
    summary : TwinConfidenceSummary
        The operator-facing aggregate over scored twin-confidence ticks.

    Returns
    -------
    str
        Prometheus exposition text using this exporter's metric prefix.
    """
    from scpn_phase_orchestrator.monitor.twin_confidence import (
        twin_confidence_prometheus_text,
    )

    text: str = twin_confidence_prometheus_text(summary, prefix=self._prefix)
    return text

OTelExporter

OTelExporter(service_name: str = 'spo')

Instrument UPDE steps with OpenTelemetry spans and metrics.

Falls back to validated no-op behaviour when opentelemetry-api is not installed, so runtime observability remains enabled by default without making OpenTelemetry a base dependency.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def __init__(self, service_name: str = "spo") -> None:
    if not isinstance(service_name, str) or not _SERVICE_NAME_RE.fullmatch(
        service_name
    ):
        raise ValueError(
            "service_name must start with a letter and contain only "
            "letters, digits, underscore, dot, or hyphen"
        )
    self._service_name = service_name
    self._enabled = _HAS_OTEL
    self._tracer: Any = None
    self._r_global_gauge: Any = None
    self._stability_gauge: Any = None
    self._step_counter: Any = None
    if self._enabled:  # pragma: no cover
        self._tracer = otel_trace.get_tracer(service_name)
        meter = otel_metrics.get_meter(service_name)
        self._r_global_gauge = meter.create_gauge(
            "spo.r_global",
            description="Global Kuramoto order parameter R",
            unit="1",
        )
        self._stability_gauge = meter.create_gauge(
            "spo.stability_proxy",
            description="Mean R across layers",
            unit="1",
        )
        self._step_counter = meter.create_counter(
            "spo.steps_total",
            description="Total UPDE integration steps",
            unit="1",
        )

Attributes

enabled property

enabled: bool

True when opentelemetry-api is installed and active.

Returns

bool True when opentelemetry-api is installed and active.

Methods:

span

span(
    name: str,
    attributes: Mapping[str, object] | None = None,
) -> Generator[Any, None, None]

Trace span context manager. No-op when OTel is absent.

Parameters

name : str The span or resource name. attributes : Mapping[str, object] | None Optional span attributes, or None.

Returns

Generator[Any, None, None] A trace-span context manager (no-op without OpenTelemetry).

Source code in src/scpn_phase_orchestrator/runtime/observability.py

@contextmanager
def span(
    self,
    name: str,
    attributes: Mapping[str, object] | None = None,
) -> Generator[Any, None, None]:
    """Trace span context manager. No-op when OTel is absent.

    Parameters
    ----------
    name : str
        The span or resource name.
    attributes : Mapping[str, object] | None
        Optional span attributes, or ``None``.

    Returns
    -------
    Generator[Any, None, None]
        A trace-span context manager (no-op without OpenTelemetry).
    """
    name = _validated_otel_name(name, field="span name")
    attributes = _validated_attributes(attributes)
    if not self._enabled:
        yield _NoOpSpan()
        return
    with self._tracer.start_as_current_span(name) as span:
        if attributes:
            for key, value in attributes.items():
                span.set_attribute(key, value)
        yield span

record_step

record_step(upde_state: UPDEState, step_idx: int) -> None

Record metrics from a completed UPDE step.

Parameters

upde_state : UPDEState The UPDE state to record or export. step_idx : int Zero-based simulation step index, or None.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def record_step(self, upde_state: UPDEState, step_idx: int) -> None:
    """Record metrics from a completed UPDE step.

    Parameters
    ----------
    upde_state : UPDEState
        The UPDE state to record or export.
    step_idx : int
        Zero-based simulation step index, or ``None``.
    """
    _validated_step_idx(step_idx)
    stability_proxy = _validated_finite_metric(
        upde_state.stability_proxy,
        field="stability_proxy",
    )
    regime = _validated_regime(upde_state.regime_id)
    if not self._enabled:
        return
    attrs = {"spo.regime": regime}
    self._r_global_gauge.set(stability_proxy, attrs)
    self._stability_gauge.set(stability_proxy, attrs)
    self._step_counter.add(1, attrs)

record_regime_change

record_regime_change(old: str, new: str) -> None

Emit a span event for regime transitions.

Parameters

old : str The previous regime label. new : str The new regime label.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def record_regime_change(self, old: str, new: str) -> None:
    """Emit a span event for regime transitions.

    Parameters
    ----------
    old : str
        The previous regime label.
    new : str
        The new regime label.
    """
    old = _validated_regime(old)
    new = _validated_regime(new)
    if not self._enabled:
        return
    with self._tracer.start_as_current_span("spo.regime_change") as span:
        span.set_attribute("spo.regime.old", old)
        span.set_attribute("spo.regime.new", new)

RuntimeObservability

RuntimeObservability(
    *,
    service_name: str = "spo",
    metric_prefix: str = "spo",
    otel_exporter: OTelExporter | None = None,
)

Default observability facade used by runtime serving surfaces.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def __init__(
    self,
    *,
    service_name: str = "spo",
    metric_prefix: str = "spo",
    otel_exporter: OTelExporter | None = None,
) -> None:
    self._metrics = MetricsExporter(metric_prefix)
    self._otel = (
        otel_exporter if otel_exporter is not None else OTelExporter(service_name)
    )

Attributes

otel_enabled property

otel_enabled: bool

True when OpenTelemetry is installed and active.

Returns

bool True when OpenTelemetry is installed and active.

Methods:

prometheus_text

prometheus_text(snapshot: RuntimeMetricSnapshot) -> str

Return default Prometheus text for a runtime metric snapshot.

Parameters

snapshot : RuntimeMetricSnapshot The runtime metric snapshot.

Returns

str The default Prometheus text for the snapshot.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def prometheus_text(self, snapshot: RuntimeMetricSnapshot) -> str:
    """Return default Prometheus text for a runtime metric snapshot.

    Parameters
    ----------
    snapshot : RuntimeMetricSnapshot
        The runtime metric snapshot.

    Returns
    -------
    str
        The default Prometheus text for the snapshot.
    """
    return self._metrics.export(
        snapshot.upde_state,
        snapshot.regime,
        snapshot.latency_ms,
        step_idx=snapshot.step_idx,
    )

digital_twin_prometheus_text

digital_twin_prometheus_text(
    evidence: Mapping[str, object]
    | PrometheusEvidenceSource,
) -> str

Return Prometheus text for live or replayed digital-twin evidence.

Parameters

evidence : Mapping[str, object] | PrometheusEvidenceSource Live or replayed digital-twin operator evidence.

Returns

str The Prometheus text for the digital-twin evidence.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def digital_twin_prometheus_text(
    self,
    evidence: Mapping[str, object] | PrometheusEvidenceSource,
) -> str:
    """Return Prometheus text for live or replayed digital-twin evidence.

    Parameters
    ----------
    evidence : Mapping[str, object] | PrometheusEvidenceSource
        Live or replayed digital-twin operator evidence.

    Returns
    -------
    str
        The Prometheus text for the digital-twin evidence.
    """
    return self._metrics.export_digital_twin_operator_evidence(evidence)

twin_confidence_prometheus_text

twin_confidence_prometheus_text(
    summary: TwinConfidenceSummary,
) -> str

Return Prometheus text for a twin-confidence operator summary.

Parameters

summary : TwinConfidenceSummary The operator-facing aggregate over scored twin-confidence ticks.

Returns

str Prometheus exposition text for the twin-confidence summary.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def twin_confidence_prometheus_text(self, summary: TwinConfidenceSummary) -> str:
    """Return Prometheus text for a twin-confidence operator summary.

    Parameters
    ----------
    summary : TwinConfidenceSummary
        The operator-facing aggregate over scored twin-confidence ticks.

    Returns
    -------
    str
        Prometheus exposition text for the twin-confidence summary.
    """
    return self._metrics.export_twin_confidence(summary)

record_step

record_step(snapshot: RuntimeMetricSnapshot) -> None

Record a runtime step through the optional OpenTelemetry backend.

Parameters

snapshot : RuntimeMetricSnapshot The runtime metric snapshot.

Source code in src/scpn_phase_orchestrator/runtime/observability.py

def record_step(self, snapshot: RuntimeMetricSnapshot) -> None:
    """Record a runtime step through the optional OpenTelemetry backend.

    Parameters
    ----------
    snapshot : RuntimeMetricSnapshot
        The runtime metric snapshot.
    """
    if snapshot.step_idx is None:
        return
    self._otel.record_step(snapshot.upde_state, snapshot.step_idx)

span

span(
    name: str,
    attributes: Mapping[str, object] | None = None,
) -> Generator[Any, None, None]

Return a validated runtime span, no-op without OpenTelemetry.

Parameters

name : str The span or resource name. attributes : Mapping[str, object] | None Optional span attributes, or None.

Returns

Generator[Any, None, None] A validated runtime span (no-op without OpenTelemetry).

Source code in src/scpn_phase_orchestrator/runtime/observability.py

@contextmanager
def span(
    self,
    name: str,
    attributes: Mapping[str, object] | None = None,
) -> Generator[Any, None, None]:
    """Return a validated runtime span, no-op without OpenTelemetry.

    Parameters
    ----------
    name : str
        The span or resource name.
    attributes : Mapping[str, object] | None
        Optional span attributes, or ``None``.

    Returns
    -------
    Generator[Any, None, None]
        A validated runtime span (no-op without OpenTelemetry).
    """
    with self._otel.span(name, attributes) as span:
        yield span

Web and gRPC services

The service modules expose the FastAPI dashboard state container and the gRPC servicer. Optional web or gRPC dependencies are handled at import time so documentation builds can inspect the public surface without launching servers.

server

FastAPI server for real-time simulation monitoring.

Usage

from scpn_phase_orchestrator.runtime.server import create_app app = create_app("domainpacks/minimal_domain/binding_spec.yaml")

Save that object in an ASGI module, then run it with an ASGI server such as uvicorn app:app --host 127.0.0.1 --port 8080.

Production profile requires API key authentication and rate limiting:

SPO_ENV=production SPO_API_KEY= uvicorn app:app

Mutable endpoints (step, reset) require X-API-Key header.

Development mode remains local by default; do not deploy it externally.

Endpoints

GET / HTML dashboard GET /api/state Current UPDE state (JSON) GET /api/studio-feed Live STUDIO feed envelope (JSON) POST /api/step Advance one step (auth required if SPO_API_KEY set) POST /api/reset Reset simulation (auth required if SPO_API_KEY set) GET /api/config Binding spec summary WS /ws/stream Real-time WebSocket stream (read-only observer)

Classes

SimulationState

SimulationState(spec: BindingSpec)

Mutable simulation state shared across API endpoints.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def __init__(self, spec: BindingSpec) -> None:
    self.spec = spec
    # threading.Lock so FastAPI async handlers and the gRPC servicer
    # (thread-pool worker) serialise against the *same* mutex. An
    # asyncio.Lock would only protect the event loop and leave gRPC
    # threads free to race on the shared engine state.
    self._lock = threading.Lock()
    self.n_osc = sum(len(ly.oscillator_ids) for ly in spec.layers)
    self.coupling = CouplingBuilder().build(
        self.n_osc,
        spec.coupling.base_strength,
        spec.coupling.decay_alpha,
    )
    self.omegas = np.array(spec.get_omegas(), dtype=np.float64)
    self.phases = extract_initial_phases(spec, self.omegas)
    self.engine = UPDEEngine(self.n_osc, dt=spec.sample_period_s)
    self.event_bus = EventBus()
    self.boundary_observer = BoundaryObserver(spec.boundaries)
    self.boundary_observer.set_event_bus(self.event_bus)
    self.regime_manager = RegimeManager(event_bus=self.event_bus)
    self.step_count = 0
    self.amplitude_mode = spec.amplitude is not None
    self.sl_engine: StuartLandauEngine | None = None
    self.sl_state: FloatArray | None = None
    self.mu: FloatArray | None = None

    self.imprint_model: ImprintModel | None = None
    self.imprint_state: ImprintState | None = None
    if spec.imprint_model is not None:
        self.imprint_model = ImprintModel(
            spec.imprint_model.decay_rate, spec.imprint_model.saturation
        )
        self.imprint_state = ImprintState(m_k=np.zeros(self.n_osc), last_update=0.0)

    self.geo_constraints: list[GeometryConstraint] = []
    if spec.geometry_prior is not None:
        ct = spec.geometry_prior.constraint_type.lower()
        if "symmetric" in ct:
            self.geo_constraints.append(SymmetryConstraint())
        if "non_negative" in ct or "nonneg" in ct:
            self.geo_constraints.append(NonNegativeConstraint())

    if self.amplitude_mode and spec.amplitude is not None:
        amp = spec.amplitude
        self.sl_engine = StuartLandauEngine(
            self.n_osc,
            dt=spec.sample_period_s,
        )
        self.mu = np.full(self.n_osc, amp.mu)
        self.coupling = CouplingBuilder().build_with_amplitude(
            self.n_osc,
            spec.coupling.base_strength,
            spec.coupling.decay_alpha,
            amp.amp_coupling_strength,
            amp.amp_coupling_decay,
        )
        self.sl_state = np.concatenate(
            [
                self.phases,
                np.sqrt(np.maximum(self.mu, 0.0)),
            ]
        )

Methods:

step

step() -> dict[str, Any]

Advance one timestep, return state snapshot.

Returns

dict[str, Any] Advance one timestep, return state snapshot.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def step(self) -> dict[str, Any]:
    """Advance one timestep, return state snapshot.

    Returns
    -------
    dict[str, Any]
        Advance one timestep, return state snapshot.
    """
    eff_knm = self.coupling.knm
    eff_alpha = self.coupling.alpha
    if self.imprint_model is not None and self.imprint_state is not None:
        eff_knm = self.imprint_model.modulate_coupling(eff_knm, self.imprint_state)
        eff_alpha = self.imprint_model.modulate_lag(eff_alpha, self.imprint_state)
    if self.geo_constraints:
        eff_knm = project_knm(eff_knm, self.geo_constraints)

    if (
        self.amplitude_mode
        and self.sl_engine is not None
        and self.mu is not None
        and self.sl_state is not None
        and self.coupling.knm_r is not None
        and self.spec.amplitude is not None
    ):
        self.sl_state = self.sl_engine.step(
            self.sl_state,
            self.omegas,
            self.mu,
            eff_knm,
            self.coupling.knm_r,
            0.0,
            0.0,
            eff_alpha,
            epsilon=self.spec.amplitude.epsilon,
        )
        self.phases = self.sl_state[: self.n_osc]
    else:
        self.phases = self.engine.step(
            self.phases,
            self.omegas,
            eff_knm,
            0.0,
            0.0,
            eff_alpha,
        )
    self.step_count += 1

    layer_states = []
    idx = 0
    for layer in self.spec.layers:
        n = len(layer.oscillator_ids)
        r, psi = compute_order_parameter(self.phases[idx : idx + n])
        layer_states.append(LayerState(R=r, psi=psi))
        idx += n
    if self.imprint_model is not None and self.imprint_state is not None:
        exposure = np.array(
            [
                layer_states[i].R
                for i, layer in enumerate(self.spec.layers)
                for _ in layer.oscillator_ids
            ]
        )
        self.imprint_state = self.imprint_model.update(
            self.imprint_state, exposure, self.spec.sample_period_s
        )

    r_global, _ = compute_order_parameter(self.phases)
    upde_state = UPDEState(
        layers=layer_states,
        cross_layer_alignment=np.eye(len(self.spec.layers)),
        stability_proxy=r_global,
        regime_id=self.regime_manager.current_regime.value,
    )
    obs_values: dict[str, float] = {"R": r_global}
    for i, ls in enumerate(layer_states):
        obs_values[f"R_{i}"] = ls.R
    boundary_state = self.boundary_observer.observe(
        obs_values, step=self.step_count
    )
    proposed = self.regime_manager.evaluate(upde_state, boundary_state)
    self.regime_manager.transition(proposed)

    return self.snapshot()

snapshot

snapshot() -> dict[str, Any]

Return the current state as a JSON-serialisable dict.

Returns

dict[str, Any] Return the current state as a JSON-serialisable dict.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def snapshot(self) -> dict[str, Any]:
    """Return the current state as a JSON-serialisable dict.

    Returns
    -------
    dict[str, Any]
        Return the current state as a JSON-serialisable dict.
    """
    layer_map = {}
    idx = 0
    for layer in self.spec.layers:
        n = len(layer.oscillator_ids)
        layer_map[layer.name] = list(range(idx, idx + n))
        idx += n

    layers = []
    for layer in self.spec.layers:
        ids = layer_map[layer.name]
        r, psi = compute_order_parameter(self.phases[ids])
        layers.append(
            {
                "name": layer.name,
                "R": round(float(r), 4),
                "psi": round(float(psi), 4),
            }
        )

    r_global, _ = compute_order_parameter(self.phases)
    result = {
        "step": self.step_count,
        "R_global": round(float(r_global), 4),
        "regime": self.regime_manager.current_regime.value,
        "layers": layers,
        "n_oscillators": self.n_osc,
        "amplitude_mode": self.amplitude_mode,
    }

    if self.amplitude_mode and self.sl_state is not None:
        amps = self.sl_state[self.n_osc :]
        result["mean_amplitude"] = round(float(np.mean(amps)), 4)

    return result

studio_feed

studio_feed() -> dict[str, object]

Return the live STUDIO feed for the current simulation snapshot.

Returns

dict[str, object] studio.control-feed.v1 envelope plus SPO runtime snapshot.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def studio_feed(self) -> dict[str, object]:
    """Return the live STUDIO feed for the current simulation snapshot.

    Returns
    -------
    dict[str, object]
        ``studio.control-feed.v1`` envelope plus SPO runtime snapshot.
    """
    return build_studio_control_feed(
        self.snapshot(),
        studio_version=__version__,
    )

reset

reset() -> dict[str, Any]

Reset to initial state.

Returns

dict[str, Any] Reset to initial state.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def reset(self) -> dict[str, Any]:
    """Reset to initial state.

    Returns
    -------
    dict[str, Any]
        Reset to initial state.
    """
    self.phases = extract_initial_phases(self.spec, self.omegas)
    if self.amplitude_mode and self.mu is not None:
        self.sl_state = np.concatenate(
            [
                self.phases,
                np.sqrt(np.maximum(self.mu, 0.0)),
            ]
        )
    self.step_count = 0
    self.regime_manager = RegimeManager(event_bus=self.event_bus)
    if self.imprint_model is not None:
        self.imprint_state = ImprintState(m_k=np.zeros(self.n_osc), last_update=0.0)
    return self.snapshot()

Functions:

create_app

create_app(spec_path: str | Path) -> object

Create FastAPI app for the given binding spec.

Parameters

spec_path : str | Path Filesystem path to the binding-spec file.

Returns

object The configured FastAPI application.

Raises

RuntimeError If the runtime operation fails. ImportError If a required optional dependency is not installed. HTTPException If the request is invalid.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def create_app(spec_path: str | Path) -> object:  # pragma: no cover
    """Create FastAPI app for the given binding spec.

    Parameters
    ----------
    spec_path : str | Path
        Filesystem path to the binding-spec file.

    Returns
    -------
    object
        The configured FastAPI application.

    Raises
    ------
    RuntimeError
        If the runtime operation fails.
    ImportError
        If a required optional dependency is not installed.
    HTTPException
        If the request is invalid.
    """
    try:
        from collections.abc import AsyncIterator
        from contextlib import asynccontextmanager

        from fastapi import Depends, FastAPI, Header, HTTPException
        from fastapi.responses import HTMLResponse
    except ImportError as exc:
        msg = "fastapi not installed. pip install fastapi uvicorn"
        raise ImportError(msg) from exc

    from scpn_phase_orchestrator.runtime.network_security import (
        FixedWindowRateLimiter,
        env_int,
        is_production_mode,
    )

    spec = load_binding_spec(spec_path)
    sim = SimulationState(spec)

    @asynccontextmanager
    async def _lifespan(_app: FastAPI) -> AsyncIterator[None]:
        """Release engine resources when the process shuts down.

        The simulation state itself is held in-process (numpy arrays), but
        future integrations (gRPC channels, database handles, external
        adapters) register cleanup here so a graceful shutdown never
        leaks a descriptor.
        """
        logger.info(
            "spo server startup: spec=%s n_osc=%d amplitude_mode=%s",
            spec.name,
            sim.n_osc,
            sim.amplitude_mode,
        )
        try:
            yield
        finally:
            logger.info("spo server shutdown: spec=%s", spec.name)
            with sim._lock:
                sim.event_bus.clear()

    app = FastAPI(title="SPO Dashboard", version=__version__, lifespan=_lifespan)

    @app.middleware("http")
    async def _log_http_request(
        request: FastAPIRequest,
        call_next: Callable[[FastAPIRequest], Awaitable[Response]],
    ) -> Response:
        """Log an HTTP request to the audit stream."""
        start = time.perf_counter()
        status_code = 500
        try:
            response = await call_next(request)
            status_code = response.status_code
            return response
        finally:
            duration_ms = (time.perf_counter() - start) * 1000.0
            path = request.url.path
            logger.info(
                "http.request: method=%s path=%s status_code=%d duration_ms=%.3f",
                request.method,
                path,
                status_code,
                duration_ms,
                extra={
                    "http_method": request.method,
                    "http_path": path,
                    "status_code": status_code,
                    "duration_ms": duration_ms,
                },
            )

    _api_key = os.environ.get("SPO_API_KEY")
    _production = is_production_mode("SPO")
    if _production and not _api_key:
        raise RuntimeError("SPO_API_KEY is required when SPO_ENV=production")
    _rate_limit = env_int("SPO_RATE_LIMIT_PER_MINUTE", 120 if _production else 0)
    _limiter = FixedWindowRateLimiter(_rate_limit) if _rate_limit > 0 else None

    async def _require_auth(
        request: FastAPIRequest,
        x_api_key: str | None = Header(None),
    ) -> None:
        """Authorise an HTTP request, raising on failure."""
        if _api_key is None:
            identity = request.client.host if request.client is not None else "local"
        elif x_api_key is None or not hmac.compare_digest(x_api_key, _api_key):
            raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")
        else:
            identity = x_api_key
        if _limiter is not None and not _limiter.allow(identity):
            raise HTTPException(status_code=429, detail="Rate limit exceeded")

    @app.get("/", response_class=HTMLResponse)
    async def dashboard() -> str:
        """Handle GET / — serve the HTML dashboard."""
        return DASHBOARD_HTML

    @app.get("/api/state")
    async def get_state() -> dict[str, Any]:
        """Handle GET /api/state — return current simulation snapshot."""
        with sim._lock:
            return sim.snapshot()

    @app.get("/api/studio-feed")
    async def get_studio_feed() -> dict[str, object]:
        """Handle GET /api/studio-feed — return live STUDIO feed envelope."""
        with sim._lock:
            return sim.studio_feed()

    @app.post("/api/step", dependencies=[Depends(_require_auth)])
    async def post_step() -> dict[str, Any]:
        """Handle POST /api/step — advance simulation one tick."""
        with sim._lock:
            snap = sim.step()
        logger.debug(
            "api.step: step=%d R_global=%.4f regime=%s",
            snap.get("step", -1),
            snap.get("R_global", float("nan")),
            snap.get("regime", ""),
        )
        return snap

    @app.post("/api/reset", dependencies=[Depends(_require_auth)])
    async def post_reset() -> dict[str, Any]:
        """Handle POST /api/reset — reset simulation to initial state."""
        with sim._lock:
            snap = sim.reset()
        logger.info(
            "api.reset: step=%d regime=%s",
            snap.get("step", 0),
            snap.get("regime", ""),
        )
        return snap

    @app.get("/api/config")
    async def get_config() -> dict[str, Any]:
        """Handle GET /api/config — return engine configuration."""
        return {
            "name": spec.name,
            "n_oscillators": sim.n_osc,
            "n_layers": len(spec.layers),
            "amplitude_mode": sim.amplitude_mode,
            "sample_period_s": spec.sample_period_s,
            "control_period_s": spec.control_period_s,
        }

    @app.get("/api/metrics")
    async def get_metrics() -> Response:
        """Handle GET /api/metrics — export Prometheus-format metrics."""
        from fastapi.responses import PlainTextResponse

        with sim._lock:
            snap = sim.snapshot()
        upde_state = UPDEState(
            layers=[
                LayerState(R=ly["R"], psi=ly.get("psi", 0.0)) for ly in snap["layers"]
            ],
            cross_layer_alignment=np.eye(len(snap["layers"])),
            stability_proxy=snap["R_global"],
            regime_id=snap["regime"],
        )
        observability = RuntimeObservability()
        text = observability.prometheus_text(
            RuntimeMetricSnapshot(
                upde_state=upde_state,
                regime=snap["regime"],
                latency_ms=0.0,
                step_idx=snap.get("step"),
            )
        )
        return PlainTextResponse(text, media_type="text/plain")

    @app.get("/api/health")
    async def health() -> dict[str, Any]:
        """Deep health check — verifies engine, monitor, and regime subsystems."""
        checks: dict[str, str] = {}
        try:
            with sim._lock:
                snap = sim.snapshot()
            checks["engine"] = "ok" if snap.get("step", -1) >= 0 else "degraded"
            r_val = snap.get("R_global", float("nan"))
            checks["R_finite"] = "ok" if np.isfinite(r_val) else "error"
            checks["regime"] = "ok" if snap.get("regime") else "unknown"
        except HEALTH_CHECK_EXCEPTIONS as exc:
            # Log the detail server-side; do not leak the exception text into the
            # health response (CodeQL py/stack-trace-exposure — information
            # exposure through an exception).
            logger.warning("health-check engine probe failed: %s", exc)
            checks["engine"] = "error"

        healthy = all(v == "ok" for v in checks.values())
        return {"status": "healthy" if healthy else "degraded", "checks": checks}

    @app.websocket("/ws/stream")
    async def ws_stream(websocket: WebSocket) -> None:
        """Read-only observer: streams snapshots without advancing simulation."""
        await websocket.accept()
        try:
            while True:
                with sim._lock:
                    state = sim.snapshot()
                await websocket.send_text(json.dumps(state))
                await asyncio.sleep(spec.sample_period_s)
        except WebSocketDisconnect:
            pass

    return app

server_grpc

gRPC service implementing the full PhaseOrchestrator API.

Install grpcio to run a live server::

pip install grpcio grpcio-tools

Without grpcio the servicer still works for in-process testing with a mocked context.

Classes

PhaseStreamServicer

PhaseStreamServicer(sim: SimulationState)

Bases: PhaseOrchestratorServicer

gRPC servicer that exposes state, step, reset, streaming, and config.

Wraps a SimulationState (from server.py) and translates its dict snapshots into proto-compatible StateResponse messages.

Source code in src/scpn_phase_orchestrator/runtime/server_grpc.py

def __init__(self, sim: SimulationState) -> None:
    self._sim = sim
    # Share the SimulationState mutex so gRPC threads and FastAPI async
    # handlers serialise against the same lock. A servicer-local lock
    # would leave the HTTP side free to race on shared engine state.
    self._lock = sim._lock
    self._api_key = os.environ.get("SPO_GRPC_API_KEY") or os.environ.get(
        "SPO_API_KEY"
    )
    self._production = is_production_mode("SPO_GRPC") or is_production_mode("SPO")
    if self._production and not self._api_key:
        raise RuntimeError(
            "SPO_GRPC_API_KEY or SPO_API_KEY is required in production"
        )
    rate_limit = env_int(
        "SPO_GRPC_RATE_LIMIT_PER_MINUTE", 120 if self._production else 0
    )
    self._limiter = FixedWindowRateLimiter(rate_limit) if rate_limit > 0 else None

Methods:

GetState

GetState(request: Any, context: Any) -> StateResponse

GRPC unary RPC: return current simulation state.

Parameters

request : Any The gRPC request message. context : Any The gRPC servicer context.

Returns

StateResponse The current simulation-state response.

Source code in src/scpn_phase_orchestrator/runtime/server_grpc.py

def GetState(self, request: Any, context: Any) -> StateResponse:
    """GRPC unary RPC: return current simulation state.

    Parameters
    ----------
    request : Any
        The gRPC request message.
    context : Any
        The gRPC servicer context.

    Returns
    -------
    StateResponse
        The current simulation-state response.
    """
    self._authorise(context)
    with self._lock:
        response = _snap_to_response(self._sim.snapshot())
    _log_state_rpc("GetState", response)
    return response

Step

Step(request: Any, context: Any) -> StateResponse

GRPC unary RPC: advance simulation by n_steps and return state.

Parameters

request : Any The gRPC request message. context : Any The gRPC servicer context.

Returns

StateResponse The simulation-state response after advancing.

Source code in src/scpn_phase_orchestrator/runtime/server_grpc.py

def Step(self, request: Any, context: Any) -> StateResponse:
    """GRPC unary RPC: advance simulation by n_steps and return state.

    Parameters
    ----------
    request : Any
        The gRPC request message.
    context : Any
        The gRPC servicer context.

    Returns
    -------
    StateResponse
        The simulation-state response after advancing.
    """
    self._authorise(context)
    raw_n_steps = getattr(request, "n_steps", 1)
    if raw_n_steps == 0:
        raw_n_steps = 1
    try:
        n = _validate_positive_int(raw_n_steps, "n_steps")
    except ValueError as exc:
        code = grpc.StatusCode.INVALID_ARGUMENT if grpc is not None else None
        self._abort(context, code, str(exc))
    with self._lock:
        for _ in range(n):
            self._sim.step()
        response = _snap_to_response(self._sim.snapshot())
    _log_state_rpc("Step", response, n_steps=n)
    return response

Reset

Reset(request: Any, context: Any) -> StateResponse

GRPC unary RPC: reset simulation and return fresh state.

Parameters

request : Any The gRPC request message. context : Any The gRPC servicer context.

Returns

StateResponse The fresh simulation-state response after reset.

Source code in src/scpn_phase_orchestrator/runtime/server_grpc.py

def Reset(self, request: Any, context: Any) -> StateResponse:
    """GRPC unary RPC: reset simulation and return fresh state.

    Parameters
    ----------
    request : Any
        The gRPC request message.
    context : Any
        The gRPC servicer context.

    Returns
    -------
    StateResponse
        The fresh simulation-state response after reset.
    """
    self._authorise(context)
    with self._lock:
        self._sim.reset()
        response = _snap_to_response(self._sim.snapshot())
    _log_state_rpc("Reset", response)
    return response

GetConfig

GetConfig(request: Any, context: Any) -> ConfigResponse

GRPC unary RPC: return engine configuration.

Parameters

request : Any The gRPC request message. context : Any The gRPC servicer context.

Returns

ConfigResponse The engine-configuration response.

Source code in src/scpn_phase_orchestrator/runtime/server_grpc.py

def GetConfig(self, request: Any, context: Any) -> ConfigResponse:
    """GRPC unary RPC: return engine configuration.

    Parameters
    ----------
    request : Any
        The gRPC request message.
    context : Any
        The gRPC servicer context.

    Returns
    -------
    ConfigResponse
        The engine-configuration response.
    """
    self._authorise(context)
    spec = self._sim.spec
    response = ConfigResponse(
        name=spec.name,
        n_oscillators=self._sim.n_osc,
        n_layers=len(spec.layers),
        amplitude_mode=self._sim.amplitude_mode,
        sample_period_s=spec.sample_period_s,
        control_period_s=spec.control_period_s,
    )
    logger.info(
        "grpc.GetConfig: name=%s n_oscillators=%d n_layers=%d",
        response.name,
        response.n_oscillators,
        response.n_layers,
        extra={
            "rpc": "GetConfig",
            "config_name": response.name,
            "n_oscillators": response.n_oscillators,
            "n_layers": response.n_layers,
            "status": "ok",
        },
    )
    return response

StreamPhases

StreamPhases(
    request: Any, context: Any
) -> Iterator[StateResponse]

Read-only observer: streams snapshots without advancing simulation.

Parameters

request : Any The gRPC request message. context : Any The gRPC servicer context.

Returns

Iterator[StateResponse] An iterator of state-response snapshots.

Source code in src/scpn_phase_orchestrator/runtime/server_grpc.py

def StreamPhases(self, request: Any, context: Any) -> Iterator[StateResponse]:
    """Read-only observer: streams snapshots without advancing simulation.

    Parameters
    ----------
    request : Any
        The gRPC request message.
    context : Any
        The gRPC servicer context.

    Returns
    -------
    Iterator[StateResponse]
        An iterator of state-response snapshots.
    """
    self._authorise(context)
    try:
        max_steps = _resolve_stream_max_steps(request)
        max_steps = _validate_positive_int(max_steps, "max_steps")
    except ValueError as exc:
        code = grpc.StatusCode.INVALID_ARGUMENT if grpc is not None else None
        self._abort(context, code, str(exc))
    interval = getattr(request, "interval_s", 0.05)
    try:
        interval = _validate_non_negative_real(interval, "interval_s")
    except ValueError as exc:
        code = grpc.StatusCode.INVALID_ARGUMENT if grpc is not None else None
        self._abort(context, code, str(exc))
    emitted = 0
    logger.info(
        "grpc.StreamPhases.start: max_steps=%d interval_s=%.4f",
        max_steps,
        interval,
        extra={
            "rpc": "StreamPhases",
            "stream_event": "start",
            "max_steps": max_steps,
            "interval_s": interval,
            "status": "ok",
        },
    )
    for _ in range(max_steps):
        if (
            context is not None
            and hasattr(context, "is_active")
            and not context.is_active()
        ):
            logger.info(
                "grpc.StreamPhases.stop: emitted=%d reason=inactive_context",
                emitted,
                extra={
                    "rpc": "StreamPhases",
                    "stream_event": "stop",
                    "emitted": emitted,
                    "reason": "inactive_context",
                    "status": "ok",
                },
            )
            return
        with self._lock:
            snap = self._sim.snapshot()
        yield _snap_to_response(snap)
        emitted += 1
        time.sleep(interval)
    logger.info(
        "grpc.StreamPhases.stop: emitted=%d reason=max_steps",
        emitted,
        extra={
            "rpc": "StreamPhases",
            "stream_event": "stop",
            "emitted": emitted,
            "reason": "max_steps",
            "status": "ok",
        },
    )

Functions:

Optional dependency detection

_compat.py exports two values:

• TWO_PI — 2π as float64
• HAS_RUST — True when spo_kernel Rust extension is importable

Individual modules handle their own optional imports locally:

Module	Guard	Package
nn/	pytest.importorskip("jax")	jax, equinox, optax
reporting/plots.py	_HAS_MPL	matplotlib
ssgf/topological_integration.py	_HAS_RIPSER	ripser
upde/engine.py	_HAS_RUST (from _compat)	spo_kernel
coupling/knm.py	_HAS_RUST (from _compat)	spo_kernel
oscillators/physical.py	_HAS_RUST (from _compat)	spo_kernel

When an optional dependency is missing, the corresponding subsystem either skips the optimised path (Rust → Python fallback) or raises ImportError with install instructions.

Environment readiness

runtime.doctor aggregates the per-module detection above into one report behind the spo doctor command. It probes the interpreter version against requires-python, the required runtime dependencies, the optional native backends (Rust/Julia/Go/Mojo), the optional feature extras, and package-local review/export adapter surfaces such as FMI co-simulation and hybrid co-compilation, returning a DoctorReport whose status is pass only when the interpreter is in range and every required dependency is importable. See the CLI reference for the command and exit codes.

doctor

Runtime environment readiness checks for spo doctor.

This module answers one question deterministically: can this interpreter run SCPN Phase Orchestrator, and which optional accelerators and feature extras are usable right now? It probes three layers without importing heavy optional packages or touching the network:

• the interpreter version against the packaged requires-python window;
• the mandatory runtime dependencies (NumPy, SciPy, PyYAML, Click, protobuf, urllib3) that the core engine imports unconditionally;
• the optional native compute backends (Rust spo_kernel, Julia juliacall, the Go toolchain/shared libraries, the Mojo toolchain) and the optional feature extras (nn, studio, queuewaves, plot, otel, notebook).
• opt-in review/export adapter surfaces that should ship with the package, such as the FMI co-simulation export and hybrid co-compiler manifest helpers.

Detection uses :func:importlib.util.find_spec for Python modules (which locates a distribution without executing its top-level import) and :func:shutil.which / filesystem probes for external toolchains, so running the diagnostics is cheap and free of side effects. A missing required dependency or an out-of-range interpreter makes the overall status fail (non-zero exit for the CLI); missing optional components are reported as warn and never fail the run.

Classes

DependencyCheck dataclass

DependencyCheck(
    name: str,
    category: str,
    required: bool,
    available: bool,
    detail: str,
    version: str | None = None,
)

Outcome of probing a single dependency, backend, or toolchain.

Attributes

name: Human-facing component name (for example ``numpy`` or ``rust``).
category: Grouping used for rendering — ``interpreter``, ``core``,
    ``backend``, or one of the optional feature-extra names.
required: Whether the component is mandatory for the core engine.
available: Whether the component was detected as usable.
detail: Short human-readable explanation (version string, path,
    install hint, or reason it is unavailable).
version: Resolved distribution version when known, otherwise ``None``.

Attributes

status property

status: str

Return ok/missing/warn for the component detection state.

Returns

str Return ok/missing/warn for the component detection state.

Methods:

to_record

to_record() -> dict[str, object]

Return a JSON-serialisable mapping with deterministic key order.

Returns

dict[str, object] Return a JSON-serialisable mapping with deterministic key order.

Source code in src/scpn_phase_orchestrator/runtime/doctor.py

def to_record(self) -> dict[str, object]:
    """Return a JSON-serialisable mapping with deterministic key order.

    Returns
    -------
    dict[str, object]
        Return a JSON-serialisable mapping with deterministic key order.
    """
    return {
        "name": self.name,
        "category": self.category,
        "required": self.required,
        "available": self.available,
        "status": self.status,
        "version": self.version,
        "detail": self.detail,
    }

DoctorReport dataclass

DoctorReport(
    checks: tuple[DependencyCheck, ...],
    python_version: str,
    platform: str = "",
)

Aggregate readiness report produced by :func:run_environment_diagnostics.

Attributes

checks: Every :class:`DependencyCheck` in deterministic display order.
python_version: The running interpreter version (``X.Y.Z``).
platform: Short OS/architecture descriptor for the audit record.

Attributes

missing_required property

missing_required: tuple[DependencyCheck, ...]

Required components that were not detected.

Returns

tuple[DependencyCheck, ...] Required components that were not detected.

missing_optional property

missing_optional: tuple[DependencyCheck, ...]

Optional components that were not detected.

Returns

tuple[DependencyCheck, ...] Optional components that were not detected.

ok property

ok: bool

True when every required component is present (overall pass).

Returns

bool True when every required component is present (overall pass).

status property

status: str

pass when ready, otherwise fail.

Returns

str pass when ready, otherwise fail.

exit_code property

exit_code: int

Process exit code: 0 when ready, 1 when a requirement is missing.

Returns

int Process exit code: 0 when ready, 1 when a requirement is missing.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a deterministic JSON-serialisable readiness record.

Returns

dict[str, object] Return a deterministic JSON-serialisable readiness record.

Source code in src/scpn_phase_orchestrator/runtime/doctor.py

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

    Returns
    -------
    dict[str, object]
        Return a deterministic JSON-serialisable readiness record.
    """
    return {
        "report": "environment-diagnostics",
        "version": "1.0.0",
        "status": self.status,
        "python_version": self.python_version,
        "platform": self.platform,
        "required_present": len(
            [c for c in self.checks if c.required and c.available]
        ),
        "required_total": len([c for c in self.checks if c.required]),
        "optional_present": len(
            [c for c in self.checks if not c.required and c.available]
        ),
        "optional_total": len([c for c in self.checks if not c.required]),
        "missing_required": [c.name for c in self.missing_required],
        "missing_optional": [c.name for c in self.missing_optional],
        "checks": [c.to_record() for c in self.checks],
    }

Functions:

run_environment_diagnostics

run_environment_diagnostics(
    *, repo_root: Path | None = None
) -> DoctorReport

Probe the interpreter, required dependencies, and optional components.

Parameters

repo_root : Path | None Optional explicit checkout root for the Go shared-library probe. When None the root is auto-detected from this module's location; pass a path in tests to exercise both branches.

Returns

DoctorReport A :class:DoctorReport whose :attr:DoctorReport.status is pass only when the interpreter is in range and every required dependency is importable.

Source code in src/scpn_phase_orchestrator/runtime/doctor.py

def run_environment_diagnostics(*, repo_root: Path | None = None) -> DoctorReport:
    """Probe the interpreter, required dependencies, and optional components.

    Parameters
    ----------
    repo_root : Path | None
        Optional explicit checkout root for the Go shared-library probe. When ``None``
        the root is auto-detected from this module's location; pass a path in tests to
        exercise both branches.

    Returns
    -------
    DoctorReport
        A :class:`DoctorReport` whose :attr:`DoctorReport.status` is ``pass`` only when
        the interpreter is in range and every required dependency is importable.
    """
    resolved_root = repo_root if repo_root is not None else _find_repo_root()

    checks: list[DependencyCheck] = [_check_python()]
    checks.extend(
        _check_modules(
            _CORE_DEPS,
            category="core",
            required=True,
            install_hint="install scpn-phase-orchestrator core dependencies",
        )
    )
    checks.append(_check_rust())
    checks.append(_check_rust_supervisor())
    checks.append(_check_julia())
    checks.append(_check_go(resolved_root))
    checks.append(_check_mojo())
    for extra, specs in _OPTIONAL_EXTRAS.items():
        checks.extend(
            _check_modules(
                specs,
                category=extra,
                required=False,
                install_hint=f"install the '{extra}' extra",
            )
        )
    for name, module_name, symbols, detail in _ADAPTER_SURFACES:
        checks.append(
            _check_adapter_surface(
                name=name,
                module_name=module_name,
                symbols=symbols,
                detail=detail,
            )
        )

    return DoctorReport(
        checks=tuple(checks),
        python_version=platform.python_version(),
        platform=f"{platform.system()} {platform.machine()}".strip(),
    )

render_report

render_report(report: DoctorReport) -> Sequence[str]

Render a :class:DoctorReport as aligned human-readable lines.

Source code in src/scpn_phase_orchestrator/runtime/doctor.py

def render_report(report: DoctorReport) -> Sequence[str]:
    """Render a :class:`DoctorReport` as aligned human-readable lines."""
    glyphs = {_STATUS_OK: "[ ok ]", _STATUS_MISSING: "[MISS]", _STATUS_WARN: "[warn]"}
    lines: list[str] = [
        f"SCPN Phase Orchestrator environment diagnostics — {report.status.upper()}",
        f"  python   {report.python_version}  ({report.platform})",
        "",
    ]
    width = max((len(c.name) for c in report.checks), default=0)
    for check in report.checks:
        glyph = glyphs.get(check.status, "[????]")
        lines.append(f"  {glyph} {check.name.ljust(width)}  {check.detail}")
    lines.append("")
    if report.missing_required:
        names = ", ".join(c.name for c in report.missing_required)
        lines.append(f"FAIL: missing required components: {names}")
    else:
        optional_present = len(
            [c for c in report.checks if not c.required and c.available]
        )
        optional_total = len([c for c in report.checks if not c.required])
        lines.append(
            "PASS: all required dependencies are present "
            f"({optional_present}/{optional_total} optional components available)."
        )
    return lines

Chaos-engineering resilience injection

runtime.chaos injects realistic, non-actuating faults — coupling drops, frequency drift, sensor noise, and drive dropout — into a controlled simulation and measures how the orchestrator recovers. A ChaosSchedule of ChaosFault windows is applied through the simulation's scenario_hook boundary, so the heavy compute stays in the existing multi-language UPDE engine and this module is the orchestration and scoring layer. run_resilience_experiment runs the spec once nominally and once perturbed under the same seed, then compute_resilience derives recovery time, peak coherence drop, stability-margin erosion, and final deviation. The spo chaos command exposes this from the CLI; all runs are review-only.

chaos

Chaos-engineering resilience injection for orchestrated phase control.

This module injects realistic, non-actuating perturbations — coupling drops, frequency drift, sensor noise, and drive dropout — into a controlled simulation of a binding spec, then measures how the orchestrator recovers. Faults are applied through the simulation's scenario_hook boundary, so the heavy compute stays in the existing multi-language UPDE engine; this module is the orchestration and resilience-scoring layer on top of it.

A resilience experiment runs the same seeded spec twice — once nominal, once with the fault schedule — and compares the two order-parameter trajectories to derive recovery time, peak coherence drop, stability-margin erosion, and final deviation. The output is review-only evidence: nothing here actuates hardware.

Classes

ChaosFault dataclass

ChaosFault(
    kind: str,
    start_step: int,
    duration_steps: int,
    magnitude: float,
)

One injected fault active over a bounded step window.

Attributes

kind : str Fault type: "coupling_drop" (scale the coupling matrix down), "frequency_drift" (offset the natural frequencies), "sensor_noise" (Gaussian phase perturbation), or "drive_dropout" (attenuate the external drive zeta). start_step : int First step at which the fault is active (>= 1 so step 0 captures the nominal coupling reference). duration_steps : int Number of consecutive steps the fault stays active. magnitude : float Fault strength. For coupling_drop and drive_dropout it is a fraction in [0, 1]; for frequency_drift it is an additive offset in rad/s; for sensor_noise it is the noise standard deviation in rad.

Attributes

end_step property

end_step: int

Return the first step after the fault window (exclusive).

Returns

int start_step + duration_steps.

Methods:

active_at

active_at(step: int) -> bool

Return whether the fault is active at step.

Parameters

step : int The simulation step index.

Returns

bool True when start_step <= step < end_step.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def active_at(self, step: int) -> bool:
    """Return whether the fault is active at ``step``.

    Parameters
    ----------
    step : int
        The simulation step index.

    Returns
    -------
    bool
        ``True`` when ``start_step <= step < end_step``.
    """
    return self.start_step <= step < self.end_step

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the fault.

Returns

dict[str, object] Deterministic, JSON-safe mapping of the fault fields.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the fault.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of the fault fields.
    """
    return {
        "kind": self.kind,
        "start_step": self.start_step,
        "duration_steps": self.duration_steps,
        "magnitude": self.magnitude,
    }

ChaosSchedule dataclass

ChaosSchedule(faults: tuple[ChaosFault, ...])

An ordered set of faults applied during a resilience experiment.

Attributes

faults : tuple[ChaosFault, ...] The faults to inject; at least one is required.

Attributes

last_fault_end property

last_fault_end: int

Return the last step at which any fault is still active, plus one.

Returns

int The maximum end_step across the scheduled faults.

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the schedule.

Returns

dict[str, object] Deterministic, JSON-safe mapping with the fault list and window end.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the schedule.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping with the fault list and window end.
    """
    return {
        "faults": [fault.to_audit_record() for fault in self.faults],
        "last_fault_end": self.last_fault_end,
    }

ResilienceMetrics dataclass

ResilienceMetrics(
    recovered: bool,
    recovery_steps: int | None,
    max_coherence_drop: float,
    stability_margin_erosion: float,
    final_deviation: float,
    metrics_hash: str,
)

Resilience evidence derived from nominal vs perturbed R trajectories.

Attributes

recovered : bool Whether the perturbed run returned within recovery_tolerance of the nominal run after the last fault ended. recovery_steps : int | None Steps after the last fault end until recovery, or None if the run never recovered within the trajectory. max_coherence_drop : float Largest positive nominal_R - perturbed_R across the trajectory. stability_margin_erosion : float Mean absolute nominal_R - perturbed_R over the post-fault-onset window. final_deviation : float Absolute difference of the final order parameters. metrics_hash : str Deterministic SHA-256 over the audit record (excluding the hash).

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the metrics.

Returns

dict[str, object] Deterministic, JSON-safe mapping of every metric field.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the metrics.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping of every metric field.
    """
    return {
        "recovered": self.recovered,
        "recovery_steps": self.recovery_steps,
        "max_coherence_drop": self.max_coherence_drop,
        "stability_margin_erosion": self.stability_margin_erosion,
        "final_deviation": self.final_deviation,
        "metrics_hash": self.metrics_hash,
    }

ChaosExperimentResult dataclass

ChaosExperimentResult(
    spec_name: str,
    steps: int,
    seed: int,
    schedule: ChaosSchedule,
    metrics: ResilienceMetrics,
    nominal_final_r: float,
    perturbed_final_r: float,
    result_hash: str,
)

Full result of one resilience experiment.

Attributes

spec_name : str Name of the binding spec exercised. steps : int Number of simulation steps. seed : int Shared RNG seed for the nominal and perturbed runs. schedule : ChaosSchedule The injected fault schedule. metrics : ResilienceMetrics The derived resilience evidence. nominal_final_r : float Final objective order parameter of the nominal run. perturbed_final_r : float Final objective order parameter of the perturbed run. result_hash : str Deterministic SHA-256 over the audit record (excluding the hash).

Methods:

to_audit_record

to_audit_record() -> dict[str, object]

Return a JSON-safe audit mapping of the experiment.

Returns

dict[str, object] Deterministic, JSON-safe mapping including schedule and metrics.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit mapping of the experiment.

    Returns
    -------
    dict[str, object]
        Deterministic, JSON-safe mapping including schedule and metrics.
    """
    return {
        "spec_name": self.spec_name,
        "steps": self.steps,
        "seed": self.seed,
        "schedule": self.schedule.to_audit_record(),
        "metrics": self.metrics.to_audit_record(),
        "nominal_final_r": self.nominal_final_r,
        "perturbed_final_r": self.perturbed_final_r,
        "non_actuating": True,
        "result_hash": self.result_hash,
    }

Functions:

make_chaos_hook

make_chaos_hook(
    schedule: ChaosSchedule,
) -> ScenarioCallback

Build a non-actuating scenario hook that injects the fault schedule.

The hook captures the nominal coupling matrix at step 0 (faults start at step >= 1) so that coupling_drop faults scale relative to the unperturbed coupling rather than compounding across the fault window.

Parameters

schedule : ChaosSchedule The fault schedule to apply.

Returns

ScenarioCallback A callable suitable for simulate(..., scenario_hook=...).

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def make_chaos_hook(schedule: ChaosSchedule) -> ScenarioCallback:
    """Build a non-actuating scenario hook that injects the fault schedule.

    The hook captures the nominal coupling matrix at step 0 (faults start at
    step >= 1) so that ``coupling_drop`` faults scale relative to the unperturbed
    coupling rather than compounding across the fault window.

    Parameters
    ----------
    schedule : ChaosSchedule
        The fault schedule to apply.

    Returns
    -------
    ScenarioCallback
        A callable suitable for ``simulate(..., scenario_hook=...)``.
    """
    nominal_knm: dict[str, FloatArray] = {}

    def hook(context: SimulationScenarioContext) -> None:
        """Apply the chaos fault hook to the simulation step."""
        if context.step == 0:
            nominal_knm["knm"] = np.array(context.coupling.knm, dtype=np.float64)
        for fault in schedule.faults:
            if not fault.active_at(context.step):
                continue
            _apply_fault(fault, context, nominal_knm.get("knm"))

    return hook

compute_resilience

compute_resilience(
    nominal_history: tuple[float, ...],
    perturbed_history: tuple[float, ...],
    *,
    fault_onset_step: int,
    last_fault_end: int,
    recovery_tolerance: float,
) -> ResilienceMetrics

Score resilience from nominal vs perturbed order-parameter trajectories.

Parameters

nominal_history, perturbed_history : tuple[float, ...] Per-step objective order parameters of the nominal and perturbed runs; they must share length. fault_onset_step : int First step at which any fault becomes active; the erosion window starts here. last_fault_end : int First step after the last fault; recovery is measured from here. recovery_tolerance : float Absolute |nominal_R - perturbed_R| at or below which the perturbed run counts as recovered.

Returns

ResilienceMetrics The derived resilience evidence with a deterministic hash.

Raises

ValueError If the histories are empty, length-mismatched, or the parameters are out of range.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def compute_resilience(
    nominal_history: tuple[float, ...],
    perturbed_history: tuple[float, ...],
    *,
    fault_onset_step: int,
    last_fault_end: int,
    recovery_tolerance: float,
) -> ResilienceMetrics:
    """Score resilience from nominal vs perturbed order-parameter trajectories.

    Parameters
    ----------
    nominal_history, perturbed_history : tuple[float, ...]
        Per-step objective order parameters of the nominal and perturbed runs;
        they must share length.
    fault_onset_step : int
        First step at which any fault becomes active; the erosion window starts
        here.
    last_fault_end : int
        First step after the last fault; recovery is measured from here.
    recovery_tolerance : float
        Absolute ``|nominal_R - perturbed_R|`` at or below which the perturbed
        run counts as recovered.

    Returns
    -------
    ResilienceMetrics
        The derived resilience evidence with a deterministic hash.

    Raises
    ------
    ValueError
        If the histories are empty, length-mismatched, or the parameters are
        out of range.
    """
    nominal = np.asarray(nominal_history, dtype=np.float64)
    perturbed = np.asarray(perturbed_history, dtype=np.float64)
    if nominal.size == 0:
        raise ValueError("nominal_history must not be empty")
    if nominal.shape != perturbed.shape:
        raise ValueError("nominal_history and perturbed_history must share length")
    steps = int(nominal.size)
    onset = _positive_int(fault_onset_step, name="fault_onset_step", minimum=0)
    end = _positive_int(last_fault_end, name="last_fault_end", minimum=0)
    tolerance = _finite_real(recovery_tolerance, name="recovery_tolerance", minimum=0.0)

    signed_drop = nominal - perturbed
    max_coherence_drop = float(max(0.0, float(np.max(signed_drop))))
    erosion_window = signed_drop[min(onset, steps) :]
    stability_margin_erosion = (
        float(np.mean(np.abs(erosion_window))) if erosion_window.size else 0.0
    )
    final_deviation = float(abs(nominal[-1] - perturbed[-1]))

    recovery_steps: int | None = None
    recovered = False
    for step in range(min(end, steps), steps):
        if abs(nominal[step] - perturbed[step]) <= tolerance:
            recovery_steps = step - min(end, steps)
            recovered = True
            break

    metrics = ResilienceMetrics(
        recovered=recovered,
        recovery_steps=recovery_steps,
        max_coherence_drop=max_coherence_drop,
        stability_margin_erosion=stability_margin_erosion,
        final_deviation=final_deviation,
        metrics_hash="",
    )
    return _with_metrics_hash(metrics)

run_resilience_experiment

run_resilience_experiment(
    spec: BindingSpec,
    schedule: ChaosSchedule,
    *,
    steps: int = 200,
    seed: int = 42,
    recovery_tolerance: float = 0.05,
) -> ChaosExperimentResult

Run a nominal and a fault-injected simulation and score resilience.

Both runs share the spec, step count, and seed, so the only difference is the injected fault schedule. The simulations are closed-loop (the supervisor reacts to the faults); the result is review-only evidence.

Parameters

spec : BindingSpec A validated binding spec. schedule : ChaosSchedule The fault schedule to inject in the perturbed run. steps : int, optional Number of simulation steps (default 200); must exceed the last fault end so recovery can be observed. seed : int, optional Shared RNG seed (default 42). recovery_tolerance : float, optional Recovery tolerance passed to :func:compute_resilience (default 0.05).

Returns

ChaosExperimentResult The schedule, derived metrics, and final order parameters.

Raises

ValueError If steps does not exceed the last fault end.

Source code in src/scpn_phase_orchestrator/runtime/chaos.py

def run_resilience_experiment(
    spec: BindingSpec,
    schedule: ChaosSchedule,
    *,
    steps: int = 200,
    seed: int = 42,
    recovery_tolerance: float = 0.05,
) -> ChaosExperimentResult:
    """Run a nominal and a fault-injected simulation and score resilience.

    Both runs share the spec, step count, and seed, so the only difference is the
    injected fault schedule. The simulations are closed-loop (the supervisor
    reacts to the faults); the result is review-only evidence.

    Parameters
    ----------
    spec : BindingSpec
        A validated binding spec.
    schedule : ChaosSchedule
        The fault schedule to inject in the perturbed run.
    steps : int, optional
        Number of simulation steps (default ``200``); must exceed the last fault
        end so recovery can be observed.
    seed : int, optional
        Shared RNG seed (default ``42``).
    recovery_tolerance : float, optional
        Recovery tolerance passed to :func:`compute_resilience` (default ``0.05``).

    Returns
    -------
    ChaosExperimentResult
        The schedule, derived metrics, and final order parameters.

    Raises
    ------
    ValueError
        If ``steps`` does not exceed the last fault end.
    """
    steps = _positive_int(steps, name="steps", minimum=1)
    if steps <= schedule.last_fault_end:
        raise ValueError("steps must exceed the schedule's last fault end")

    nominal = simulate(spec, steps=steps, seed=seed, policy_enabled=True)
    perturbed = simulate(
        spec,
        steps=steps,
        seed=seed,
        policy_enabled=True,
        scenario_hook=make_chaos_hook(schedule),
    )
    fault_onset = min(fault.start_step for fault in schedule.faults)
    metrics = compute_resilience(
        nominal.r_good_history,
        perturbed.r_good_history,
        fault_onset_step=fault_onset,
        last_fault_end=schedule.last_fault_end,
        recovery_tolerance=recovery_tolerance,
    )
    result = ChaosExperimentResult(
        spec_name=nominal.spec_name,
        steps=steps,
        seed=seed,
        schedule=schedule,
        metrics=metrics,
        nominal_final_r=nominal.r_good,
        perturbed_final_r=perturbed.r_good,
        result_hash="",
    )
    return _with_result_hash(result)

Deterministic (bounded-jitter) execution mode

runtime.deterministic runs an arbitrary per-step callable against a fixed period with the timing guarantees a plain loop cannot give: every step is scheduled at t0 + i·period on the monotonic clock (sleep, with an optional final spin, then measured jitter), each step is timed against a worst-case execution-time budget so an overrun is a fatal deadline miss by default (miss_policy='abort'). Callers must explicitly choose miss_policy='observe' for diagnostic runs that record misses and continue. The cyclic garbage collector is frozen and disabled for the hot path so GC pauses leave the jitter budget. run_deterministic_loop returns an ExecutionTimingReport with per-step latencies and jitters plus aggregate statistics (mean / max / p99 latency, max absolute jitter, deadline-miss count). The loop is timing-only and non-actuating: it never inspects or mutates the step's state, so it drives the simulation step, a controller tick, or any periodic task without coupling to a specific engine. Advanced callers may pass a monotonic clock_ns and matching wait_until hook for deterministic simulator clocks; production calls default to time.perf_counter_ns and the built-in sleep/spin waiter.

deterministic

Bounded-jitter, hard-deadline execution mode for the control step loop.

A control loop only earns credibility if each step lands on time. This module runs an arbitrary per-step callable against a fixed period with three guarantees a plain for loop cannot give:

• Bounded jitter — every step is scheduled at t0 + i·period on the monotonic clock; the loop sleeps to that boundary (optionally finishing the last busy_wait_margin_s with a spin) and records the actual start offset so jitter is measured, not assumed.
• WCET budget — each step is timed against a worst-case execution-time budget; an overrun is a deadline miss. The default miss_policy='abort' raises :class:DeadlineExceededError; callers must explicitly request miss_policy='observe' to record and continue.
• No-GC hot path — the cyclic garbage collector is frozen and disabled for the duration of the loop, removing GC pauses from the jitter budget, and restored to its prior state afterwards.

The loop is non-actuating and timing-only: it never inspects or mutates the step's state. The caller closes over its own state in the step callable, so this drives the simulation step, a controller tick, or any periodic task without coupling to a specific engine. Step results stay exactly as deterministic as the callable; this module makes their timing bounded, which is the property hard-real-time control needs.

Classes

DeadlineExceededError

DeadlineExceededError(
    step_index: int, latency_s: float, wcet_s: float
)

Bases: RuntimeError

Raised when a step overruns its WCET budget under miss_policy='abort'.

Attributes

step_index : int The zero-based index of the step that overran. latency_s : float The measured execution time of that step, in seconds. wcet_s : float The worst-case execution-time budget that was exceeded, in seconds.

Source code in src/scpn_phase_orchestrator/runtime/deterministic.py

def __init__(self, step_index: int, latency_s: float, wcet_s: float) -> None:
    self.step_index = step_index
    self.latency_s = latency_s
    self.wcet_s = wcet_s
    super().__init__(
        f"step {step_index} took {latency_s * 1e3:.3f} ms, "
        f"over the {wcet_s * 1e3:.3f} ms WCET budget"
    )

DeadlineBudget dataclass

DeadlineBudget(
    period_s: float,
    wcet_s: float | None = None,
    miss_policy: str = "abort",
    freeze_gc: bool = True,
    busy_wait_margin_s: float = 0.0,
)

Timing budget for a bounded-jitter step loop.

Attributes

period_s : float Target wall-clock period between consecutive step starts, in seconds. wcet_s : float Worst-case execution-time budget for a single step, in seconds. A step whose measured latency exceeds this is a deadline miss. Defaults to the full period_s (a step may use the whole period). miss_policy : str 'abort' raises :class:DeadlineExceededError on the first miss. 'observe' records deadline misses and continues, and must be selected explicitly for diagnostic runs. freeze_gc : bool Freeze (gc.freeze) and disable the cyclic garbage collector for the loop, restoring the prior state afterwards. Removes GC pauses from the jitter budget. busy_wait_margin_s : float Spin (busy-wait) for the final busy_wait_margin_s before each scheduled boundary instead of sleeping, trading CPU for lower jitter. 0.0 (default) sleeps the whole remainder.

Attributes

effective_wcet_s property

effective_wcet_s: float

Return the WCET budget, defaulting to the full period.

ExecutionTimingReport dataclass

ExecutionTimingReport(
    latencies_s: FloatArray,
    jitters_s: FloatArray,
    period_s: float,
    wcet_s: float,
    deadline_misses: int,
    gc_frozen: bool,
    wall_time_s: float,
)

Timing record of a bounded-jitter step loop.

Attributes

latencies_s : FloatArray Per-step execution time, shape (steps,). jitters_s : FloatArray Per-step start offset from the scheduled boundary (signed; positive is late), shape (steps,). period_s : float The target period. wcet_s : float The effective WCET budget against which misses were counted. deadline_misses : int Number of steps whose latency exceeded wcet_s. gc_frozen : bool Whether the garbage collector was frozen/disabled for the loop. wall_time_s : float Total wall-clock time of the loop.

Attributes

steps property

steps: int

Number of completed steps.

max_latency_s property

max_latency_s: float

Largest single-step execution time.

mean_latency_s property

mean_latency_s: float

Mean step execution time.

max_abs_jitter_s property

max_abs_jitter_s: float

Largest absolute start-offset from the scheduled boundary.

deadline_met property

deadline_met: bool

Whether every step stayed within the WCET budget.

Methods:

latency_percentile_s

latency_percentile_s(percentile: float) -> float

Return the step latency at a given percentile.

Parameters

percentile : float Percentile in [0, 100].

Returns

float The latency at percentile in seconds, or 0.0 if no steps ran.

Raises

ValueError If percentile is not a real number in [0, 100].

Source code in src/scpn_phase_orchestrator/runtime/deterministic.py

def latency_percentile_s(self, percentile: float) -> float:
    """Return the step latency at a given percentile.

    Parameters
    ----------
    percentile : float
        Percentile in ``[0, 100]``.

    Returns
    -------
    float
        The latency at ``percentile`` in seconds, or ``0.0`` if no steps ran.

    Raises
    ------
    ValueError
        If ``percentile`` is not a real number in ``[0, 100]``.
    """
    if isinstance(percentile, bool) or not isinstance(percentile, Real):
        raise ValueError(
            f"percentile must be a real in [0, 100], got {percentile!r}"
        )
    value = float(percentile)
    if not np.isfinite(value) or value < 0.0 or value > 100.0:
        raise ValueError(f"percentile must lie in [0, 100], got {percentile!r}")
    if not self.latencies_s.size:
        return 0.0
    return float(np.percentile(self.latencies_s, value))

summary

summary() -> dict[str, float | int | bool]

Return a flat scalar summary for logging or metric export.

Returns

dict[str, float | int | bool] Step count, the period and WCET budget, mean / max / p99 latency, maximum absolute jitter, deadline-miss count and verdict, the GC policy, and the loop wall-time.

Source code in src/scpn_phase_orchestrator/runtime/deterministic.py

def summary(self) -> dict[str, float | int | bool]:
    """Return a flat scalar summary for logging or metric export.

    Returns
    -------
    dict[str, float | int | bool]
        Step count, the period and WCET budget, mean / max / p99 latency,
        maximum absolute jitter, deadline-miss count and verdict, the GC
        policy, and the loop wall-time.
    """
    return {
        "steps": self.steps,
        "period_s": self.period_s,
        "wcet_s": self.wcet_s,
        "mean_latency_s": self.mean_latency_s,
        "max_latency_s": self.max_latency_s,
        "p99_latency_s": self.latency_percentile_s(99.0),
        "max_abs_jitter_s": self.max_abs_jitter_s,
        "deadline_misses": self.deadline_misses,
        "deadline_met": self.deadline_met,
        "gc_frozen": self.gc_frozen,
        "wall_time_s": self.wall_time_s,
    }

Functions:

run_deterministic_loop

run_deterministic_loop(
    step: StepCallable,
    *,
    steps: int,
    budget: DeadlineBudget,
    clock_ns: _MonotonicClockNs | None = None,
    wait_until: _WaitUntilNs | None = None,
) -> ExecutionTimingReport

Drive step for steps iterations under a bounded-jitter budget.

Parameters

step : StepCallable Per-step callable receiving the zero-based step index. It owns all state through its closure; its return value is ignored. steps : int Number of iterations (>= 0). budget : DeadlineBudget The period, WCET budget, miss policy, GC policy, and spin margin. clock_ns : Callable[[], int], optional Monotonic nanosecond clock. Defaults to time.perf_counter_ns. Supplying this with wait_until lets tests and deterministic simulators exercise scheduling branches without depending on host wall time. wait_until : Callable[[int, int], None], optional Wait hook receiving the target monotonic nanosecond timestamp and busy wait margin. Defaults to the production sleep/spin waiter. Custom hooks must observe the same time base as clock_ns.

Returns

ExecutionTimingReport Per-step latencies and jitters plus aggregate timing statistics.

Raises

ValueError If steps is not a non-negative integer. DeadlineExceededError If a step overruns budget.wcet_s and miss_policy == 'abort'.

Source code in src/scpn_phase_orchestrator/runtime/deterministic.py

def run_deterministic_loop(
    step: StepCallable,
    *,
    steps: int,
    budget: DeadlineBudget,
    clock_ns: _MonotonicClockNs | None = None,
    wait_until: _WaitUntilNs | None = None,
) -> ExecutionTimingReport:
    """Drive ``step`` for ``steps`` iterations under a bounded-jitter budget.

    Parameters
    ----------
    step : StepCallable
        Per-step callable receiving the zero-based step index. It owns all state
        through its closure; its return value is ignored.
    steps : int
        Number of iterations (``>= 0``).
    budget : DeadlineBudget
        The period, WCET budget, miss policy, GC policy, and spin margin.
    clock_ns : Callable[[], int], optional
        Monotonic nanosecond clock. Defaults to ``time.perf_counter_ns``.
        Supplying this with ``wait_until`` lets tests and deterministic
        simulators exercise scheduling branches without depending on host wall
        time.
    wait_until : Callable[[int, int], None], optional
        Wait hook receiving the target monotonic nanosecond timestamp and busy
        wait margin. Defaults to the production sleep/spin waiter. Custom hooks
        must observe the same time base as ``clock_ns``.

    Returns
    -------
    ExecutionTimingReport
        Per-step latencies and jitters plus aggregate timing statistics.

    Raises
    ------
    ValueError
        If ``steps`` is not a non-negative integer.
    DeadlineExceededError
        If a step overruns ``budget.wcet_s`` and ``miss_policy == 'abort'``.
    """
    steps = _validate_steps(steps)
    monotonic_ns = time.perf_counter_ns if clock_ns is None else clock_ns
    wait_until_ns: _WaitUntilNs
    if wait_until is None:

        def default_wait_until_ns(target_ns: int, margin_ns: int) -> None:
            """Busy-wait until ``target_ns`` using the loop's monotonic clock."""
            _sleep_until(target_ns, margin_ns, clock_ns=monotonic_ns)

        wait_until_ns = default_wait_until_ns
    else:
        wait_until_ns = wait_until
    period_ns = round(budget.period_s * 1e9)
    wcet_s = budget.effective_wcet_s
    margin_ns = round(budget.busy_wait_margin_s * 1e9)
    latencies: FloatArray = np.zeros(steps, dtype=np.float64)
    jitters: FloatArray = np.zeros(steps, dtype=np.float64)
    deadline_misses = 0

    gc_was_enabled = gc.isenabled()
    if budget.freeze_gc:
        gc.collect()
        gc.freeze()
        gc.disable()
    loop_start_ns = monotonic_ns()
    try:
        for index in range(steps):
            scheduled_ns = loop_start_ns + index * period_ns
            if monotonic_ns() < scheduled_ns:
                wait_until_ns(scheduled_ns, margin_ns)
            actual_start_ns = monotonic_ns()
            jitters[index] = (actual_start_ns - scheduled_ns) / 1e9
            step(index)
            latency_s = (monotonic_ns() - actual_start_ns) / 1e9
            latencies[index] = latency_s
            if latency_s > wcet_s:
                deadline_misses += 1
                if budget.miss_policy == "abort":
                    raise DeadlineExceededError(index, latency_s, wcet_s)
        wall_time_s = (monotonic_ns() - loop_start_ns) / 1e9
    finally:
        if budget.freeze_gc:
            gc.unfreeze()
            if gc_was_enabled:
                gc.enable()

    return ExecutionTimingReport(
        latencies_s=latencies,
        jitters_s=jitters,
        period_s=float(budget.period_s),
        wcet_s=wcet_s,
        deadline_misses=deadline_misses,
        gc_frozen=budget.freeze_gc,
        wall_time_s=wall_time_s,
    )

Post-quantum audit-chain seal

runtime.audit_pqc adds an additive, post-quantum seal over the audit hash chain. The audit logger already chains each record with SHA-256 and signs it with HMAC; HMAC is symmetric and not post-quantum. seal_audit_log signs the chain tip (the _hash of the last record — the SHA-256 commitment to the whole log) with ML-DSA (FIPS 204, available natively in cryptography), producing an AuditChainSeal that anyone holding the trusted public key can verify, long after the run and against a future quantum adversary. verify_audit_log_seal re-reads the chain tip and record count and rejects the seal if either changed, so any post-hoc edit is detected. ML-DSA-65 is the default; ML-DSA-44/87 are selectable, and the seal records its algorithm so SLH-DSA (FIPS 205) can be added later without breaking existing seals. The seal is additive — it does not touch the HMAC flow — so it carries no regression risk.

audit_pqc

Post-quantum seal over the audit hash chain.

The audit logger already chains every record with SHA-256 and signs each one with HMAC. HMAC is symmetric: a verifier needs the secret key, and the scheme is not post-quantum. This module adds an additive second seal — it does not touch the HMAC flow, so there is no regression risk — that commits to the whole log with a post-quantum, publicly verifiable signature.

The chain tip (_hash of the last record) is the SHA-256 commitment to the entire log: change any record and the tip changes. seal_audit_log signs that tip with ML-DSA (FIPS 204, the NIST module-lattice signature standard), producing an :class:AuditChainSeal that anyone holding the trusted public key can verify long after the run — and against a future quantum adversary.

ML-DSA-65 (NIST security category 3) is the default; ML-DSA-44 and ML-DSA-87 are available for lower/higher assurance. FIPS 205 (SLH-DSA / SPHINCS+) is the hash-based alternative; the seal records its algorithm so a second scheme can be added without breaking existing seals.

ML-DSA is provided by the optional cryptography dependency. Install the pqc extra (pip install scpn-phase-orchestrator[pqc]) to use this module; cryptography is imported lazily, so importing the module never requires it. ML-DSA additionally needs an OpenSSL 3.5+ backend, which not every platform wheel bundles (e.g. the Windows cryptography wheel); on an older backend the seal functions raise cryptography.exceptions.UnsupportedAlgorithm.

The seal is signed over a domain-separated message binding the algorithm, the record count, and the tip hash, so a signature cannot be replayed across schemes or truncated logs.

Classes

AuditChainSeal dataclass

AuditChainSeal(
    algorithm: str,
    public_key_id: str,
    public_key_hex: str,
    tip_hash: str,
    record_count: int,
    signature_hex: str,
)

A post-quantum signature committing to an audit hash chain.

Attributes

algorithm : str The ML-DSA variant used (one of :data:MLDSA_VARIANTS). public_key_id : str Short identifier of the public key (SHA-256 prefix). public_key_hex : str The raw ML-DSA public key, hex-encoded, for verification. tip_hash : str The SHA-256 chain tip (_hash of the last record) that is sealed. record_count : int Number of records the chain contained when sealed. signature_hex : str The ML-DSA signature over the domain-separated seal message, hex-encoded.

Methods:

to_dict

to_dict() -> dict[str, str | int]

Return a JSON-serialisable mapping of the seal.

Returns

dict[str, str | int] The six seal fields as plain JSON-serialisable values.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def to_dict(self) -> dict[str, str | int]:
    """Return a JSON-serialisable mapping of the seal.

    Returns
    -------
    dict[str, str | int]
        The six seal fields as plain JSON-serialisable values.
    """
    return {
        "algorithm": self.algorithm,
        "public_key_id": self.public_key_id,
        "public_key_hex": self.public_key_hex,
        "tip_hash": self.tip_hash,
        "record_count": self.record_count,
        "signature_hex": self.signature_hex,
    }

from_dict classmethod

from_dict(data: dict[str, Any]) -> AuditChainSeal

Return a seal parsed from a mapping (e.g. loaded JSON).

Parameters

data : dict[str, Any] A mapping carrying the six seal fields.

Returns

AuditChainSeal The reconstructed seal.

Raises

ValueError If any required field is missing.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> AuditChainSeal:
    """Return a seal parsed from a mapping (e.g. loaded JSON).

    Parameters
    ----------
    data : dict[str, Any]
        A mapping carrying the six seal fields.

    Returns
    -------
    AuditChainSeal
        The reconstructed seal.

    Raises
    ------
    ValueError
        If any required field is missing.
    """
    required = {
        "algorithm",
        "public_key_id",
        "public_key_hex",
        "tip_hash",
        "record_count",
        "signature_hex",
    }
    missing = required - set(data)
    if missing:
        raise ValueError(f"seal is missing fields: {sorted(missing)}")
    return cls(
        algorithm=str(data["algorithm"]),
        public_key_id=str(data["public_key_id"]),
        public_key_hex=str(data["public_key_hex"]),
        tip_hash=str(data["tip_hash"]),
        record_count=int(data["record_count"]),
        signature_hex=str(data["signature_hex"]),
    )

Functions:

generate_signing_seed

generate_signing_seed() -> str

Return a fresh ML-DSA signing seed.

Returns

str A cryptographically random 32-byte seed, hex-encoded.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def generate_signing_seed() -> str:
    """Return a fresh ML-DSA signing seed.

    Returns
    -------
    str
        A cryptographically random 32-byte seed, hex-encoded.
    """
    return os.urandom(SEED_BYTES).hex()

signing_key_from_seed

signing_key_from_seed(
    seed_hex: str, *, algorithm: str = DEFAULT_VARIANT
) -> Any

Return a deterministic ML-DSA private key derived from a seed.

Parameters

seed_hex : str A 32-byte seed, hex-encoded (see :func:generate_signing_seed). algorithm : str The ML-DSA variant (one of :data:MLDSA_VARIANTS).

Returns

cryptography ML-DSA private key The deterministic private key for the seed and variant.

Raises

ValueError If the seed or algorithm is invalid.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def signing_key_from_seed(seed_hex: str, *, algorithm: str = DEFAULT_VARIANT) -> Any:
    """Return a deterministic ML-DSA private key derived from a seed.

    Parameters
    ----------
    seed_hex : str
        A 32-byte seed, hex-encoded (see :func:`generate_signing_seed`).
    algorithm : str
        The ML-DSA variant (one of :data:`MLDSA_VARIANTS`).

    Returns
    -------
    cryptography ML-DSA private key
        The deterministic private key for the seed and variant.

    Raises
    ------
    ValueError
        If the seed or algorithm is invalid.
    """
    algorithm = _require_variant(algorithm)
    seed = _validate_seed(seed_hex)
    return _private_class(algorithm).from_seed_bytes(seed)

sign_bytes

sign_bytes(
    message: bytes,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> bytes

Sign an arbitrary message with an ML-DSA private key.

This is the generic post-quantum signing primitive that higher-level sealers (the audit-chain seal, the DSSE attestation envelope) build on, so the cryptography backend is loaded in exactly one place.

Parameters

message : bytes The raw message to sign (already domain-separated by the caller). private_key : MLDSA private key An ML-DSA private key matching algorithm. algorithm : str The ML-DSA variant; must match private_key.

Returns

bytes The raw ML-DSA signature.

Raises

ValueError If the algorithm, message, or private key is invalid.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def sign_bytes(
    message: bytes,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> bytes:
    """Sign an arbitrary message with an ML-DSA private key.

    This is the generic post-quantum signing primitive that higher-level sealers
    (the audit-chain seal, the DSSE attestation envelope) build on, so the
    ``cryptography`` backend is loaded in exactly one place.

    Parameters
    ----------
    message : bytes
        The raw message to sign (already domain-separated by the caller).
    private_key : MLDSA private key
        An ML-DSA private key matching ``algorithm``.
    algorithm : str
        The ML-DSA variant; must match ``private_key``.

    Returns
    -------
    bytes
        The raw ML-DSA signature.

    Raises
    ------
    ValueError
        If the algorithm, message, or private key is invalid.
    """
    algorithm = _require_variant(algorithm)
    if not isinstance(message, (bytes, bytearray)):
        raise ValueError("message must be raw bytes")
    if not isinstance(private_key, _private_class(algorithm)):
        raise ValueError(f"private_key must be an {algorithm} private key")
    return bytes(private_key.sign(bytes(message)))

verify_bytes

verify_bytes(
    message: bytes,
    signature: bytes,
    trusted_public_key_hex: str,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> bool

Verify an ML-DSA signature over a message against a trusted public key.

Parameters

message : bytes The raw message the signature is expected to cover. signature : bytes The raw ML-DSA signature. trusted_public_key_hex : str The hex-encoded raw ML-DSA public key the verifier trusts. algorithm : str The ML-DSA variant to verify under.

Returns

bool True if the signature is valid for the trusted key, else False.

Raises

ValueError If the algorithm is unknown, the message or signature is not raw bytes, or the trusted key is not a hex string.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def verify_bytes(
    message: bytes,
    signature: bytes,
    trusted_public_key_hex: str,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> bool:
    """Verify an ML-DSA signature over a message against a trusted public key.

    Parameters
    ----------
    message : bytes
        The raw message the signature is expected to cover.
    signature : bytes
        The raw ML-DSA signature.
    trusted_public_key_hex : str
        The hex-encoded raw ML-DSA public key the verifier trusts.
    algorithm : str
        The ML-DSA variant to verify under.

    Returns
    -------
    bool
        ``True`` if the signature is valid for the trusted key, else ``False``.

    Raises
    ------
    ValueError
        If the algorithm is unknown, the message or signature is not raw bytes, or
        the trusted key is not a hex string.
    """
    algorithm = _require_variant(algorithm)
    if not isinstance(message, (bytes, bytearray)):
        raise ValueError("message must be raw bytes")
    if not isinstance(signature, (bytes, bytearray)):
        raise ValueError("signature must be raw bytes")
    if not isinstance(trusted_public_key_hex, str):
        raise ValueError("trusted_public_key_hex must be a hex string")
    try:
        trusted_bytes = bytes.fromhex(trusted_public_key_hex)
    except ValueError as exc:
        raise ValueError("trusted_public_key_hex must be valid hex") from exc
    from cryptography.exceptions import InvalidSignature

    try:
        public_key = _public_class(algorithm).from_public_bytes(trusted_bytes)
    except ValueError:
        return False
    try:
        public_key.verify(bytes(signature), bytes(message))
    except InvalidSignature:
        return False
    return True

public_key_id

public_key_id(public_bytes: bytes) -> str

Return the short identifier of a public key.

Parameters

public_bytes : bytes The raw ML-DSA public key bytes.

Returns

str The first 16 hex characters of the key's SHA-256 digest.

Raises

ValueError If public_bytes is not raw bytes.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def public_key_id(public_bytes: bytes) -> str:
    """Return the short identifier of a public key.

    Parameters
    ----------
    public_bytes : bytes
        The raw ML-DSA public key bytes.

    Returns
    -------
    str
        The first 16 hex characters of the key's SHA-256 digest.

    Raises
    ------
    ValueError
        If ``public_bytes`` is not raw bytes.
    """
    if not isinstance(public_bytes, (bytes, bytearray)):
        raise ValueError("public_bytes must be raw bytes")
    return hashlib.sha256(bytes(public_bytes)).hexdigest()[:16]

seal_audit_chain

seal_audit_chain(
    tip_hash: str,
    record_count: int,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> AuditChainSeal

Sign an audit chain tip with ML-DSA and return the seal.

Parameters

tip_hash : str The SHA-256 chain tip (32-byte digest, hex). record_count : int Number of records in the sealed chain. private_key : MLDSA private key An ML-DSA private key matching algorithm (see :func:signing_key_from_seed). algorithm : str The ML-DSA variant; must match private_key.

Returns

AuditChainSeal The post-quantum seal over the chain tip.

Raises

ValueError If the algorithm, tip hash, record count, or private key is invalid.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def seal_audit_chain(
    tip_hash: str,
    record_count: int,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> AuditChainSeal:
    """Sign an audit chain tip with ML-DSA and return the seal.

    Parameters
    ----------
    tip_hash : str
        The SHA-256 chain tip (32-byte digest, hex).
    record_count : int
        Number of records in the sealed chain.
    private_key : MLDSA private key
        An ML-DSA private key matching ``algorithm`` (see
        :func:`signing_key_from_seed`).
    algorithm : str
        The ML-DSA variant; must match ``private_key``.

    Returns
    -------
    AuditChainSeal
        The post-quantum seal over the chain tip.

    Raises
    ------
    ValueError
        If the algorithm, tip hash, record count, or private key is invalid.
    """
    algorithm = _require_variant(algorithm)
    tip_hash = _validate_tip_hash(tip_hash)
    record_count = _validate_record_count(record_count)
    if not isinstance(private_key, _private_class(algorithm)):
        raise ValueError(f"private_key must be an {algorithm} private key")
    public_bytes = private_key.public_key().public_bytes_raw()
    message = _signing_message(algorithm, record_count, tip_hash)
    signature = private_key.sign(message)
    return AuditChainSeal(
        algorithm=algorithm,
        public_key_id=public_key_id(public_bytes),
        public_key_hex=public_bytes.hex(),
        tip_hash=tip_hash,
        record_count=record_count,
        signature_hex=signature.hex(),
    )

verify_audit_chain_seal

verify_audit_chain_seal(
    seal: AuditChainSeal, trusted_public_key_hex: str
) -> bool

Verify a seal's signature against a trusted public key.

The verifier must supply the public key it trusts; the key embedded in the seal is only used as a convenience and is checked to match the trusted one, so an attacker cannot re-sign a forged log under their own key.

Parameters

seal : AuditChainSeal The seal to verify. trusted_public_key_hex : str The hex-encoded raw ML-DSA public key the verifier trusts.

Returns

bool True if the signature is valid for the trusted key, else False.

Raises

ValueError If the seal's algorithm is unknown or the trusted key is malformed.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def verify_audit_chain_seal(seal: AuditChainSeal, trusted_public_key_hex: str) -> bool:
    """Verify a seal's signature against a trusted public key.

    The verifier must supply the public key it trusts; the key embedded in the
    seal is only used as a convenience and is checked to match the trusted one,
    so an attacker cannot re-sign a forged log under their own key.

    Parameters
    ----------
    seal : AuditChainSeal
        The seal to verify.
    trusted_public_key_hex : str
        The hex-encoded raw ML-DSA public key the verifier trusts.

    Returns
    -------
    bool
        ``True`` if the signature is valid for the trusted key, else ``False``.

    Raises
    ------
    ValueError
        If the seal's algorithm is unknown or the trusted key is malformed.
    """
    algorithm = _require_variant(seal.algorithm)
    if not isinstance(trusted_public_key_hex, str):
        raise ValueError("trusted_public_key_hex must be a hex string")
    try:
        trusted_bytes = bytes.fromhex(trusted_public_key_hex)
    except ValueError as exc:
        raise ValueError("trusted_public_key_hex must be valid hex") from exc
    if public_key_id(trusted_bytes) != seal.public_key_id:
        return False
    if trusted_bytes.hex() != seal.public_key_hex:
        return False
    from cryptography.exceptions import InvalidSignature

    try:
        public_key = _public_class(algorithm).from_public_bytes(trusted_bytes)
        signature = bytes.fromhex(seal.signature_hex)
    except ValueError:
        return False
    message = _signing_message(algorithm, seal.record_count, seal.tip_hash)
    try:
        public_key.verify(signature, message)
    except InvalidSignature:
        return False
    return True

read_audit_chain_tip

read_audit_chain_tip(path: Path) -> tuple[str, int]

Return the chain tip _hash and record count of an audit JSONL file.

Parameters

path : Path Path to the audit JSONL stream.

Returns

tuple[str, int] (tip_hash, record_count).

Raises

ValueError If the file is empty or the last record carries no _hash. FileNotFoundError If the file does not exist.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def read_audit_chain_tip(path: Path) -> tuple[str, int]:
    """Return the chain tip ``_hash`` and record count of an audit JSONL file.

    Parameters
    ----------
    path : Path
        Path to the audit JSONL stream.

    Returns
    -------
    tuple[str, int]
        ``(tip_hash, record_count)``.

    Raises
    ------
    ValueError
        If the file is empty or the last record carries no ``_hash``.
    FileNotFoundError
        If the file does not exist.
    """
    text = Path(path).read_text(encoding="utf-8")
    records = [line for line in text.splitlines() if line.strip()]
    if not records:
        raise ValueError("audit stream is empty; nothing to seal")
    try:
        last = json.loads(records[-1])
    except json.JSONDecodeError as exc:
        raise ValueError("audit stream tail is not valid JSON") from exc
    tip_hash = last.get("_hash")
    if not isinstance(tip_hash, str) or not tip_hash:
        raise ValueError("audit stream tail carries no '_hash' chain tip")
    return _validate_tip_hash(tip_hash), len(records)

seal_audit_log

seal_audit_log(
    path: Path,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> AuditChainSeal

Read an audit JSONL file's chain tip and return a post-quantum seal.

Parameters

path : Path Path to the audit JSONL stream to seal. private_key : cryptography ML-DSA private key The signing key (see :func:signing_key_from_seed). algorithm : str The ML-DSA variant; must match private_key.

Returns

AuditChainSeal The post-quantum seal over the file's chain tip.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def seal_audit_log(
    path: Path,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> AuditChainSeal:
    """Read an audit JSONL file's chain tip and return a post-quantum seal.

    Parameters
    ----------
    path : Path
        Path to the audit JSONL stream to seal.
    private_key : cryptography ML-DSA private key
        The signing key (see :func:`signing_key_from_seed`).
    algorithm : str
        The ML-DSA variant; must match ``private_key``.

    Returns
    -------
    AuditChainSeal
        The post-quantum seal over the file's chain tip.
    """
    tip_hash, record_count = read_audit_chain_tip(path)
    return seal_audit_chain(tip_hash, record_count, private_key, algorithm=algorithm)

verify_audit_log_seal

verify_audit_log_seal(
    path: Path,
    seal: AuditChainSeal,
    trusted_public_key_hex: str,
) -> bool

Verify a seal against an audit file and a trusted public key.

Parameters

path : Path Path to the audit JSONL stream to check. seal : AuditChainSeal The seal previously produced for this log. trusted_public_key_hex : str The hex-encoded raw ML-DSA public key the verifier trusts.

Returns

bool True only if the file's current chain tip and record count match the seal and the signature verifies under the trusted key, so the seal is rejected if the log was altered after sealing.

Source code in src/scpn_phase_orchestrator/runtime/audit_pqc.py

def verify_audit_log_seal(
    path: Path,
    seal: AuditChainSeal,
    trusted_public_key_hex: str,
) -> bool:
    """Verify a seal against an audit file and a trusted public key.

    Parameters
    ----------
    path : Path
        Path to the audit JSONL stream to check.
    seal : AuditChainSeal
        The seal previously produced for this log.
    trusted_public_key_hex : str
        The hex-encoded raw ML-DSA public key the verifier trusts.

    Returns
    -------
    bool
        ``True`` only if the file's current chain tip and record count match the
        seal *and* the signature verifies under the trusted key, so the seal is
        rejected if the log was altered after sealing.
    """
    tip_hash, record_count = read_audit_chain_tip(path)
    if tip_hash != seal.tip_hash or record_count != seal.record_count:
        return False
    return verify_audit_chain_seal(seal, trusted_public_key_hex)

Error handling philosophy

SPO follows a fail-fast strategy at system boundaries:

• Input validation: ValueError for invalid shapes, NaN, etc.
• Engine divergence: EngineError with step number and divergence magnitude
• Binding errors: BindingError with field path and expected type
• Audit tampering: AuditError with chain break location

Internal code trusts validated data — no redundant checks on hot paths. This keeps engine step latency under 1 ms for N ≤ 64.

Role in the production boundary

This module is the project-wide seam between raw exceptions and deterministic control outcomes. Every subsystem raises typed SPO errors into the same hierarchy, so operational handlers can route failures consistently across CLI, services, and embedded drivers.

In practice this gives operators a bounded failure vocabulary:

• Validation faults are recoverable through corrected input or fallback profiles.
• Engine faults can trigger deterministic supervisor actions.
• Audit faults stop replay promotion by default to protect chain integrity.

Why optional dependency handling matters

The documented optional import and HAS_* flags are part of reliability policy, not just packaging convenience. They prevent import-time crashes when one runtime path is unavailable, while keeping production-facing Python behavior explicit:

• Required behaviour remains available on the baseline path.
• Optional accelerators are additive when present.
• Audit and CLI surfaces can still start on minimal environments.

This pattern prevents “one package drift takes down all controls” incidents by making capability and observability dependencies explicit in the import graph.

Enterprise usage note

When teams evaluate SPO for regulated or multi-region deployment, this contract is a critical documentation checkpoint because it documents which failures are expected to halt, which failures are reviewed, and which can be auto-recovered without risking audit continuity.