Skip to content

Assurance Case

The assurance subsystem composes existing SPO runtime evidence — audit-chain integrity, replay determinism, formal verification, twin-confidence scoring, and the conformal admission gate — into a single hash-sealed assurance-case bundle, and maps that evidence to the published clauses of three standards a regulated deployment is commonly measured against:

  • Regulation (EU) 2024/1689 (the EU AI Act) — high-risk requirements;
  • ISO/IEC 42001:2023 — AI management system clauses and Annex A controls;
  • ANSI/UL 4600 — claim-based safety case for autonomous products.

The bundle is review-only: actuation_permitted is always False, the bundle hash seals the evidence and conformance records deterministically, and a disclaimer states that the bundle is a technical evidence-mapping aid, not a legal conformity assessment. Clauses with no contributing technical evidence are recorded as not_addressed so coverage gaps are explicit rather than implied.

spo assurance-case --system my-deployment \
  --audit-log run.jsonl \
  --evidence-file twin_confidence.json \
  --output assurance_bundle.json \
  --report-out conformity_report.md \
  --report-pdf-out conformity_report.pdf

--report-out additionally renders a human-readable Markdown conformity report from the same sealed bundle — a per-standard, clause-by-clause table of conformance status, contributing evidence, and rationale, anchored to the bundle hash for traceability. --report-pdf-out renders the same report as a deterministic, dependency-free text PDF — the distributable artefact an assessor files.

For operator review packages, spo certification-evidence wraps the same assurance bundle with deterministic test vectors and a manifest:

spo certification-evidence --system my-deployment \
  --run-result run_summary.json \
  --output-dir review_package

--run-result takes a serialised SimulationResult summary and auto-derives the run's audit-stream integrity, conformal admission-gate, and closed-loop control-safety-envelope evidence (control mode, applied-action and boundary-violation totals, recorded when the policy feedback was active), so a package can be assembled from a run summary without hand-authoring evidence JSON (--audit-log and --evidence-file remain available and compose with it). With --audit-log, adding --verify-determinism re-executes the logged run and records a replay_determinism evidence item for the reproducibility clauses. --formal-package takes a serialised FormalVerificationPackage manifest (from the supervisor formal exporters) and adds a formal_verification evidence item for the formal-argument clauses, recording which model-checking properties were posed against which exported artefacts. --twin-confidence-file takes a serialised TwinConfidenceScore and adds a twin_confidence evidence item for the drift-monitoring clauses, restating the calibrated confidence, operator status, divergences, and content hash of the scored tick.

With --audit-log, --sign-envelope additionally writes signed_envelope.json — a deterministic binding of the package hash to the run's audit-chain tip, so the package is anchored to a specific, tamper-evident, replayable execution. --signing-seed-file supplies an ML-DSA seed (FIPS 204) and adds a post-quantum seal over that tip to the envelope, making the binding publicly verifiable (it implies --sign-envelope). The ML-DSA seal needs the pqc extra and an OpenSSL 3.5+ backend.

The package directory contains:

  • manifest.json — file digests, the assurance bundle hash, standards covered, coverage summary, package hash, and review-only disclaimers;
  • assurance_bundle.json — the existing scpn_assurance_case_bundle_v1 payload;
  • conformity_report.md — a human-readable, per-standard clause-by-clause conformity report rendered from the bundle and sealed into the manifest digest;
  • conformity_report.pdf — the same conformity report as a deterministic text PDF (the filable artefact), also sealed into the manifest digest;
  • test_vectors.json — recomputable evidence content-hash vectors and clause-rationale hash vectors;
  • signed_envelope.json — (only with --sign-envelope) the package hash bound to the run's audit-chain tip, optionally carrying a post-quantum ML-DSA seal.

The package is standards-shaped evidence for reviewer triage. It does not claim legal compliance, certification, or runtime actuation permission.

Regulatory clause catalogue

scpn_phase_orchestrator.assurance.standards records each referenceable clause with its standard, identifier, official title, and a provenance note. Clause identifiers and titles are taken from the public structure of each standard; the clause text must be confirmed against the official standard before any external submission.

standards

Reference catalogue of regulatory clauses for assurance-case mapping.

The catalogue records the clause identifiers and official titles of the three standards an SPO deployment is most often measured against:

  • Regulation (EU) 2024/1689 (the EU AI Act) — high-risk requirements;
  • ISO/IEC 42001:2023 — AI management system clauses and Annex A controls;
  • ANSI/UL 4600 — claim-based safety case for autonomous products.

Each :class:RegulatoryClause carries a provenance note identifying the source the identifier and title were taken from. The catalogue is a structured reference aid for assembling evidence; it is not a legal interpretation, and clause text must be confirmed against the official standard before any external submission. See :data:REGULATORY_DISCLAIMER.

Classes

RegulatoryClause dataclass 

RegulatoryClause(
    standard: str,
    clause_id: str,
    title: str,
    provenance: str,
)

A single referenceable clause of a regulatory standard.

Parameters

standard: Human-readable standard name (e.g. "EU AI Act 2024/1689"). clause_id: Stable clause identifier within the standard (e.g. "Article 12", "Clause 9", "A.6", "data-integrity"). title: Official clause title. provenance: Note identifying the source of the identifier and title.

Attributes
key property 
key: str

Return a globally unique standard::clause_id key.

Returns

str The composite key.

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

Return a JSON-safe clause record.

Returns

dict[str, object] A JSON-safe clause record.

Source code in src/scpn_phase_orchestrator/assurance/standards.py 
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe clause record.

    Returns
    -------
    dict[str, object]
        A JSON-safe clause record.
    """
    return {
        "standard": self.standard,
        "clause_id": self.clause_id,
        "title": self.title,
        "provenance": self.provenance,
    }

Functions:

clause_catalogue 

clause_catalogue() -> tuple[RegulatoryClause, ...]

Return every catalogued clause across all supported standards.

Returns

tuple[RegulatoryClause, ...] The full clause catalogue.

Source code in src/scpn_phase_orchestrator/assurance/standards.py 
156
157
158
159
160
161
162
163
164
def clause_catalogue() -> tuple[RegulatoryClause, ...]:
    """Return every catalogued clause across all supported standards.

    Returns
    -------
    tuple[RegulatoryClause, ...]
        The full clause catalogue.
    """
    return _ALL_CLAUSES

clause_for_key 

clause_for_key(key: str) -> RegulatoryClause

Return the clause registered under key.

Parameters

key: A standard::clause_id key.

Returns

RegulatoryClause The matching clause.

Raises

KeyError If no clause is registered under key.

Source code in src/scpn_phase_orchestrator/assurance/standards.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def clause_for_key(key: str) -> RegulatoryClause:
    """Return the clause registered under ``key``.

    Parameters
    ----------
    key:
        A ``standard::clause_id`` key.

    Returns
    -------
    RegulatoryClause
        The matching clause.

    Raises
    ------
    KeyError
        If no clause is registered under ``key``.
    """
    try:
        return _CLAUSE_BY_KEY[key]
    except KeyError as exc:
        raise KeyError(f"unknown clause key: {key}") from exc

Evidence items

scpn_phase_orchestrator.assurance.evidence wraps the JSON-safe audit record of an originating surface in a content-addressed EvidenceItem, so the bundle can reference evidence by a stable identifier and detect later mutation. The shared canonical hashing path accepts only strict JSON records: NaN, Infinity, and -Infinity are rejected before any digest is emitted, so hashes remain portable across JSON implementations and non-Python verifiers.

evidence

Typed, content-addressed evidence items for the assurance-case bundle.

An :class:EvidenceItem wraps the JSON-safe audit record produced by an existing SPO surface (audit-chain integrity, replay determinism, formal verification, twin-confidence, the conformal admission gate, or the closed-loop control envelope) together with a content hash, so the bundle can reference evidence by a stable identifier and detect any later mutation.

Classes

EvidenceItem dataclass 

EvidenceItem(
    evidence_id: str,
    category: str,
    summary: str,
    record: Mapping[str, object],
    content_hash: str = "",
)

One content-addressed piece of assurance evidence.

Parameters

evidence_id: Stable identifier, unique within a bundle (e.g. "audit-chain-integrity"). category: One of :data:EVIDENCE_CATEGORIES. summary: Short human-readable description of what the evidence shows. record: The JSON-safe audit record produced by the originating surface. content_hash: SHA-256 of the canonical serialisation of record. Defaults to the computed hash; an explicit mismatching value is rejected.

Methods:
__post_init__ 
__post_init__() -> None

Validate the evidence metadata and compute or verify its content hash.

Source code in src/scpn_phase_orchestrator/assurance/evidence.py 
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
def __post_init__(self) -> None:
    """Validate the evidence metadata and compute or verify its content hash."""
    if not self.evidence_id.strip():
        raise ValueError("evidence_id must be a non-empty string")
    if self.category not in EVIDENCE_CATEGORIES:
        raise ValueError(
            f"category must be one of {sorted(EVIDENCE_CATEGORIES)}, "
            f"got {self.category!r}"
        )
    if not self.summary.strip():
        raise ValueError("summary must be a non-empty string")
    if not isinstance(self.record, Mapping):
        raise ValueError("record must be a mapping")
    computed = canonical_record_hash(dict(self.record))
    if not self.content_hash:
        object.__setattr__(self, "content_hash", computed)
    else:
        require_sha256(self.content_hash, "content_hash")
        if self.content_hash != computed:
            raise ValueError(
                "content_hash does not match the canonical hash of record"
            )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe evidence record.

Returns

dict[str, object] A JSON-safe evidence record.

Source code in src/scpn_phase_orchestrator/assurance/evidence.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe evidence record.

    Returns
    -------
    dict[str, object]
        A JSON-safe evidence record.
    """
    return {
        "evidence_id": self.evidence_id,
        "category": self.category,
        "summary": self.summary,
        "content_hash": self.content_hash,
        "record": dict(self.record),
    }

Functions:

build_evidence_item 

build_evidence_item(
    evidence_id: str,
    category: str,
    summary: str,
    record: Mapping[str, object],
) -> EvidenceItem

Construct an :class:EvidenceItem with a computed content hash.

Parameters

evidence_id: Stable identifier, unique within a bundle. category: One of :data:EVIDENCE_CATEGORIES. summary: Short human-readable description. record: The JSON-safe audit record.

Returns

EvidenceItem The constructed evidence item.

Source code in src/scpn_phase_orchestrator/assurance/evidence.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def build_evidence_item(
    evidence_id: str,
    category: str,
    summary: str,
    record: Mapping[str, object],
) -> EvidenceItem:
    """Construct an :class:`EvidenceItem` with a computed content hash.

    Parameters
    ----------
    evidence_id:
        Stable identifier, unique within a bundle.
    category:
        One of :data:`EVIDENCE_CATEGORIES`.
    summary:
        Short human-readable description.
    record:
        The JSON-safe audit record.

    Returns
    -------
    EvidenceItem
        The constructed evidence item.
    """
    return EvidenceItem(
        evidence_id=evidence_id,
        category=category,
        summary=summary,
        record=dict(record),
    )

Run-derived evidence

scpn_phase_orchestrator.assurance.run_evidence maps the trust-relevant fields of a serialised SimulationResult record — the close-time audit-stream integrity result and the conformal admission-gate decisions — into evidence items. It consumes the JSON-safe record (not the runtime object), so the assurance package stays free of the numeric runtime import chain, and it emits nothing for a surface that did not run.

run_evidence

Derive assurance evidence directly from a serialised simulation run record.

A completed run produces a JSON-safe SimulationResult.to_record() summary. This module maps the trust-relevant fields of that record — the close-time audit event-stream integrity result, the conformal twin-confidence admission-gate decisions, and the closed-loop control-safety envelope (control mode, applied actions, and boundary-violation totals) — into :class:~scpn_phase_orchestrator.assurance.evidence.EvidenceItem records, so a deployment can assemble a conformity package from a run record without hand-authoring evidence JSON.

The helper consumes the serialised record (a Mapping), not the SimulationResult object, so the assurance package stays free of the heavy runtime/numeric import chain and can run against a persisted run summary. Fields that are absent or describe an inactive gate produce no evidence item — the mapping never fabricates evidence for a surface that did not run.

Classes

Functions:

build_run_evidence 

build_run_evidence(
    run_record: Mapping[str, object],
) -> tuple[EvidenceItem, ...]

Build assurance evidence from a serialised simulation run record.

Parameters

run_record: A JSON-safe SimulationResult.to_record() mapping.

Returns

tuple[EvidenceItem, ...] Evidence for the trust surfaces the run attests: the audit event-stream integrity result (audit-chain) when an event stream was written, the conformal admission-gate decisions (conformal-gate) when the gate scored at least one tick, and the closed-loop control-safety envelope (control-envelope) when the policy feedback was active. Empty if the run attests none of them.

Source code in src/scpn_phase_orchestrator/assurance/run_evidence.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def build_run_evidence(run_record: Mapping[str, object]) -> tuple[EvidenceItem, ...]:
    """Build assurance evidence from a serialised simulation run record.

    Parameters
    ----------
    run_record:
        A JSON-safe ``SimulationResult.to_record()`` mapping.

    Returns
    -------
    tuple[EvidenceItem, ...]
        Evidence for the trust surfaces the run attests: the audit event-stream
        integrity result (``audit-chain``) when an event stream was written, the
        conformal admission-gate decisions (``conformal-gate``) when the gate
        scored at least one tick, and the closed-loop control-safety envelope
        (``control-envelope``) when the policy feedback was active. Empty if the
        run attests none of them.
    """
    items: list[EvidenceItem] = []

    integrity = run_record.get("audit_event_stream_integrity")
    if isinstance(integrity, Mapping):
        items.append(
            build_evidence_item(
                evidence_id="run-audit-stream-integrity",
                category=AUDIT_LOGGING,
                summary="Close-time audit event-stream integrity result for the run",
                record=dict(integrity),
            )
        )

    conformal_total = run_record.get("conformal_admission_total")
    if _is_positive_int(conformal_total):
        items.append(
            build_evidence_item(
                evidence_id="run-conformal-admission",
                category=CONFORMAL_GATE,
                summary="Conformal twin-confidence admission decisions for the run",
                record={
                    "conformal_admission_total": conformal_total,
                    "conformal_admission_rejections": run_record.get(
                        "conformal_admission_rejections"
                    ),
                    "last_conformal_admission": run_record.get(
                        "last_conformal_admission"
                    ),
                },
            )
        )

    if run_record.get("policy_enabled") is True:
        items.append(
            build_evidence_item(
                evidence_id="run-control-envelope",
                category=CONTROL_ENVELOPE,
                summary="Closed-loop control-safety envelope for the run",
                record={
                    "control_mode": run_record.get("control_mode"),
                    "policy_enabled": True,
                    "action_total": run_record.get("action_total"),
                    "boundary_violation_total": run_record.get(
                        "boundary_violation_total"
                    ),
                    "final_regime": run_record.get("final_regime"),
                },
            )
        )

    return tuple(items)

Formal-verification evidence

scpn_phase_orchestrator.assurance.formal_evidence maps a serialised FormalVerificationPackage.to_audit_record() manifest — the supervisor formal exporters' artefact hashes, model-checking property library, and non-executing checker commands — into a single formal_verification evidence item. Like the run-derived evidence, it consumes the JSON manifest (not the package object), so the assurance package stays free of the supervisor import chain, and it restates the manifest verbatim: it records which properties were posed against which artefacts, never that any external checker accepted them.

formal_evidence

Derive formal-verification assurance evidence from a verification-package manifest.

The supervisor formal exporters (:mod:scpn_phase_orchestrator.supervisor.formal_export) assemble a deterministic :class:~scpn_phase_orchestrator.supervisor.formal_export.FormalVerificationPackage — exported PRISM/TLA/SMT artefact hashes, the model-checking property library, and the exact (non-executing) checker commands — whose to_audit_record() is a JSON-safe manifest. This module maps that manifest into a single :class:~scpn_phase_orchestrator.assurance.evidence.EvidenceItem in the formal_verification category, so a conformity package can attest the formal argument the supervisor produced.

The helper consumes the serialised manifest (a Mapping), not the FormalVerificationPackage object, mirroring :func:~scpn_phase_orchestrator.assurance.run_evidence.build_run_evidence: the assurance package stays free of the supervisor import chain and can attest a manifest persisted to disk. It restates the manifest verbatim and never fabricates properties or checker results — it records which properties were posed against which artefacts, not that any checker accepted them.

Classes

Functions:

build_formal_verification_evidence 

build_formal_verification_evidence(
    package_record: Mapping[str, object],
) -> EvidenceItem

Build a formal-verification evidence item from a verification-package manifest.

Parameters

package_record: A JSON-safe FormalVerificationPackage.to_audit_record() mapping. It must carry a non-empty package_name and package_hash, a list of properties, and an artifact_hashes mapping.

Returns

EvidenceItem A formal_verification evidence item whose record is the manifest verbatim, summarising how many properties were posed against how many exported artefacts.

Raises

ValueError If the manifest is missing a required field or a field has the wrong type.

Source code in src/scpn_phase_orchestrator/assurance/formal_evidence.py 
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def build_formal_verification_evidence(
    package_record: Mapping[str, object],
) -> EvidenceItem:
    """Build a formal-verification evidence item from a verification-package manifest.

    Parameters
    ----------
    package_record:
        A JSON-safe ``FormalVerificationPackage.to_audit_record()`` mapping. It
        must carry a non-empty ``package_name`` and ``package_hash``, a list of
        ``properties``, and an ``artifact_hashes`` mapping.

    Returns
    -------
    EvidenceItem
        A ``formal_verification`` evidence item whose record is the manifest
        verbatim, summarising how many properties were posed against how many
        exported artefacts.

    Raises
    ------
    ValueError
        If the manifest is missing a required field or a field has the wrong type.
    """
    if not isinstance(package_record, Mapping):
        raise ValueError("formal verification package manifest must be a mapping")

    package_name = _require_non_empty_str(package_record, "package_name")
    _require_non_empty_str(package_record, "package_hash")

    properties = package_record.get("properties")
    if not isinstance(properties, Sequence) or isinstance(properties, str | bytes):
        raise ValueError(
            "formal verification package manifest field 'properties' must be a list"
        )
    artifact_hashes = package_record.get("artifact_hashes")
    if not isinstance(artifact_hashes, Mapping):
        raise ValueError(
            "formal verification package manifest field 'artifact_hashes' "
            "must be a mapping"
        )

    summary = (
        f"Formal verification package {package_name!r}: "
        f"{len(properties)} propert{'y' if len(properties) == 1 else 'ies'} "
        f"over {len(artifact_hashes)} "
        f"artefact{'' if len(artifact_hashes) == 1 else 's'}"
    )
    return build_evidence_item(
        evidence_id="formal-verification-package",
        category=FORMAL_VERIFICATION,
        summary=summary,
        record=dict(package_record),
    )

Twin-confidence evidence

scpn_phase_orchestrator.assurance.twin_confidence_evidence maps a serialised TwinConfidenceScore.to_audit_record() — calibrated confidence, operator status, raw divergences, one-sided z-scores, band flags, backend, and content hash — into a single twin_confidence evidence item, closing the one evidence category the clause map referenced without a producer. Like the run-derived and formal evidence it consumes the JSON record (not the score object) and restates it verbatim, rejecting a confidence outside [0, 1].

twin_confidence_evidence

Derive twin-confidence assurance evidence from a serialised confidence score.

The twin-confidence monitor (:mod:scpn_phase_orchestrator.monitor.twin_confidence) scores each digital-twin tick against a calibrated baseline and produces a :class:~scpn_phase_orchestrator.monitor.twin_confidence.TwinConfidenceScore whose to_audit_record() is a JSON-safe mapping (calibrated confidence, operator status, raw divergences, one-sided z-scores, band flags, backend, and a content hash). This module maps that record into a single :class:~scpn_phase_orchestrator.assurance.evidence.EvidenceItem in the twin_confidence category, so a conformity package can attest the live drift monitoring the deployment ran — the one evidence category the assurance-case clause map references but no producer previously emitted.

The helper consumes the serialised score (a Mapping), not the TwinConfidenceScore object, mirroring :func:~scpn_phase_orchestrator.assurance.formal_evidence.build_formal_verification_evidence and :func:~scpn_phase_orchestrator.assurance.run_evidence.build_run_evidence: the assurance package stays free of the monitor's numeric import chain and can attest a score persisted to disk. It restates the score verbatim and never fabricates a confidence value — a record missing a required field or carrying a confidence outside [0, 1] is rejected rather than coerced.

Classes

Functions:

build_twin_confidence_evidence 

build_twin_confidence_evidence(
    score_record: Mapping[str, object],
) -> EvidenceItem

Build a twin-confidence evidence item from a serialised confidence score.

Parameters

score_record: A JSON-safe TwinConfidenceScore.to_audit_record() mapping. It must carry a confidence in [0, 1], a non-empty status, and a non-empty score_hash.

Returns

EvidenceItem A twin_confidence evidence item whose record is the score verbatim, summarising the operator status and calibrated confidence.

Raises

ValueError If the score is not a mapping, a required field is missing, or a field has the wrong type or an out-of-range value.

Source code in src/scpn_phase_orchestrator/assurance/twin_confidence_evidence.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def build_twin_confidence_evidence(
    score_record: Mapping[str, object],
) -> EvidenceItem:
    """Build a twin-confidence evidence item from a serialised confidence score.

    Parameters
    ----------
    score_record:
        A JSON-safe ``TwinConfidenceScore.to_audit_record()`` mapping. It must
        carry a ``confidence`` in ``[0, 1]``, a non-empty ``status``, and a
        non-empty ``score_hash``.

    Returns
    -------
    EvidenceItem
        A ``twin_confidence`` evidence item whose record is the score verbatim,
        summarising the operator status and calibrated confidence.

    Raises
    ------
    ValueError
        If the score is not a mapping, a required field is missing, or a field has
        the wrong type or an out-of-range value.
    """
    if not isinstance(score_record, Mapping):
        raise ValueError("twin-confidence score must be a mapping")

    confidence = _require_unit_confidence(score_record)
    status = _require_non_empty_str(score_record, "status")
    _require_non_empty_str(score_record, "score_hash")

    summary = (
        f"Twin-confidence score {status!r} at calibrated confidence {confidence:.3f}"
    )
    return build_evidence_item(
        evidence_id="twin-confidence-score",
        category=TWIN_CONFIDENCE,
        summary=summary,
        record=dict(score_record),
    )

Bundle assembly

scpn_phase_orchestrator.assurance.case maps each catalogued clause to the evidence that addresses it, records the conformance status and rationale, and seals the result into a deterministic, fail-closed bundle.

case

Assemble SPO runtime evidence into a hash-sealed assurance-case bundle.

The bundle links each catalogued regulatory clause (:mod:scpn_phase_orchestrator.assurance.standards) to the SPO evidence that addresses it, records the conformance status and rationale, and seals the whole into a deterministic hash. The bundle is review-only: actuation_permitted is always False as documentary review metadata, not as a runtime actuation gate. Live runtime limits remain in the actuation projector and safety-tier checks. The bundle carries the :data:~scpn_phase_orchestrator.assurance.standards.REGULATORY_DISCLAIMER.

Classes

ClauseConformance dataclass 

ClauseConformance(
    clause: RegulatoryClause,
    status: str,
    evidence_ids: tuple[str, ...],
    rationale: str,
)

Conformance status of one clause against the bundle's evidence.

Parameters

clause: The catalogued regulatory clause. status: One of :data:CONFORMANCE_STATUSES. evidence_ids: Evidence identifiers addressing the clause (empty iff not_addressed). rationale: Explanation linking the evidence to the clause.

Methods:
__post_init__ 
__post_init__() -> None

Validate status, evidence references, and rationale text.

Source code in src/scpn_phase_orchestrator/assurance/case.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
def __post_init__(self) -> None:
    """Validate status, evidence references, and rationale text."""
    if self.status not in CONFORMANCE_STATUSES:
        raise ValueError(
            f"status must be one of {sorted(CONFORMANCE_STATUSES)}, "
            f"got {self.status!r}"
        )
    if self.status == NOT_ADDRESSED and self.evidence_ids:
        raise ValueError("not_addressed clauses must carry no evidence_ids")
    if self.status != NOT_ADDRESSED and not self.evidence_ids:
        raise ValueError(
            f"{self.status} clauses must carry at least one evidence_id"
        )
    if not self.rationale.strip():
        raise ValueError("rationale must be a non-empty string")
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe conformance record.

Returns

dict[str, object] A JSON-safe conformance record.

Source code in src/scpn_phase_orchestrator/assurance/case.py 
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe conformance record.

    Returns
    -------
    dict[str, object]
        A JSON-safe conformance record.
    """
    return {
        "clause": self.clause.to_audit_record(),
        "status": self.status,
        "evidence_ids": list(self.evidence_ids),
        "rationale": self.rationale,
    }

AssuranceCaseBundle dataclass 

AssuranceCaseBundle(
    system_name: str,
    version: str,
    evidence: tuple[EvidenceItem, ...],
    conformance: tuple[ClauseConformance, ...],
    standards_covered: tuple[str, ...],
    bundle_hash: str = "",
    schema: str = ASSURANCE_CASE_SCHEMA,
    disclaimer: str = REGULATORY_DISCLAIMER,
    actuation_permitted: bool = False,
)

A hash-sealed, review-only assurance-case evidence bundle.

Parameters

system_name: Name of the system the bundle describes. version: Bundle schema instance version (semantic, e.g. "1.0.0"). evidence: The collected evidence items (unique evidence_id). conformance: Per-clause conformance, one entry per catalogued clause. standards_covered: Standards the clause catalogue spans. bundle_hash: SHA-256 over the canonical bundle seed; recomputed and checked. schema: Schema identifier; fixed to :data:ASSURANCE_CASE_SCHEMA. disclaimer: Regulatory disclaimer; fixed to REGULATORY_DISCLAIMER. actuation_permitted: Always False as a documentary review flag. The live actuation path does not consult assurance bundles; runtime permission remains governed by the safety-tier checks and :class:ActionProjector constraints.

Methods:
__post_init__ 
__post_init__() -> None

Validate bundle invariants and compute or verify the bundle hash.

Source code in src/scpn_phase_orchestrator/assurance/case.py 
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
def __post_init__(self) -> None:
    """Validate bundle invariants and compute or verify the bundle hash."""
    if not self.system_name.strip():
        raise ValueError("system_name must be a non-empty string")
    if not self.version.strip():
        raise ValueError("version must be a non-empty string")
    if self.schema != ASSURANCE_CASE_SCHEMA:
        raise ValueError(f"schema must be {ASSURANCE_CASE_SCHEMA!r}")
    if self.actuation_permitted:
        raise ValueError("assurance bundles are review-only; actuation_permitted")
    ids = [item.evidence_id for item in self.evidence]
    if len(ids) != len(set(ids)):
        raise ValueError("evidence_id values must be unique")
    known = set(ids)
    for entry in self.conformance:
        missing = set(entry.evidence_ids) - known
        if missing:
            raise ValueError(
                f"conformance references unknown evidence_ids: {sorted(missing)}"
            )
    computed = canonical_record_hash(self._seed())
    if not self.bundle_hash:
        object.__setattr__(self, "bundle_hash", computed)
    else:
        require_sha256(self.bundle_hash, "bundle_hash")
        if self.bundle_hash != computed:
            raise ValueError("bundle_hash does not match the canonical bundle seed")
coverage_summary 
coverage_summary() -> dict[str, dict[str, int]]

Return per-standard counts of clause conformance statuses.

Returns

dict[str, dict[str, int]] Maps each standard to addressed / partially_addressed / not_addressed / total counts.

Source code in src/scpn_phase_orchestrator/assurance/case.py 
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
def coverage_summary(self) -> dict[str, dict[str, int]]:
    """Return per-standard counts of clause conformance statuses.

    Returns
    -------
    dict[str, dict[str, int]]
        Maps each standard to ``addressed`` / ``partially_addressed`` /
        ``not_addressed`` / ``total`` counts.
    """
    summary: dict[str, dict[str, int]] = {}
    for entry in self.conformance:
        bucket = summary.setdefault(
            entry.clause.standard,
            {ADDRESSED: 0, PARTIALLY_ADDRESSED: 0, NOT_ADDRESSED: 0, "total": 0},
        )
        bucket[entry.status] += 1
        bucket["total"] += 1
    return summary
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe bundle record.

Returns

dict[str, object] A JSON-safe bundle record.

Source code in src/scpn_phase_orchestrator/assurance/case.py 
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe bundle record.

    Returns
    -------
    dict[str, object]
        A JSON-safe bundle record.
    """
    return {
        "schema": self.schema,
        "version": self.version,
        "system_name": self.system_name,
        "standards_covered": sorted(self.standards_covered),
        "disclaimer": self.disclaimer,
        "actuation_permitted": self.actuation_permitted,
        "bundle_hash": self.bundle_hash,
        "coverage_summary": self.coverage_summary(),
        "evidence": [item.to_audit_record() for item in self.evidence],
        "conformance": [entry.to_audit_record() for entry in self.conformance],
    }

Functions:

build_assurance_case_bundle 

build_assurance_case_bundle(
    system_name: str,
    evidence: Sequence[EvidenceItem],
    *,
    version: str = "1.0.0",
) -> AssuranceCaseBundle

Assemble an assurance-case bundle from collected evidence.

Each catalogued clause is mapped to the evidence addressing it via :data:DEFAULT_EVIDENCE_CLAUSE_MAP; clauses with no contributing evidence are recorded as not_addressed so gaps are explicit.

Parameters

system_name: Name of the system the bundle describes. evidence: The evidence items to include (unique evidence_id). version: Bundle instance version.

Returns

AssuranceCaseBundle The sealed bundle.

Raises

ValueError If evidence_id values are not unique.

Source code in src/scpn_phase_orchestrator/assurance/case.py 
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
def build_assurance_case_bundle(
    system_name: str,
    evidence: Sequence[EvidenceItem],
    *,
    version: str = "1.0.0",
) -> AssuranceCaseBundle:
    """Assemble an assurance-case bundle from collected evidence.

    Each catalogued clause is mapped to the evidence addressing it via
    :data:`DEFAULT_EVIDENCE_CLAUSE_MAP`; clauses with no contributing evidence
    are recorded as ``not_addressed`` so gaps are explicit.

    Parameters
    ----------
    system_name:
        Name of the system the bundle describes.
    evidence:
        The evidence items to include (unique ``evidence_id``).
    version:
        Bundle instance version.

    Returns
    -------
    AssuranceCaseBundle
        The sealed bundle.

    Raises
    ------
    ValueError
        If ``evidence_id`` values are not unique.
    """
    ids = [item.evidence_id for item in evidence]
    if len(ids) != len(set(ids)):
        raise ValueError("evidence_id values must be unique")
    category_evidence: dict[str, list[str]] = {}
    for item in evidence:
        category_evidence.setdefault(item.category, []).append(item.evidence_id)
    conformance = tuple(
        _conformance_for_clause(clause, category_evidence)
        for clause in clause_catalogue()
    )
    standards_covered = tuple(
        sorted({clause.standard for clause in clause_catalogue()})
    )
    return AssuranceCaseBundle(
        system_name=system_name,
        version=version,
        evidence=tuple(evidence),
        conformance=conformance,
        standards_covered=standards_covered,
    )

Certification evidence package

scpn_phase_orchestrator.assurance.certification assembles the review package around the assurance-case bundle. It keeps package assembly deterministic and hash-sealed while preserving the same review-only boundary as the underlying assurance case.

certification

Assemble review packages from assurance-case evidence bundles.

The package writer is deliberately narrow: it wraps the existing assurance-case bundle with deterministic hash test vectors and a manifest that seals every emitted file. The output is a technical evidence package for human review, not a certification claim or live actuation gate.

Classes

CertificationEvidencePackage dataclass 

CertificationEvidencePackage(
    assurance_bundle: AssuranceCaseBundle,
    test_vectors: Mapping[str, object],
    manifest: Mapping[str, object],
    file_contents: Mapping[str, bytes],
)

A deterministic standards-shaped review package.

Parameters

assurance_bundle: The underlying hash-sealed assurance-case bundle. test_vectors: JSON-safe deterministic vectors that let reviewers recompute evidence hashes and clause-conformance rationale hashes. manifest: JSON-safe package manifest containing file digests and package digest. file_contents: Mapping from relative package paths to deterministic file bytes. Text artefacts are UTF-8 encoded; conformity_report.pdf is raw PDF bytes.

Methods:
to_files 
to_files() -> dict[str, bytes]

Return package files keyed by relative path.

Returns

dict[str, bytes] The package file bytes, including manifest.json and the rendered conformity_report.pdf.

Source code in src/scpn_phase_orchestrator/assurance/certification.py 
83
84
85
86
87
88
89
90
91
92
def to_files(self) -> dict[str, bytes]:
    """Return package files keyed by relative path.

    Returns
    -------
    dict[str, bytes]
        The package file bytes, including ``manifest.json`` and the rendered
        ``conformity_report.pdf``.
    """
    return dict(self.file_contents)

Functions:

build_certification_evidence_package 

build_certification_evidence_package(
    system_name: str,
    evidence: Sequence[EvidenceItem],
    *,
    version: str = "1.0.0",
) -> CertificationEvidencePackage

Build a deterministic review package from assurance evidence.

Parameters

system_name: Name of the reviewed system. evidence: Assurance evidence items to include. version: Package schema instance version.

Returns

CertificationEvidencePackage The assembled package with deterministic file bytes (JSON, Markdown, and the rendered conformity-report PDF).

Source code in src/scpn_phase_orchestrator/assurance/certification.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def build_certification_evidence_package(
    system_name: str,
    evidence: Sequence[EvidenceItem],
    *,
    version: str = "1.0.0",
) -> CertificationEvidencePackage:
    """Build a deterministic review package from assurance evidence.

    Parameters
    ----------
    system_name:
        Name of the reviewed system.
    evidence:
        Assurance evidence items to include.
    version:
        Package schema instance version.

    Returns
    -------
    CertificationEvidencePackage
        The assembled package with deterministic file bytes (JSON, Markdown, and
        the rendered conformity-report PDF).
    """
    bundle = build_assurance_case_bundle(system_name, evidence, version=version)
    bundle_payload = _dump_json(bundle.to_audit_record())
    test_vectors = _build_test_vectors(bundle)
    vector_payload = _dump_json(test_vectors)
    report_payload = render_conformity_report(bundle)
    report_pdf = render_conformity_report_pdf(bundle)
    file_rows = [
        {
            "path": "assurance_bundle.json",
            "sha256": _sha256_text(bundle_payload),
            "bytes": len(bundle_payload.encode("utf-8")),
        },
        {
            "path": "conformity_report.md",
            "sha256": _sha256_text(report_payload),
            "bytes": len(report_payload.encode("utf-8")),
        },
        {
            "path": "conformity_report.pdf",
            "sha256": _sha256_bytes(report_pdf),
            "bytes": len(report_pdf),
        },
        {
            "path": "test_vectors.json",
            "sha256": _sha256_text(vector_payload),
            "bytes": len(vector_payload.encode("utf-8")),
        },
    ]
    package_seed: dict[str, object] = {
        "schema": CERTIFICATION_EVIDENCE_PACKAGE_SCHEMA,
        "version": version,
        "system_name": system_name,
        "assurance_bundle_hash": bundle.bundle_hash,
        "files": file_rows,
    }
    manifest: dict[str, object] = {
        **package_seed,
        "standards_covered": list(bundle.standards_covered),
        "coverage_summary": bundle.coverage_summary(),
        "disclaimer": CERTIFICATION_EVIDENCE_PACKAGE_DISCLAIMER,
        "assurance_disclaimer": REGULATORY_DISCLAIMER,
        "package_hash": canonical_record_hash(package_seed),
    }
    return CertificationEvidencePackage(
        assurance_bundle=bundle,
        test_vectors=test_vectors,
        manifest=manifest,
        file_contents={
            "assurance_bundle.json": bundle_payload.encode("utf-8"),
            "conformity_report.md": report_payload.encode("utf-8"),
            "conformity_report.pdf": report_pdf,
            "test_vectors.json": vector_payload.encode("utf-8"),
            "manifest.json": _dump_json(manifest).encode("utf-8"),
        },
    )

Signed certification envelope

scpn_phase_orchestrator.assurance.envelope binds a certification package to the run that produced it. A SignedCertificationEnvelope commits, in one deterministic record, to the package hash, the run's audit-chain tip (the SHA-256 commitment to the whole audit log, so the package is anchored to a specific, replayable, tamper- evident execution), and an optional post-quantum seal over that tip (scpn_phase_orchestrator.runtime.audit_pqc.AuditChainSeal, ML-DSA / FIPS 204). It reuses the audit seal verbatim and performs no signing or log reading itself — the CLI layer reads the tip and produces the seal — so the assurance leaf only validates and binds. verify_signed_certification_envelope re-derives the envelope hash, checks the package binding, and verifies any attached seal against a trusted public key.

envelope

Bind a certification package to the run that produced it, optionally PQC-signed.

A :class:~scpn_phase_orchestrator.assurance.certification.CertificationEvidencePackage is hash-sealed but free-floating: its package_hash proves the package's own contents are consistent, yet nothing ties it to the specific run whose evidence it carries. This module adds that outer binding.

A :class:SignedCertificationEnvelope commits, in one deterministic record, to:

  • the package's package_hash (which run-evidence the package describes);
  • the audit-chain tip of the run — the SHA-256 commitment to the whole audit log (_hash of the last record), so the envelope is anchored to a specific, tamper-evident execution that can be replayed and re-verified; and
  • an optional post-quantum seal over that tip (:class:~scpn_phase_orchestrator.runtime.audit_pqc.AuditChainSeal, ML-DSA / FIPS 204), making the binding publicly verifiable long after the run.

The envelope reuses the audit seal verbatim — the seal genuinely commits to an audit-chain tip under its own domain, so there is no cross-protocol confusion: the envelope merely records that the package describes that sealed run. It performs no signing or log reading itself (the CLI layer reads the tip and produces the seal with a signing key); it validates the pieces, requires any attached seal to commit to the same tip and record count, and seals the binding with a deterministic envelope_hash. Verification re-derives the hash, checks the package binding, and — when a seal is present — verifies it against a trusted public key.

Classes

SignedCertificationEnvelope dataclass 

SignedCertificationEnvelope(
    package_hash: str,
    audit_chain_tip: str,
    audit_record_count: int,
    seal: Mapping[str, str | int] | None,
    envelope_hash: str = "",
)

A deterministic binding of a certification package to its run.

Attributes

package_hash: The package_hash of the bound certification package (SHA-256 hex). audit_chain_tip: The run's audit-chain tip — _hash of the last audit record, the SHA-256 commitment to the whole log (32-byte digest, hex). audit_record_count: The number of records in the sealed audit chain. seal: The post-quantum seal over the tip as a JSON-safe mapping (:meth:~scpn_phase_orchestrator.runtime.audit_pqc.AuditChainSeal.to_dict), or None for an unsigned (anchor-only) envelope. envelope_hash: SHA-256 over the canonical serialisation of the schema, package hash, audit tip, record count, and seal. Defaults to the computed hash; an explicit mismatching value is rejected.

Methods:
__post_init__ 
__post_init__() -> None

Validate the fields and compute or check the sealing envelope_hash.

Source code in src/scpn_phase_orchestrator/assurance/envelope.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def __post_init__(self) -> None:
    """Validate the fields and compute or check the sealing ``envelope_hash``."""
    require_sha256(self.package_hash, "package_hash")
    require_sha256(self.audit_chain_tip, "audit_chain_tip")
    _validate_record_count(self.audit_record_count)
    computed = canonical_record_hash(self._sealable())
    if not self.envelope_hash:
        object.__setattr__(self, "envelope_hash", computed)
    else:
        require_sha256(self.envelope_hash, "envelope_hash")
        if self.envelope_hash != computed:
            raise ValueError(
                "envelope_hash does not match the canonical hash of the envelope"
            )
to_record 
to_record() -> dict[str, object]

Return a JSON-safe record of the envelope.

Returns

dict[str, object] The schema tag, package hash, audit anchor, optional seal, and the sealing envelope_hash.

Source code in src/scpn_phase_orchestrator/assurance/envelope.py 
116
117
118
119
120
121
122
123
124
125
126
127
def to_record(self) -> dict[str, object]:
    """Return a JSON-safe record of the envelope.

    Returns
    -------
    dict[str, object]
        The schema tag, package hash, audit anchor, optional seal, and the
        sealing ``envelope_hash``.
    """
    record = self._sealable()
    record["envelope_hash"] = self.envelope_hash
    return record

Functions:

build_signed_certification_envelope 

build_signed_certification_envelope(
    package_hash: str,
    audit_chain_tip: str,
    audit_record_count: int,
    *,
    seal: AuditChainSeal | None = None,
) -> SignedCertificationEnvelope

Bind a certification package hash to a run's audit-chain tip.

Parameters

package_hash: The package_hash of the certification package to anchor (SHA-256 hex). audit_chain_tip: The run's audit-chain tip hash (32-byte SHA-256 digest, hex). audit_record_count: The number of records in the sealed audit chain. seal: An optional post-quantum seal over the same tip. When given, it must commit to audit_chain_tip and audit_record_count.

Returns

SignedCertificationEnvelope The deterministic, optionally signed binding.

Raises

ValueError If a field is malformed or a supplied seal commits to a different tip or record count than the anchor.

Source code in src/scpn_phase_orchestrator/assurance/envelope.py 
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def build_signed_certification_envelope(
    package_hash: str,
    audit_chain_tip: str,
    audit_record_count: int,
    *,
    seal: AuditChainSeal | None = None,
) -> SignedCertificationEnvelope:
    """Bind a certification package hash to a run's audit-chain tip.

    Parameters
    ----------
    package_hash:
        The ``package_hash`` of the certification package to anchor (SHA-256 hex).
    audit_chain_tip:
        The run's audit-chain tip hash (32-byte SHA-256 digest, hex).
    audit_record_count:
        The number of records in the sealed audit chain.
    seal:
        An optional post-quantum seal over the same tip. When given, it must
        commit to ``audit_chain_tip`` and ``audit_record_count``.

    Returns
    -------
    SignedCertificationEnvelope
        The deterministic, optionally signed binding.

    Raises
    ------
    ValueError
        If a field is malformed or a supplied seal commits to a different tip or
        record count than the anchor.
    """
    require_sha256(package_hash, "package_hash")
    require_sha256(audit_chain_tip, "audit_chain_tip")
    _validate_record_count(audit_record_count)
    seal_record: Mapping[str, str | int] | None = None
    if seal is not None:
        if seal.tip_hash != audit_chain_tip:
            raise ValueError("seal tip_hash does not match the anchor audit_chain_tip")
        if seal.record_count != audit_record_count:
            raise ValueError(
                "seal record_count does not match the anchor audit_record_count"
            )
        seal_record = seal.to_dict()
    return SignedCertificationEnvelope(
        package_hash=package_hash,
        audit_chain_tip=audit_chain_tip,
        audit_record_count=audit_record_count,
        seal=seal_record,
    )

verify_signed_certification_envelope 

verify_signed_certification_envelope(
    envelope: SignedCertificationEnvelope,
    *,
    package_hash: str,
    trusted_public_key_hex: str | None = None,
) -> bool

Verify an envelope binds a package and, if signed, carries a valid seal.

Parameters

envelope: The envelope to verify. package_hash: The package_hash of the package the envelope is expected to bind. The envelope is rejected if it anchors a different package. trusted_public_key_hex: The hex-encoded raw ML-DSA public key the verifier trusts. Required when the envelope carries a seal; ignored for an anchor-only envelope.

Returns

bool True only if the envelope hash re-derives, the bound package hash matches, and any attached seal verifies under the trusted key for the anchored tip. False otherwise (including a sealed envelope verified without a trusted key).

Source code in src/scpn_phase_orchestrator/assurance/envelope.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def verify_signed_certification_envelope(
    envelope: SignedCertificationEnvelope,
    *,
    package_hash: str,
    trusted_public_key_hex: str | None = None,
) -> bool:
    """Verify an envelope binds a package and, if signed, carries a valid seal.

    Parameters
    ----------
    envelope:
        The envelope to verify.
    package_hash:
        The ``package_hash`` of the package the envelope is expected to bind. The
        envelope is rejected if it anchors a different package.
    trusted_public_key_hex:
        The hex-encoded raw ML-DSA public key the verifier trusts. Required when
        the envelope carries a seal; ignored for an anchor-only envelope.

    Returns
    -------
    bool
        ``True`` only if the envelope hash re-derives, the bound package hash
        matches, and any attached seal verifies under the trusted key for the
        anchored tip. ``False`` otherwise (including a sealed envelope verified
        without a trusted key).
    """
    if envelope.envelope_hash != canonical_record_hash(envelope._sealable()):
        return False
    if envelope.package_hash != package_hash:
        return False
    if envelope.seal is None:
        return True
    if trusted_public_key_hex is None:
        return False
    seal = AuditChainSeal.from_dict(dict(envelope.seal))
    if seal.tip_hash != envelope.audit_chain_tip:
        return False
    if seal.record_count != envelope.audit_record_count:
        return False
    return verify_audit_chain_seal(seal, trusted_public_key_hex)

Supply-chain provenance (SLSA / DSSE)

The certification envelope attests to a run; the provenance layer attests to a build — which release artefacts were produced, from which resolved inputs, by which builder. scpn_phase_orchestrator.assurance.provenance assembles a deterministic in-toto Statement v1 carrying a SLSA provenance v1 predicate: the produced artefacts as digest-pinned subjects, the build definition (build type, external parameters, digest-pinned resolved dependencies), and the run details (builder identity and invocation). The run details also carry the optional builder.version map, the builder's own digest-pinned builderDependencies, and the build byproducts (for example a digest-pinned SBOM); each is omitted when empty, so a minimal statement is byte-identical to one without them. pypi_resolved_dependency turns a hash-pinned lock-file entry into a Package-URL-addressed resolved dependency, so the resolvedDependencies block can carry the full dependency tree. It reads no wall clock and makes no network call, so the same build inputs always serialise to the same statement.

scpn_phase_orchestrator.assurance.dsse wraps that statement in a DSSE envelope — the wire format cosign attest produces — and signs its pre-authentication encoding with ML-DSA (FIPS 204), reusing the single post-quantum primitive in scpn_phase_orchestrator.runtime.audit_pqc. Each signature records its algorithm so a second scheme can be added without breaking existing envelopes; SLH-DSA (FIPS 205 / SPHINCS+) is the reserved hash-based alternative and is added once the cryptography backend ships it. Verification is offline and self-contained: the verifier supplies the trusted public key, whose short id must match the signature.

spo provenance-attest build_provenance.json \
  --signing-seed-file signing.seed > attestation.json

spo provenance-verify attestation.json --public-key-file signer.pub

Signing needs the pqc extra and an OpenSSL 3.5+ backend. Publishing the envelope to a Rekor transparency log or verifying it with cosign is an optional operator step that needs network and OIDC, and is left to the operator; the envelope itself is deterministic and verifiable without either.

The release workflow (.github/workflows/release.yml) wires this into the build: after building the sdist and SBOM it runs tools/build_release_provenance_spec.py to assemble the spec — the sdist as a subject, the SBOM as a byproduct, the hash-pinned lock files as resolved dependencies, and the tag, commit, and runner metadata as the build definition and run details — then signs it with spo provenance-attest using the SPO_PROVENANCE_SIGNING_SEED repository secret, and attaches provenance_attestation.json and provenance_signing_key.pub to the GitHub Release. The seed is written to a private file, used, and deleted within the step; it is never committed. When the secret is not configured the step is skipped and the release still carries GitHub's own keyless build-provenance attestation. Consumers should obtain the public key from a trusted channel before pinning it.

provenance

Build a deterministic SLSA v1 provenance statement for released artefacts.

The audit-chain seal (:mod:scpn_phase_orchestrator.runtime.audit_pqc) and the certification envelope (:mod:scpn_phase_orchestrator.assurance.envelope) both attest to a run: what the software did once it executed. This module attests to the build: which artefacts were produced, from which resolved inputs, by which builder. That is the supply-chain provenance a downstream consumer needs to answer "is this wheel the one that came out of the declared build, and nothing else?".

The output is an in-toto Statement v1 <https://in-toto.io/Statement/v1> carrying a SLSA provenance v1 <https://slsa.dev/provenance/v1> predicate:

  • subject — the produced artefacts, each pinned by SHA-256 digest;
  • predicate.buildDefinition — the build type, the external parameters that drove it, optional internal parameters, and the resolved dependencies (source commit, toolchain), each itself digest-pinned;
  • predicate.runDetails — the builder identity and the invocation metadata.

Everything is derived from caller-supplied values only — there are no wall-clock reads, environment probes, or network calls — so the same build inputs always serialise to the same statement and the same :func:provenance_statement_hash. Signing lives in :mod:scpn_phase_orchestrator.assurance.dsse; this module produces the payload that gets signed. The statement makes no live-actuation or conformity claim: it is a factual record of a build, meeting the content obligation of SLSA Build Level 2 (signed provenance describing the build), with the signing and provenance-generation obligations met by the DSSE layer and the hosting build service respectively.

Classes

ArtifactSubject dataclass 

ArtifactSubject(name: str, sha256: str)

A produced artefact pinned by its SHA-256 digest.

Attributes

name: The artefact name (e.g. the wheel filename), non-empty. sha256: The artefact's SHA-256 digest, lowercase hex (64 characters).

Methods:
__post_init__ 
__post_init__() -> None

Validate the subject name and digest.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
124
125
126
127
def __post_init__(self) -> None:
    """Validate the subject name and digest."""
    _require_non_empty_str(self.name, "subject name")
    require_sha256(self.sha256, "subject sha256")
to_dict 
to_dict() -> dict[str, object]

Return the in-toto subject mapping (name + digest.sha256).

Returns

dict[str, object] The {"name": …, "digest": {"sha256": …}} subject entry.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
129
130
131
132
133
134
135
136
137
def to_dict(self) -> dict[str, object]:
    """Return the in-toto subject mapping (``name`` + ``digest.sha256``).

    Returns
    -------
    dict[str, object]
        The ``{"name": …, "digest": {"sha256": …}}`` subject entry.
    """
    return {"name": self.name, "digest": {"sha256": self.sha256}}

ResourceDescriptor dataclass 

ResourceDescriptor(uri: str, sha256: str, name: str = '')

A resolved build input pinned by digest (source commit, toolchain, dependency).

Attributes

uri: The resource locator (e.g. git+https://…@<commit> or a package URL), non-empty. sha256: The resource's SHA-256 digest, lowercase hex (64 characters). name: An optional human-readable name; the empty string omits it from the output.

Methods:
__post_init__ 
__post_init__() -> None

Validate the resource locator and digest.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
159
160
161
162
def __post_init__(self) -> None:
    """Validate the resource locator and digest."""
    _require_non_empty_str(self.uri, "resolved dependency uri")
    require_sha256(self.sha256, "resolved dependency sha256")
to_dict 
to_dict() -> dict[str, object]

Return the in-toto resource-descriptor mapping.

Returns

dict[str, object] The uri + digest.sha256 mapping, carrying name when set.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def to_dict(self) -> dict[str, object]:
    """Return the in-toto resource-descriptor mapping.

    Returns
    -------
    dict[str, object]
        The ``uri`` + ``digest.sha256`` mapping, carrying ``name`` when set.
    """
    descriptor: dict[str, object] = {
        "uri": self.uri,
        "digest": {"sha256": self.sha256},
    }
    if self.name:
        descriptor["name"] = self.name
    return descriptor

BuildDefinition dataclass 

BuildDefinition(
    build_type: str,
    external_parameters: Mapping[str, object],
    internal_parameters: Mapping[str, object] = dict(),
    resolved_dependencies: tuple[
        ResourceDescriptor, ...
    ] = (),
)

The reproducible definition of how the artefacts were built.

Attributes

build_type: A URI naming the build type / recipe convention, non-empty. external_parameters: The externally supplied parameters that fully drove the build (JSON object). internal_parameters: Builder-internal parameters, defaulting to an empty object. resolved_dependencies: The digest-pinned inputs the build resolved (source, toolchain, deps).

Methods:
__post_init__ 
__post_init__() -> None

Validate the build type and parameter blocks.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
202
203
204
205
206
def __post_init__(self) -> None:
    """Validate the build type and parameter blocks."""
    _require_non_empty_str(self.build_type, "build_type")
    _require_json_mapping(self.external_parameters, "external_parameters")
    _require_json_mapping(self.internal_parameters, "internal_parameters")
to_dict 
to_dict() -> dict[str, object]

Return the SLSA buildDefinition mapping.

resolvedDependencies is sorted by (uri, name) so the same set of inputs always serialises identically. internalParameters is omitted when empty to keep the statement minimal.

Returns

dict[str, object] The buildType / externalParameters / resolvedDependencies mapping, carrying internalParameters when non-empty.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
def to_dict(self) -> dict[str, object]:
    """Return the SLSA ``buildDefinition`` mapping.

    ``resolvedDependencies`` is sorted by ``(uri, name)`` so the same set of
    inputs always serialises identically. ``internalParameters`` is omitted when
    empty to keep the statement minimal.

    Returns
    -------
    dict[str, object]
        The ``buildType`` / ``externalParameters`` / ``resolvedDependencies``
        mapping, carrying ``internalParameters`` when non-empty.
    """
    definition: dict[str, object] = {
        "buildType": self.build_type,
        "externalParameters": _require_json_mapping(
            self.external_parameters, "external_parameters"
        ),
    }
    internal = _require_json_mapping(
        self.internal_parameters, "internal_parameters"
    )
    if internal:
        definition["internalParameters"] = internal
    definition["resolvedDependencies"] = _descriptor_dicts(
        self.resolved_dependencies
    )
    return definition

RunDetails dataclass 

RunDetails(
    builder_id: str,
    invocation_id: str,
    started_on: str = "",
    finished_on: str = "",
    builder_version: Mapping[str, str] = dict(),
    builder_dependencies: tuple[
        ResourceDescriptor, ...
    ] = (),
    byproducts: tuple[ResourceDescriptor, ...] = (),
)

The identity of the builder and the invocation that produced the artefacts.

Attributes

builder_id: A URI identifying the build platform / builder, non-empty. invocation_id: A stable identifier for this build invocation, non-empty. started_on: Optional RFC 3339 build-start timestamp; the empty string omits it. Supplied by the caller (never read from the wall clock) to preserve determinism. finished_on: Optional RFC 3339 build-finish timestamp; the empty string omits it. builder_version: Optional str-to-str map of the builder's own component versions (e.g. the runner image and toolchain versions); omitted when empty. builder_dependencies: The builder's own digest-pinned dependencies — the actions, images, and toolchains the build platform itself resolved, distinct from the sources the build consumed. Omitted when empty. byproducts: Digest-pinned artefacts the build produced that are not release subjects — the SBOM, build logs, or intermediate manifests. Omitted when empty.

Methods:
__post_init__ 
__post_init__() -> None

Validate the builder identity, invocation id, and version map.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
273
274
275
276
277
def __post_init__(self) -> None:
    """Validate the builder identity, invocation id, and version map."""
    _require_non_empty_str(self.builder_id, "builder_id")
    _require_non_empty_str(self.invocation_id, "invocation_id")
    _require_str_mapping(self.builder_version, "builder_version")
to_dict 
to_dict() -> dict[str, object]

Return the SLSA runDetails mapping.

The builder block always carries its id and adds version and builderDependencies only when supplied. The metadata block always carries invocationId and adds startedOn / finishedOn only when the caller supplied them. byproducts is added to runDetails only when non-empty, so a statement without them is byte-identical to the prior format.

Returns

dict[str, object] The builder + metadata mapping, carrying byproducts when non-empty.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def to_dict(self) -> dict[str, object]:
    """Return the SLSA ``runDetails`` mapping.

    The ``builder`` block always carries its ``id`` and adds ``version`` and
    ``builderDependencies`` only when supplied. The ``metadata`` block always
    carries ``invocationId`` and adds ``startedOn`` / ``finishedOn`` only when
    the caller supplied them. ``byproducts`` is added to ``runDetails`` only when
    non-empty, so a statement without them is byte-identical to the prior format.

    Returns
    -------
    dict[str, object]
        The ``builder`` + ``metadata`` mapping, carrying ``byproducts`` when
        non-empty.
    """
    builder: dict[str, object] = {"id": self.builder_id}
    version = _require_str_mapping(self.builder_version, "builder_version")
    if version:
        builder["version"] = version
    if self.builder_dependencies:
        builder["builderDependencies"] = _descriptor_dicts(
            self.builder_dependencies
        )
    metadata: dict[str, object] = {"invocationId": self.invocation_id}
    if self.started_on:
        metadata["startedOn"] = self.started_on
    if self.finished_on:
        metadata["finishedOn"] = self.finished_on
    run_details: dict[str, object] = {"builder": builder, "metadata": metadata}
    if self.byproducts:
        run_details["byproducts"] = _descriptor_dicts(self.byproducts)
    return run_details

SlsaProvenanceStatement dataclass 

SlsaProvenanceStatement(
    subjects: tuple[ArtifactSubject, ...],
    build_definition: BuildDefinition,
    run_details: RunDetails,
)

A complete in-toto Statement v1 carrying a SLSA provenance v1 predicate.

Attributes

subjects: The produced artefacts, each digest-pinned; must be non-empty. build_definition: How the artefacts were built. run_details: Who built them and under which invocation.

Methods:
__post_init__ 
__post_init__() -> None

Reject an empty subject list; the sub-objects self-validate.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
331
332
333
334
def __post_init__(self) -> None:
    """Reject an empty subject list; the sub-objects self-validate."""
    if not self.subjects:
        raise ValueError("provenance statement requires at least one subject")
to_statement 
to_statement() -> dict[str, object]

Return the JSON-safe in-toto Statement.

subject is sorted by artefact name so a given set of artefacts always serialises identically.

Returns

dict[str, object] The _type / subject / predicateType / predicate mapping, ready to canonicalise, hash, and wrap in a DSSE envelope.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
def to_statement(self) -> dict[str, object]:
    """Return the JSON-safe in-toto Statement.

    ``subject`` is sorted by artefact name so a given set of artefacts always
    serialises identically.

    Returns
    -------
    dict[str, object]
        The ``_type`` / ``subject`` / ``predicateType`` / ``predicate`` mapping,
        ready to canonicalise, hash, and wrap in a DSSE envelope.
    """
    return {
        "_type": IN_TOTO_STATEMENT_TYPE,
        "subject": [
            subject.to_dict()
            for subject in sorted(self.subjects, key=lambda item: item.name)
        ],
        "predicateType": SLSA_PROVENANCE_PREDICATE_TYPE,
        "predicate": {
            "buildDefinition": self.build_definition.to_dict(),
            "runDetails": self.run_details.to_dict(),
        },
    }
statement_hash 
statement_hash() -> str

Return the SHA-256 of the canonical statement serialisation.

Returns

str Lowercase hexadecimal SHA-256 over the canonical statement JSON.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
361
362
363
364
365
366
367
368
369
def statement_hash(self) -> str:
    """Return the SHA-256 of the canonical statement serialisation.

    Returns
    -------
    str
        Lowercase hexadecimal SHA-256 over the canonical statement JSON.
    """
    return canonical_record_hash(self.to_statement())

Functions:

build_slsa_provenance_statement 

build_slsa_provenance_statement(
    subjects: tuple[ArtifactSubject, ...],
    build_definition: BuildDefinition,
    run_details: RunDetails,
) -> SlsaProvenanceStatement

Assemble a validated SLSA provenance statement.

Parameters

subjects: The produced artefacts (non-empty), each digest-pinned. build_definition: The reproducible build definition. run_details: The builder identity and invocation metadata.

Returns

SlsaProvenanceStatement The validated statement.

Raises

ValueError If the subject list is empty or any field is malformed.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
def build_slsa_provenance_statement(
    subjects: tuple[ArtifactSubject, ...],
    build_definition: BuildDefinition,
    run_details: RunDetails,
) -> SlsaProvenanceStatement:
    """Assemble a validated SLSA provenance statement.

    Parameters
    ----------
    subjects:
        The produced artefacts (non-empty), each digest-pinned.
    build_definition:
        The reproducible build definition.
    run_details:
        The builder identity and invocation metadata.

    Returns
    -------
    SlsaProvenanceStatement
        The validated statement.

    Raises
    ------
    ValueError
        If the subject list is empty or any field is malformed.
    """
    return SlsaProvenanceStatement(
        subjects=subjects,
        build_definition=build_definition,
        run_details=run_details,
    )

pypi_resolved_dependency 

pypi_resolved_dependency(
    name: str, version: str, sha256: str
) -> ResourceDescriptor

Return a digest-pinned resolved dependency for a PyPI package.

Turns one hash-pinned lock-file entry into a :class:ResourceDescriptor whose uri is a Package URL <https://github.com/package-url/purl-spec>_ (pkg:pypi/<name>@<version>). The package name is normalised to the PyPI form (lowercased, runs of ./-/_ collapsed to a single -, per PEP 503) for the URL, while the original name is preserved in the descriptor name so the resolved-dependency tree records exactly what the lock pinned. Assembling one descriptor per pinned artefact yields the fuller resolvedDependencies block a consumer needs to reproduce the build's inputs.

Parameters

name: The package name as written in the lock file, non-empty. version: The pinned package version, non-empty. sha256: The pinned artefact's SHA-256 digest, lowercase hex (64 characters).

Returns

ResourceDescriptor The digest-pinned dependency, ready to include in a :class:BuildDefinition.

Raises

ValueError If the name or version is empty, or the digest is malformed.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def pypi_resolved_dependency(
    name: str, version: str, sha256: str
) -> ResourceDescriptor:
    """Return a digest-pinned resolved dependency for a PyPI package.

    Turns one hash-pinned lock-file entry into a :class:`ResourceDescriptor` whose
    ``uri`` is a `Package URL <https://github.com/package-url/purl-spec>`_
    (``pkg:pypi/<name>@<version>``). The package name is normalised to the PyPI form
    (lowercased, runs of ``.``/``-``/``_`` collapsed to a single ``-``, per PEP 503)
    for the URL, while the original name is preserved in the descriptor ``name`` so
    the resolved-dependency tree records exactly what the lock pinned. Assembling one
    descriptor per pinned artefact yields the fuller ``resolvedDependencies`` block a
    consumer needs to reproduce the build's inputs.

    Parameters
    ----------
    name:
        The package name as written in the lock file, non-empty.
    version:
        The pinned package version, non-empty.
    sha256:
        The pinned artefact's SHA-256 digest, lowercase hex (64 characters).

    Returns
    -------
    ResourceDescriptor
        The digest-pinned dependency, ready to include in a
        :class:`BuildDefinition`.

    Raises
    ------
    ValueError
        If the name or version is empty, or the digest is malformed.
    """
    normalised = _require_non_empty_str(name, "dependency name").lower()
    for separator in ("_", "."):
        normalised = normalised.replace(separator, "-")
    while "--" in normalised:
        normalised = normalised.replace("--", "-")
    pinned_version = _require_non_empty_str(version, "dependency version")
    return ResourceDescriptor(
        uri=f"pkg:pypi/{normalised}@{pinned_version}",
        sha256=sha256,
        name=name,
    )

provenance_statement_hash 

provenance_statement_hash(
    statement: SlsaProvenanceStatement,
) -> str

Return the canonical SHA-256 digest of a provenance statement.

Parameters

statement: The statement to hash.

Returns

str Lowercase hexadecimal SHA-256 over the canonical statement JSON.

Source code in src/scpn_phase_orchestrator/assurance/provenance.py 
452
453
454
455
456
457
458
459
460
461
462
463
464
465
def provenance_statement_hash(statement: SlsaProvenanceStatement) -> str:
    """Return the canonical SHA-256 digest of a provenance statement.

    Parameters
    ----------
    statement:
        The statement to hash.

    Returns
    -------
    str
        Lowercase hexadecimal SHA-256 over the canonical statement JSON.
    """
    return statement.statement_hash()

dsse

Wrap a SLSA provenance statement in a post-quantum-signed DSSE envelope.

DSSE <https://github.com/secure-systems-lab/dsse> (Dead Simple Signing Envelope) is the wire format sigstore/cosign <https://docs.sigstore.dev> produces for cosign attest and consumes for cosign verify-attestation: a base64 payload, its payloadType, and a list of signatures over the DSSE pre-authentication encoding (PAE) of the payload. Signing the PAE — rather than the raw JSON — is what makes the signature bind the payload type as well as the bytes, so an attestation cannot be re-labelled as a different document type.

This module carries the SLSA provenance statement from :mod:scpn_phase_orchestrator.assurance.provenance as the DSSE payload (payloadType application/vnd.in-toto+json) and signs the PAE with ML-DSA (FIPS 204), reusing the single post-quantum primitive in :mod:scpn_phase_orchestrator.runtime.audit_pqc. Each signature records its algorithm so a second scheme can be added without breaking existing envelopes; SLH-DSA (FIPS 205 / SPHINCS+) is the reserved hash-based alternative and will be added once the cryptography backend ships it (it does not as of the pinned version, so no SLH-DSA claim is made here).

The envelope is deterministic and offline: it holds no timestamps and makes no network call. Verification is self-contained — the verifier supplies the trusted public key, whose short id must match the signature keyid — so an attestation can be checked long after the build and against a future quantum adversary. Pushing the same envelope to a Rekor transparency log or verifying it with cosign is an optional operator step that needs network and OIDC, and is therefore out of this deterministic core.

Classes

DsseSignature dataclass 

DsseSignature(
    keyid: str, algorithm: str, signature_b64: str
)

One signature over a DSSE envelope's pre-authentication encoding.

Attributes

keyid: Short identifier of the signing public key (SHA-256 prefix), matching :func:~scpn_phase_orchestrator.runtime.audit_pqc.public_key_id. algorithm: The signature scheme (an ML-DSA variant), recorded so a second scheme can be added without ambiguity. signature_b64: The raw signature, standard-base64 encoded.

Methods:
__post_init__ 
__post_init__() -> None

Validate the algorithm and that the signature is decodable base64.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
108
109
110
111
112
113
def __post_init__(self) -> None:
    """Validate the algorithm and that the signature is decodable base64."""
    _require_algorithm(self.algorithm)
    if not isinstance(self.keyid, str) or not self.keyid:
        raise ValueError("keyid must be a non-empty string")
    _decode_b64(self.signature_b64, "signature_b64")
to_dict 
to_dict() -> dict[str, str]

Return the DSSE signature mapping (keyid / algorithm / sig).

Returns

dict[str, str] The keyid / algorithm / sig wire mapping.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
def to_dict(self) -> dict[str, str]:
    """Return the DSSE signature mapping (``keyid`` / ``algorithm`` / ``sig``).

    Returns
    -------
    dict[str, str]
        The ``keyid`` / ``algorithm`` / ``sig`` wire mapping.
    """
    return {
        "keyid": self.keyid,
        "algorithm": self.algorithm,
        "sig": self.signature_b64,
    }
from_dict classmethod 
from_dict(data: Mapping[str, object]) -> DsseSignature

Return a signature parsed from a DSSE signature mapping.

Parameters

data: A mapping carrying keyid, algorithm, and sig.

Returns

DsseSignature The reconstructed signature.

Raises

ValueError If a required field is missing or malformed.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
@classmethod
def from_dict(cls, data: Mapping[str, object]) -> DsseSignature:
    """Return a signature parsed from a DSSE signature mapping.

    Parameters
    ----------
    data:
        A mapping carrying ``keyid``, ``algorithm``, and ``sig``.

    Returns
    -------
    DsseSignature
        The reconstructed signature.

    Raises
    ------
    ValueError
        If a required field is missing or malformed.
    """
    for required in ("keyid", "algorithm", "sig"):
        if required not in data:
            raise ValueError(f"signature is missing field: {required}")
    return cls(
        keyid=str(data["keyid"]),
        algorithm=str(data["algorithm"]),
        signature_b64=str(data["sig"]),
    )

DsseEnvelope dataclass 

DsseEnvelope(
    payload_b64: str,
    payload_type: str,
    signatures: tuple[DsseSignature, ...],
)

A DSSE v1 envelope carrying a base64 payload and its signatures.

Attributes

payload_b64: The statement JSON, standard-base64 encoded. payload_type: The payload media type (application/vnd.in-toto+json). signatures: The signatures over the payload's pre-authentication encoding.

Methods:
__post_init__ 
__post_init__() -> None

Validate a decodable base64 payload and at least one signature.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
191
192
193
194
195
196
197
def __post_init__(self) -> None:
    """Validate a decodable base64 payload and at least one signature."""
    _decode_b64(self.payload_b64, "payload")
    if not isinstance(self.payload_type, str) or not self.payload_type:
        raise ValueError("payloadType must be a non-empty string")
    if not self.signatures:
        raise ValueError("envelope requires at least one signature")
payload_bytes 
payload_bytes() -> bytes

Return the decoded payload bytes.

Returns

bytes The base64-decoded payload (the canonical statement JSON bytes).

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
199
200
201
202
203
204
205
206
207
def payload_bytes(self) -> bytes:
    """Return the decoded payload bytes.

    Returns
    -------
    bytes
        The base64-decoded payload (the canonical statement JSON bytes).
    """
    return _decode_b64(self.payload_b64, "payload")
statement 
statement() -> dict[str, object]

Return the wrapped in-toto statement as a mapping.

Returns

dict[str, object] The decoded, JSON-parsed statement.

Raises

ValueError If the payload is not valid JSON object bytes.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def statement(self) -> dict[str, object]:
    """Return the wrapped in-toto statement as a mapping.

    Returns
    -------
    dict[str, object]
        The decoded, JSON-parsed statement.

    Raises
    ------
    ValueError
        If the payload is not valid JSON object bytes.
    """
    try:
        parsed = json.loads(self.payload_bytes())
    except json.JSONDecodeError as exc:
        raise ValueError("payload is not valid JSON") from exc
    if not isinstance(parsed, dict):
        raise ValueError("payload is not a JSON object")
    return parsed
to_dict 
to_dict() -> dict[str, object]

Return the DSSE wire mapping (payload / payloadType / signatures).

Returns

dict[str, object] The payload / payloadType / signatures wire mapping.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
def to_dict(self) -> dict[str, object]:
    """Return the DSSE wire mapping (payload / payloadType / signatures).

    Returns
    -------
    dict[str, object]
        The ``payload`` / ``payloadType`` / ``signatures`` wire mapping.
    """
    return {
        "payload": self.payload_b64,
        "payloadType": self.payload_type,
        "signatures": [signature.to_dict() for signature in self.signatures],
    }
from_dict classmethod 
from_dict(data: Mapping[str, object]) -> DsseEnvelope

Return an envelope parsed from a DSSE wire mapping.

Parameters

data: A mapping carrying payload, payloadType, and signatures.

Returns

DsseEnvelope The reconstructed envelope.

Raises

ValueError If a required field is missing or the signatures are malformed.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
@classmethod
def from_dict(cls, data: Mapping[str, object]) -> DsseEnvelope:
    """Return an envelope parsed from a DSSE wire mapping.

    Parameters
    ----------
    data:
        A mapping carrying ``payload``, ``payloadType``, and ``signatures``.

    Returns
    -------
    DsseEnvelope
        The reconstructed envelope.

    Raises
    ------
    ValueError
        If a required field is missing or the signatures are malformed.
    """
    for required in ("payload", "payloadType", "signatures"):
        if required not in data:
            raise ValueError(f"envelope is missing field: {required}")
    raw_signatures = data["signatures"]
    if not isinstance(raw_signatures, (list, tuple)):
        raise ValueError("signatures must be a list")
    signatures = tuple(
        DsseSignature.from_dict(_as_mapping(item)) for item in raw_signatures
    )
    return cls(
        payload_b64=str(data["payload"]),
        payload_type=str(data["payloadType"]),
        signatures=signatures,
    )

Functions:

sign_provenance_statement 

sign_provenance_statement(
    statement: SlsaProvenanceStatement,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> DsseEnvelope

Wrap a provenance statement in a DSSE envelope and sign it with ML-DSA.

Parameters

statement: The SLSA provenance statement to attest. private_key: An ML-DSA private key matching algorithm (see :func:~scpn_phase_orchestrator.runtime.audit_pqc.signing_key_from_seed). algorithm: The ML-DSA variant; must match private_key.

Returns

DsseEnvelope The signed attestation envelope.

Raises

ValueError If the algorithm or private key is invalid.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
def sign_provenance_statement(
    statement: SlsaProvenanceStatement,
    private_key: Any,
    *,
    algorithm: str = DEFAULT_VARIANT,
) -> DsseEnvelope:
    """Wrap a provenance statement in a DSSE envelope and sign it with ML-DSA.

    Parameters
    ----------
    statement:
        The SLSA provenance statement to attest.
    private_key:
        An ML-DSA private key matching ``algorithm`` (see
        :func:`~scpn_phase_orchestrator.runtime.audit_pqc.signing_key_from_seed`).
    algorithm:
        The ML-DSA variant; must match ``private_key``.

    Returns
    -------
    DsseEnvelope
        The signed attestation envelope.

    Raises
    ------
    ValueError
        If the algorithm or private key is invalid.
    """
    algorithm = _require_algorithm(algorithm)
    body = json.dumps(
        statement.to_statement(), sort_keys=True, separators=(",", ":")
    ).encode("utf-8")
    signature = sign_bytes(
        _pae(DSSE_PAYLOAD_TYPE, body), private_key, algorithm=algorithm
    )
    public_bytes = private_key.public_key().public_bytes_raw()
    dsse_signature = DsseSignature(
        keyid=public_key_id(public_bytes),
        algorithm=algorithm,
        signature_b64=_encode_b64(signature),
    )
    return DsseEnvelope(
        payload_b64=_encode_b64(body),
        payload_type=DSSE_PAYLOAD_TYPE,
        signatures=(dsse_signature,),
    )

verify_dsse_envelope 

verify_dsse_envelope(
    envelope: DsseEnvelope, trusted_public_key_hex: str
) -> bool

Verify that an envelope carries a valid signature under a trusted key.

The verifier supplies the public key it trusts; the envelope is accepted only if a signature whose keyid matches that key verifies over the payload's pre-authentication encoding. This binds the attestation to a known signer, so a forged envelope signed under a different key is rejected.

Parameters

envelope: The DSSE envelope to verify. trusted_public_key_hex: The hex-encoded raw ML-DSA public key the verifier trusts.

Returns

bool True if a matching signature verifies, else False.

Raises

ValueError If the trusted key is not a hex string.

Source code in src/scpn_phase_orchestrator/assurance/dsse.py 
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
def verify_dsse_envelope(
    envelope: DsseEnvelope,
    trusted_public_key_hex: str,
) -> bool:
    """Verify that an envelope carries a valid signature under a trusted key.

    The verifier supplies the public key it trusts; the envelope is accepted only if
    a signature whose ``keyid`` matches that key verifies over the payload's
    pre-authentication encoding. This binds the attestation to a known signer, so a
    forged envelope signed under a different key is rejected.

    Parameters
    ----------
    envelope:
        The DSSE envelope to verify.
    trusted_public_key_hex:
        The hex-encoded raw ML-DSA public key the verifier trusts.

    Returns
    -------
    bool
        ``True`` if a matching signature verifies, else ``False``.

    Raises
    ------
    ValueError
        If the trusted key is not a hex string.
    """
    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
    trusted_keyid = public_key_id(trusted_bytes)
    body = envelope.payload_bytes()
    pae = _pae(envelope.payload_type, body)
    for signature in envelope.signatures:
        # Every signature's algorithm is validated at construction, so no algorithm
        # re-check is needed here — only the keyid must match the trusted key.
        if signature.keyid != trusted_keyid:
            continue
        raw_signature = _decode_b64(signature.signature_b64, "signature_b64")
        if verify_bytes(
            pae, raw_signature, trusted_public_key_hex, algorithm=signature.algorithm
        ):
            return True
    return False

Conformity report

scpn_phase_orchestrator.assurance.report renders an assurance-case bundle as a deterministic Markdown conformity report — the document a regulatory assessor reads. It restates the sealed bundle verbatim (coverage rollup, per-standard clause conformance with status, evidence, and rationale, and the evidence inventory) under the regulatory disclaimer and anchored to the bundle hash. It adds no claim beyond the bundle and is review-only. The certification evidence package seals the rendered report as conformity_report.md. render_conformity_report_pdf renders the same content as a deterministic, dependency-free text PDF — the distributable artefact an assessor files — built on the reusable scpn_phase_orchestrator.reporting.markdown_to_pdf_bytes helper.

report

Render an assurance-case bundle as a human-readable conformity report.

The certification evidence package seals machine-readable JSON (the bundle, the hash test vectors, and a manifest). A regulatory assessor, however, reads a document: this module renders the same sealed evidence as a deterministic Markdown conformity report — a per-standard, clause-by-clause table of conformance status, contributing evidence, and rationale, prefixed by the coverage rollup and the regulatory disclaimer and anchored to the bundle hash for traceability.

The report is review-only and adds no new claims: every status, evidence identifier, and rationale is read verbatim from the :class:~scpn_phase_orchestrator.assurance.case.AssuranceCaseBundle. Rendering is deterministic — standards, clauses, and evidence are emitted in a stable sort order so the report digest is reproducible.

Classes

Functions:

render_conformity_report 

render_conformity_report(
    bundle: AssuranceCaseBundle,
) -> str

Render an assurance-case bundle as a Markdown conformity report.

The report restates the bundle verbatim — coverage rollup, per-standard clause conformance (status, contributing evidence, rationale), and the evidence inventory — under the regulatory disclaimer and anchored to the bundle hash. It adds no claim not already present in bundle and is review-only. Rendering is deterministic for a given bundle.

Parameters

bundle: The hash-sealed assurance-case bundle to render.

Returns

str The Markdown conformity report, terminated by a single newline.

Source code in src/scpn_phase_orchestrator/assurance/report.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def render_conformity_report(bundle: AssuranceCaseBundle) -> str:
    """Render an assurance-case bundle as a Markdown conformity report.

    The report restates the bundle verbatim — coverage rollup, per-standard
    clause conformance (status, contributing evidence, rationale), and the
    evidence inventory — under the regulatory disclaimer and anchored to the
    bundle hash. It adds no claim not already present in ``bundle`` and is
    review-only. Rendering is deterministic for a given bundle.

    Parameters
    ----------
    bundle:
        The hash-sealed assurance-case bundle to render.

    Returns
    -------
    str
        The Markdown conformity report, terminated by a single newline.
    """
    lines: list[str] = []
    lines.extend(_header_block(bundle))
    lines.extend(_coverage_block(bundle))
    lines.extend(_clause_blocks(bundle))
    lines.extend(_evidence_block(bundle))
    lines.append("---")
    lines.append("")
    lines.append(
        "Review-only conformity evidence report. This document is a technical "
        "evidence aid, not a conformity assessment or certification of "
        "compliance. Traceability anchor: bundle hash "
        f"`{bundle.bundle_hash}`."
    )
    return "\n".join(lines) + "\n"

render_conformity_report_pdf 

render_conformity_report_pdf(
    bundle: AssuranceCaseBundle,
) -> bytes

Render the conformity report as a deterministic, dependency-free PDF.

Produces the same content as :func:render_conformity_report in a minimal single-font text PDF — the distributable artefact an assessor files. The bytes carry no timestamp and are reproducible for a given bundle. The text PDF renderer is imported lazily so importing this module stays light.

Parameters

bundle: The hash-sealed assurance-case bundle to render.

Returns

bytes The rendered conformity report PDF.

Source code in src/scpn_phase_orchestrator/assurance/report.py 
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def render_conformity_report_pdf(bundle: AssuranceCaseBundle) -> bytes:
    """Render the conformity report as a deterministic, dependency-free PDF.

    Produces the same content as :func:`render_conformity_report` in a minimal
    single-font text PDF — the distributable artefact an assessor files. The
    bytes carry no timestamp and are reproducible for a given bundle. The text
    PDF renderer is imported lazily so importing this module stays light.

    Parameters
    ----------
    bundle:
        The hash-sealed assurance-case bundle to render.

    Returns
    -------
    bytes
        The rendered conformity report PDF.
    """
    from scpn_phase_orchestrator.reporting import markdown_to_pdf_bytes

    return markdown_to_pdf_bytes(render_conformity_report(bundle))

Oscillation-Monitoring Evidence (NERC PRC-028-1 / PRC-030-1)

scpn_phase_orchestrator.assurance.prc_oscillation is the audit-package end of the dVOC grid pack. screen_oscillation_modes takes the modes recovered by the matrix-pencil estimator, screens each damping ratio for PRC-028-1 disturbance-data analysis and PRC-030-1 unexpected IBR event mitigation workflows, and seals the screening into a content-addressed, review-only PRCOscillationEvidence record. Undamped modes and positive but poorly damped modes are flagged for operator review. Each finding also carries the engineering mode family from the matrix-pencil estimator, and the record aggregates mode_family_counts, so inter-area and sub-synchronous oscillation signals are visible in the same sealed package. The capture timestamp is supplied by the caller, so the record is deterministic and reproducible. Like the assurance-case bundle, it is a technical evidence-mapping aid, not a legal conformity assessment, and it never actuates.

prc_oscillation

NERC PRC oscillation-monitoring compliance evidence from detected modes.

This is the audit-package end of the dVOC grid pack. The matrix-pencil estimator (:mod:~scpn_phase_orchestrator.monitor.oscillation_modes) detects the electromechanical modes of a ringdown and their damping ratios; :func:screen_oscillation_modes screens those damping ratios for NERC PRC-028-1 disturbance-data analysis and PRC-030-1 unexpected IBR event mitigation workflows: a mode whose damping ratio sits below a few percent is poorly damped, and a mode with non-positive damping is undamped (growing). The screening also preserves an engineering mode-family label for each finding, so inter-area and sub-synchronous signals stay visible in the hash-sealed, review-only :class:PRCOscillationEvidence record.

The record is content-addressed with the same canonical-JSON SHA-256 hashing the assurance-case bundle uses, so it can be referenced by a stable digest and any later mutation is detectable. The capture timestamp is supplied by the caller (it is the measurement time of the PMU/ringdown event, not a wall-clock reading taken here) so the record is deterministic and reproducible.

This is a technical evidence-mapping aid, not a legal conformity assessment: it links measured modal damping to the disturbance-monitoring and unexpected-event mitigation workflows those standards support. The exact identifiers, thresholds, and reporting obligations must be confirmed against the issued standards — see :data:PRC_OSCILLATION_DISCLAIMER. The screening is review-only: it reads detected modes and reports findings; it never changes bindings, layers, or coupling.

References

  • NERC PRC-028-1 (disturbance monitoring and reporting for inverter-based resources) and PRC-030-1 (unexpected inverter-based resource event mitigation), developed under FERC Order 901.

Classes

PRCModeFinding dataclass 

PRCModeFinding(
    mode_index: int,
    frequency_hz: float,
    damping_ratio: float,
    amplitude: float,
    mode_family: str,
    classification: str,
    flagged: bool,
)

The screening outcome for a single detected mode.

Attributes

mode_index : int Position of the mode in the screened sequence. frequency_hz : float Modal oscillation frequency in hertz. damping_ratio : float Dimensionless damping ratio of the mode. amplitude : float Modal amplitude in the units of the source signal. mode_family : str Engineering oscillation family such as inter_area or sub_synchronous. classification : str One of :data:UNDAMPED, :data:POORLY_DAMPED, or :data:ACCEPTABLE. flagged : bool Whether the mode breaches a screening threshold (undamped or poorly damped).

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

Return a JSON-safe mapping of the finding.

Returns

dict[str, object] The mode index, frequency, damping ratio, amplitude, classification, and flagged status.

Source code in src/scpn_phase_orchestrator/assurance/prc_oscillation.py 
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the finding.

    Returns
    -------
    dict[str, object]
        The mode index, frequency, damping ratio, amplitude, classification,
        and flagged status.
    """
    return {
        "mode_index": self.mode_index,
        "frequency_hz": self.frequency_hz,
        "damping_ratio": self.damping_ratio,
        "amplitude": self.amplitude,
        "mode_family": self.mode_family,
        "classification": self.classification,
        "flagged": self.flagged,
    }

PRCOscillationEvidence dataclass 

PRCOscillationEvidence(
    event_id: str,
    captured_at: str,
    signal_source: str,
    sampling_rate_hz: float,
    poorly_damped_threshold: float,
    undamped_threshold: float,
    findings: tuple[PRCModeFinding, ...],
    mode_family_counts: Mapping[str, int],
    flagged_count: int,
    worst_damping_ratio: float | None,
    verdict: str,
    standard: str,
    disclaimer: str,
)

A hash-sealed oscillation-monitoring compliance-evidence record.

Attributes

event_id : str Caller-assigned identifier for the oscillation event. captured_at : str Measurement timestamp of the event, supplied by the caller. signal_source : str Identifier of the screened signal (a bus, tie-line, or order parameter). sampling_rate_hz : float Sampling rate of the ringdown the modes were estimated from. poorly_damped_threshold : float Damping ratio below which a positively-damped mode is flagged. undamped_threshold : float Damping ratio at or below which a mode is flagged undamped (growing). findings : tuple[PRCModeFinding, ...] Per-mode screening outcomes, in the order screened. mode_family_counts : Mapping[str, int] Read-only number of screened modes by engineering family. flagged_count : int Number of findings flagged for review. worst_damping_ratio : float | None Lowest damping ratio across the findings, or None when no modes were detected. verdict : str :data:FLAGGED_FOR_REVIEW if any mode is flagged, else :data:NO_EXCEEDANCE. standard : str The standard family the record is mapped to. disclaimer : str The review-only regulatory disclaimer. content_hash : str SHA-256 of the canonical record (excluding this field); computed on construction.

Methods:
__post_init__ 
__post_init__() -> None

Compute the content hash from the canonical evidence payload.

Source code in src/scpn_phase_orchestrator/assurance/prc_oscillation.py 
207
208
209
210
211
212
213
214
def __post_init__(self) -> None:
    """Compute the content hash from the canonical evidence payload."""
    object.__setattr__(
        self, "mode_family_counts", MappingProxyType(dict(self.mode_family_counts))
    )
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the whole evidence record.

Returns

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

Source code in src/scpn_phase_orchestrator/assurance/prc_oscillation.py 
234
235
236
237
238
239
240
241
242
243
244
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the whole evidence record.

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

Functions:

screen_oscillation_modes 

screen_oscillation_modes(
    modes: Sequence[OscillationMode],
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    sampling_rate_hz: float,
    poorly_damped_threshold: float = DEFAULT_DAMPING_THRESHOLD,
    undamped_threshold: float = 0.0,
) -> PRCOscillationEvidence

Screen detected oscillation modes into a PRC compliance-evidence record.

Parameters

modes : Sequence[OscillationMode] Modes recovered from a ringdown, e.g. by :func:~scpn_phase_orchestrator.monitor.oscillation_modes.estimate_oscillation_modes. event_id : str Caller-assigned identifier for the oscillation event. captured_at : str Measurement timestamp of the event, supplied by the caller. signal_source : str Identifier of the screened signal. sampling_rate_hz : float Sampling rate of the ringdown, in hertz (> 0). poorly_damped_threshold : float Damping ratio below which a positively-damped mode is flagged poorly damped. undamped_threshold : float Damping ratio at or below which a mode is flagged undamped; must be below poorly_damped_threshold.

Returns

PRCOscillationEvidence The hash-sealed, review-only screening record.

Raises

ValueError If an identifier is empty, the sampling rate is not positive, the thresholds are not ordered finite reals, or an element is not an :class:~scpn_phase_orchestrator.monitor.oscillation_modes.OscillationMode.

Source code in src/scpn_phase_orchestrator/assurance/prc_oscillation.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
def screen_oscillation_modes(
    modes: Sequence[OscillationMode],
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    sampling_rate_hz: float,
    poorly_damped_threshold: float = DEFAULT_DAMPING_THRESHOLD,
    undamped_threshold: float = 0.0,
) -> PRCOscillationEvidence:
    """Screen detected oscillation modes into a PRC compliance-evidence record.

    Parameters
    ----------
    modes : Sequence[OscillationMode]
        Modes recovered from a ringdown, e.g. by
        :func:`~scpn_phase_orchestrator.monitor.oscillation_modes.estimate_oscillation_modes`.
    event_id : str
        Caller-assigned identifier for the oscillation event.
    captured_at : str
        Measurement timestamp of the event, supplied by the caller.
    signal_source : str
        Identifier of the screened signal.
    sampling_rate_hz : float
        Sampling rate of the ringdown, in hertz (``> 0``).
    poorly_damped_threshold : float
        Damping ratio below which a positively-damped mode is flagged poorly
        damped.
    undamped_threshold : float
        Damping ratio at or below which a mode is flagged undamped; must be below
        ``poorly_damped_threshold``.

    Returns
    -------
    PRCOscillationEvidence
        The hash-sealed, review-only screening record.

    Raises
    ------
    ValueError
        If an identifier is empty, the sampling rate is not positive, the
        thresholds are not ordered finite reals, or an element is not an
        :class:`~scpn_phase_orchestrator.monitor.oscillation_modes.OscillationMode`.
    """
    event = _non_empty_str(event_id, "event_id")
    captured = _non_empty_str(captured_at, "captured_at")
    source = _non_empty_str(signal_source, "signal_source")
    fs = _positive_real(sampling_rate_hz, "sampling_rate_hz")
    undamped = _real_scalar(undamped_threshold, "undamped_threshold")
    poorly = _real_scalar(poorly_damped_threshold, "poorly_damped_threshold")
    if not undamped < poorly:
        raise ValueError("undamped_threshold must be below poorly_damped_threshold")

    findings = tuple(
        _screen_mode(index, mode, undamped, poorly) for index, mode in enumerate(modes)
    )
    flagged_count = sum(1 for finding in findings if finding.flagged)
    worst = min((finding.damping_ratio for finding in findings), default=None)
    verdict = FLAGGED_FOR_REVIEW if flagged_count else NO_EXCEEDANCE
    return PRCOscillationEvidence(
        event_id=event,
        captured_at=captured,
        signal_source=source,
        sampling_rate_hz=fs,
        poorly_damped_threshold=poorly,
        undamped_threshold=undamped,
        findings=findings,
        mode_family_counts=_mode_family_counts(findings),
        flagged_count=flagged_count,
        worst_damping_ratio=worst,
        verdict=verdict,
        standard=PRC_OSCILLATION_STANDARD,
        disclaimer=PRC_OSCILLATION_DISCLAIMER,
    )

Ride-Through Evidence (NERC PRC-029-1)

scpn_phase_orchestrator.assurance.prc_ride_through screens operator-provided high-side transformer voltage and frequency samples against the approved NERC PRC-029-1 ride-through tables. It carries both voltage categories from Attachment 1 — AC-connected wind IBRs and all other IBRs — plus the Attachment 2 frequency bands. The screener aggregates cumulative duration inside the standard's voltage and frequency review windows, records the operation region, minimum ride-through duration, observed value range, and review classification for each non-nominal band, then seals the record as PRCRideThroughEvidence.

The record is review-only. It does not evaluate real/reactive-current performance, phase-jump exceptions, hardware-limit exemptions, reporting duties, or legal compliance. Observations outside the review envelope use assessor_review_required, not pass/fail language.

prc_ride_through

Review-only NERC PRC-029 ride-through evidence screening.

This module maps operator-provided high-side transformer voltage and frequency time series into deterministic, hash-sealed evidence for NERC PRC-029-1 review. It implements the published Attachment 1 voltage ride-through tables for AC-connected wind IBRs and all other IBRs, plus the Attachment 2 frequency ride-through table, then records only technical screening findings. It does not assert conformance, evaluate real/reactive-current performance, apply hardware limitation exemptions, or replace qualified assessor review.

Classes

PRCRideThroughFinding dataclass 

PRCRideThroughFinding(
    channel: str,
    band: str,
    operation_region: str,
    start_s: float,
    end_s: float,
    duration_s: float,
    window_duration_s: float,
    observed_min: float,
    observed_max: float,
    minimum_ride_through_s: float | None,
    window_s: float | None,
    classification: str,
    flagged: bool,
)

One aggregated PRC-029 voltage or frequency ride-through observation.

Attributes

channel : str "voltage" or "frequency". band : str Deterministic threshold-band identifier. operation_region : str Operation-region class from the PRC-029 ride-through tables. start_s, end_s : float First and last time covered by the aggregate observation. duration_s : float Total observed duration in the band across the trace. window_duration_s : float Maximum cumulative duration inside the relevant PRC-029 assessment window. observed_min, observed_max : float Minimum and maximum observed measurement values inside the band. minimum_ride_through_s : float | None Published minimum ride-through duration for the band, or None for may-trip zones. window_s : float | None Assessment window used for cumulative-duration screening. classification : str Screening classification. This is review language, not a legal verdict. flagged : bool Whether the observation needs qualified assessor review.

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

Return a JSON-safe mapping of the finding.

Returns

dict[str, object] Stable audit fields for one ride-through observation.

Source code in src/scpn_phase_orchestrator/assurance/prc_ride_through.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the finding.

    Returns
    -------
    dict[str, object]
        Stable audit fields for one ride-through observation.
    """
    return {
        "channel": self.channel,
        "band": self.band,
        "operation_region": self.operation_region,
        "start_s": self.start_s,
        "end_s": self.end_s,
        "duration_s": self.duration_s,
        "window_duration_s": self.window_duration_s,
        "observed_min": self.observed_min,
        "observed_max": self.observed_max,
        "minimum_ride_through_s": self.minimum_ride_through_s,
        "window_s": self.window_s,
        "classification": self.classification,
        "flagged": self.flagged,
    }

PRCRideThroughEvidence dataclass 

PRCRideThroughEvidence(
    event_id: str,
    captured_at: str,
    signal_source: str,
    ibr_category: str,
    sample_count: int,
    duration_s: float,
    findings: tuple[PRCRideThroughFinding, ...],
    channel_counts: Mapping[str, int],
    flagged_count: int,
    verdict: str,
    standard: str,
    disclaimer: str,
)

Hash-sealed PRC-029 ride-through screening evidence.

Attributes

event_id : str Caller-assigned event identifier. captured_at : str Measurement timestamp supplied by the caller. signal_source : str Operator-facing source label. ibr_category : str PRC-029 voltage-table category: :data:AC_WIND_IBR or :data:OTHER_IBR. sample_count : int Number of time-series samples consumed. duration_s : float Elapsed time from first to last sample. findings : tuple[PRCRideThroughFinding, ...] Aggregated voltage and frequency screening observations. channel_counts : Mapping[str, int] Read-only number of observations by channel. flagged_count : int Number of observations that require qualified review. verdict : str Review-only verdict string. standard : str Standard family the record is mapped to. disclaimer : str Review-only disclaimer. content_hash : str SHA-256 of the canonical record excluding this field.

Methods:
__post_init__ 
__post_init__() -> None

Freeze channel counts and compute the canonical content hash.

Source code in src/scpn_phase_orchestrator/assurance/prc_ride_through.py 
235
236
237
238
239
240
241
242
def __post_init__(self) -> None:
    """Freeze channel counts and compute the canonical content hash."""
    object.__setattr__(
        self, "channel_counts", MappingProxyType(dict(self.channel_counts))
    )
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the evidence record.

Returns

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

Source code in src/scpn_phase_orchestrator/assurance/prc_ride_through.py 
261
262
263
264
265
266
267
268
269
270
271
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the evidence record.

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

Functions:

screen_ride_through_samples 

screen_ride_through_samples(
    time_s: Sequence[object],
    voltage_pu: Sequence[object],
    frequency_hz: Sequence[object],
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    ibr_category: str = OTHER_IBR,
) -> PRCRideThroughEvidence

Screen voltage and frequency samples into PRC-029 review evidence.

Parameters

time_s : Sequence[object] Monotonic sample times in seconds. voltage_pu : Sequence[object] Voltage measurements in per unit at the applicable PRC-029 measurement point. frequency_hz : Sequence[object] Frequency measurements in hertz at the applicable PRC-029 measurement point. event_id : str Caller-assigned event identifier. captured_at : str Measurement timestamp stamped into the evidence. signal_source : str Operator-facing source label. ibr_category : str Voltage ride-through table selector, either :data:AC_WIND_IBR or :data:OTHER_IBR.

Returns

PRCRideThroughEvidence Deterministic, review-only PRC-029 screening evidence.

Raises

ValueError If identifiers, category, samples, or time ordering are invalid.

Source code in src/scpn_phase_orchestrator/assurance/prc_ride_through.py 
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
def screen_ride_through_samples(
    time_s: Sequence[object],
    voltage_pu: Sequence[object],
    frequency_hz: Sequence[object],
    *,
    event_id: str,
    captured_at: str,
    signal_source: str,
    ibr_category: str = OTHER_IBR,
) -> PRCRideThroughEvidence:
    """Screen voltage and frequency samples into PRC-029 review evidence.

    Parameters
    ----------
    time_s : Sequence[object]
        Monotonic sample times in seconds.
    voltage_pu : Sequence[object]
        Voltage measurements in per unit at the applicable PRC-029 measurement
        point.
    frequency_hz : Sequence[object]
        Frequency measurements in hertz at the applicable PRC-029 measurement
        point.
    event_id : str
        Caller-assigned event identifier.
    captured_at : str
        Measurement timestamp stamped into the evidence.
    signal_source : str
        Operator-facing source label.
    ibr_category : str
        Voltage ride-through table selector, either :data:`AC_WIND_IBR` or
        :data:`OTHER_IBR`.

    Returns
    -------
    PRCRideThroughEvidence
        Deterministic, review-only PRC-029 screening evidence.

    Raises
    ------
    ValueError
        If identifiers, category, samples, or time ordering are invalid.
    """
    event = _non_empty_str(event_id, "event_id")
    captured = _non_empty_str(captured_at, "captured_at")
    source = _non_empty_str(signal_source, "signal_source")
    category = _ibr_category(ibr_category)
    times = _as_real_array(time_s, "time_s")
    voltage = _as_real_array(voltage_pu, "voltage_pu")
    frequency = _as_real_array(frequency_hz, "frequency_hz")
    _validate_shapes(times, voltage, frequency)

    voltage_findings = _collect_channel_findings(
        "voltage",
        _segments(times, voltage, lambda value: _voltage_band(value, category)),
    )
    frequency_findings = _collect_channel_findings(
        "frequency",
        _segments(times, frequency, _frequency_band),
    )
    findings = (*voltage_findings, *frequency_findings)
    flagged_count = sum(1 for finding in findings if finding.flagged)
    verdict = ASSESSOR_REVIEW_REQUIRED if flagged_count else WITHIN_REVIEW_ENVELOPE
    return PRCRideThroughEvidence(
        event_id=event,
        captured_at=captured,
        signal_source=source,
        ibr_category=category,
        sample_count=int(times.shape[0]),
        duration_s=float(times[-1] - times[0]),
        findings=findings,
        channel_counts=_channel_counts(findings),
        flagged_count=flagged_count,
        verdict=verdict,
        standard=PRC_RIDE_THROUGH_STANDARD,
        disclaimer=PRC_RIDE_THROUGH_DISCLAIMER,
    )

Power-Grid PRC Assessor Bundle

scpn_phase_orchestrator.assurance.power_grid_prc_bundle binds the three power-grid PRC review artefacts into one deterministic handoff package:

  • scpn_dvoc_oscillation_damping_audit_v1 from the offline dVOC/Koopman-MPC damping screen;
  • scpn_pmu_ringdown_prc_audit_v1 from an operator PMU frequency ringdown CSV;
  • scpn_ibr_ride_through_prc029_audit_v1 from an operator voltage/frequency ride-through CSV.

The builder verifies the source JSON SHA-256 metadata, exact child schema, review-only claim boundary, and each child content_hash before sealing the bundle as scpn_power_grid_prc_audit_bundle_v1. The bundle keeps the full child records for assessor replay and carries no live-actuation or conformity claim.

power_grid_prc_bundle

Hash-sealed power-grid PRC assessor bundles.

The power-grid review lane emits several independent evidence records: the dVOC oscillation-damping audit, an operator PMU ringdown screen, and an IBR ride-through screen. This module binds those records into one deterministic, review-only handoff package. It verifies each child content hash before the bundle is sealed, so an assessor can detect both source-file mutation and evidence-record mutation.

Classes

PowerGridPRCInputArtifact dataclass 

PowerGridPRCInputArtifact(
    role: str,
    source_name: str,
    source_sha256: str,
    record: Mapping[str, object],
)

A source evidence record prepared for bundle assembly.

Attributes

role : str Required evidence role in the power-grid PRC bundle. source_name : str Operator-facing basename or label of the evidence JSON source. source_sha256 : str SHA-256 digest of the exact evidence JSON bytes consumed by the bundle builder. record : Mapping[str, object] Parsed evidence record. Its own content_hash is rechecked before the bundle is emitted.

PowerGridPRCArtifact dataclass 

PowerGridPRCArtifact(
    role: str,
    source_name: str,
    source_sha256: str,
    evidence_schema: str,
    evidence_hash: str,
    record: Mapping[str, object],
)

A validated evidence artifact inside the assessor bundle.

Attributes

role : str Required evidence role in the bundle. source_name : str Source evidence JSON basename or label. source_sha256 : str SHA-256 digest of the exact evidence JSON bytes consumed. evidence_schema : str Schema identifier in the child evidence record. evidence_hash : str Validated child content_hash. record : Mapping[str, object] Parsed child evidence record, preserved verbatim for assessor replay.

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

Return a JSON-safe mapping of the artifact.

Returns

dict[str, object] Stable source metadata plus the preserved child evidence record.

Source code in src/scpn_phase_orchestrator/assurance/power_grid_prc_bundle.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the artifact.

    Returns
    -------
    dict[str, object]
        Stable source metadata plus the preserved child evidence record.
    """
    return {
        "role": self.role,
        "source_name": self.source_name,
        "source_sha256": self.source_sha256,
        "evidence_schema": self.evidence_schema,
        "evidence_hash": self.evidence_hash,
        "record": dict(self.record),
    }

PowerGridPRCAuditBundle dataclass 

PowerGridPRCAuditBundle(
    schema: str,
    bundle_id: str,
    created_at: str,
    operator_context: str,
    artifacts: tuple[PowerGridPRCArtifact, ...],
    evidence_hashes: Mapping[str, str],
    claim_boundary: str = POWER_GRID_PRC_CLAIM_BOUNDARY,
    review_only: bool = True,
    disclaimer: str = POWER_GRID_PRC_AUDIT_BUNDLE_DISCLAIMER,
)

A hash-sealed power-grid PRC assessor handoff bundle.

Attributes

schema : str Bundle schema identifier. bundle_id : str Operator-assigned bundle identifier. created_at : str Timestamp supplied by the caller. operator_context : str Human-readable review context for the assessor handoff. artifacts : tuple[PowerGridPRCArtifact, ...] Validated child artifacts in required role order. evidence_hashes : Mapping[str, str] Read-only map from role to child evidence hash. claim_boundary : str Review-only claim boundary. review_only : bool Always True for this bundle. disclaimer : str Regulatory and live-actuation disclaimer. content_hash : str SHA-256 of the canonical bundle payload excluding this field.

Methods:
__post_init__ 
__post_init__() -> None

Freeze hash maps and compute the canonical bundle digest.

Source code in src/scpn_phase_orchestrator/assurance/power_grid_prc_bundle.py 
182
183
184
185
186
187
188
189
190
191
def __post_init__(self) -> None:
    """Freeze hash maps and compute the canonical bundle digest."""
    object.__setattr__(
        self,
        "evidence_hashes",
        MappingProxyType(dict(self.evidence_hashes)),
    )
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the assessor bundle.

Returns

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

Source code in src/scpn_phase_orchestrator/assurance/power_grid_prc_bundle.py 
207
208
209
210
211
212
213
214
215
216
217
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the assessor bundle.

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

Functions:

build_power_grid_prc_audit_bundle 

build_power_grid_prc_audit_bundle(
    *,
    bundle_id: str,
    created_at: str,
    operator_context: str,
    artifacts: Sequence[PowerGridPRCInputArtifact],
) -> PowerGridPRCAuditBundle

Build a deterministic power-grid PRC assessor handoff bundle.

Parameters

bundle_id : str Operator-assigned bundle identifier. created_at : str Timestamp supplied by the caller. operator_context : str Human-readable review context. artifacts : Sequence[PowerGridPRCInputArtifact] Candidate child evidence records with source-file digests.

Returns

PowerGridPRCAuditBundle Hash-sealed bundle containing exactly the required child roles.

Raises

ValueError If identifiers, source metadata, child schemas, child hashes, or the role set are invalid.

Source code in src/scpn_phase_orchestrator/assurance/power_grid_prc_bundle.py 
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def build_power_grid_prc_audit_bundle(
    *,
    bundle_id: str,
    created_at: str,
    operator_context: str,
    artifacts: Sequence[PowerGridPRCInputArtifact],
) -> PowerGridPRCAuditBundle:
    """Build a deterministic power-grid PRC assessor handoff bundle.

    Parameters
    ----------
    bundle_id : str
        Operator-assigned bundle identifier.
    created_at : str
        Timestamp supplied by the caller.
    operator_context : str
        Human-readable review context.
    artifacts : Sequence[PowerGridPRCInputArtifact]
        Candidate child evidence records with source-file digests.

    Returns
    -------
    PowerGridPRCAuditBundle
        Hash-sealed bundle containing exactly the required child roles.

    Raises
    ------
    ValueError
        If identifiers, source metadata, child schemas, child hashes, or the
        role set are invalid.
    """
    bundle = _non_empty_str(bundle_id, "bundle_id")
    timestamp = _non_empty_str(created_at, "created_at")
    context = _non_empty_str(operator_context, "operator_context")
    ordered_artifacts = _validate_artifacts(artifacts)
    evidence_hashes = {
        artifact.role: artifact.evidence_hash for artifact in ordered_artifacts
    }
    return PowerGridPRCAuditBundle(
        schema=POWER_GRID_PRC_AUDIT_BUNDLE_SCHEMA,
        bundle_id=bundle,
        created_at=timestamp,
        operator_context=context,
        artifacts=ordered_artifacts,
        evidence_hashes=evidence_hashes,
    )

Early-Warning Assurance Evidence

scpn_phase_orchestrator.assurance.early_warning_evidence is the auditable envelope around the early-warning detector suite. A fair head-to-head established that early-warning detection is a commodity — no single indicator beats the others by a decisive margin — so what this module supplies is not a better detector but a content-addressed record that pins which indicators contributed and their robust z-scores at the alarm window, the provenance of the screened signal, the claim boundary (a review-only technical artefact, not a clinical, operational, or safety decision, nor a certification), and, when a ground-truth transition onset is supplied, the honest lead time — including a non-positive lead when the alarm was late rather than suppressing it.

seal_early_warning is the detector-neutral primitive: it depends only on the alarm decision, the provenance, and a pre-extracted set of EarlyWarningIndicator contributions, so it seals any present or future detector (including the real-EEG capstone) without importing detector internals. The seal_*_alarm adapters bridge each concrete suite detector — and the fused ensemble — onto that primitive. The record is content-addressed with the same canonical-JSON SHA-256 the assurance-case bundle and the NERC PRC oscillation evidence use, so a sealed alarm can be referenced by a stable digest and any later mutation is detectable. It never actuates.

early_warning_evidence

Hash-sealed, claim-bounded assurance evidence for an early-warning alarm.

The early-warning detector suite — critical slowing down (:mod:~scpn_phase_orchestrator.monitor.critical_slowing_down), rising synchronisation (:mod:~scpn_phase_orchestrator.monitor.synchronisation), and ordinal-transition entropy (:mod:~scpn_phase_orchestrator.monitor.explosive_sync) — reads a passive observable and emits a warning record. A fair head-to-head (bench/early_warning_leadtime.py) established that the detection is a commodity: none of these indicators beats the others by a decisive margin. What is not a commodity, and what this module supplies, is the auditable envelope around the alarm: a content-addressed record that pins which indicators contributed and their robust z-scores at the alarm window, the provenance of the screened signal, the claim boundary (this is a review-only technical artefact, not a clinical/operational/safety decision or a certification), and, when a ground-truth transition onset is supplied, the honest lead time — including a non-positive lead when the alarm was late.

The record is content-addressed with the same canonical-JSON SHA-256 hashing the assurance-case bundle and the NERC PRC oscillation evidence use (:func:~scpn_phase_orchestrator.assurance._hashing.canonical_record_hash), so a sealed alarm can be referenced by a stable digest and any later mutation is detectable. The capture timestamp and the ground-truth onset are supplied by the caller (they are properties of the measured event, not wall-clock readings taken here) so the record is deterministic and reproducible.

:func:seal_early_warning is the neutral primitive — it depends only on the alarm decision, the provenance, and a pre-extracted set of :class:EarlyWarningIndicator contributions, so it seals any present or future detector (including the real-EEG harness) without importing detector internals. The three seal_*_alarm adapters bridge each concrete suite detector's warning dataclass onto that primitive.

References

  • Scheffer et al. 2009, Nature 461, 53 — generic early-warning signals for critical transitions (the framework the sealed indicators contribute to).

Classes

EarlyWarningIndicator dataclass 

EarlyWarningIndicator(
    name: str,
    direction: str,
    robust_z: float,
    baseline_median: float,
    z_threshold: float,
    breached: bool,
)

A single indicator's contribution to an early-warning alarm.

Attributes

name : str Indicator label, e.g. variance, lag1_autocorrelation, order_parameter, or transition_entropy. direction : str :data:RISE if the indicator warns by rising above its baseline, or :data:DROP if it warns by falling below it. robust_z : float Median / MAD robust z-score of the indicator at the reported window (the alarm window if the detector triggered, else the closest approach among the post-baseline windows). baseline_median : float Median of the indicator over the leading baseline windows. z_threshold : float Robust z-score magnitude at or beyond which the indicator breaches its gate. breached : bool Whether robust_z crossed the gate in the indicator's alarm direction at the reported window.

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

Return a JSON-safe mapping of the indicator contribution.

Returns

dict[str, object] The indicator label, alarm direction, robust z-score, baseline median, gate threshold, and breach status.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the indicator contribution.

    Returns
    -------
    dict[str, object]
        The indicator label, alarm direction, robust z-score, baseline
        median, gate threshold, and breach status.
    """
    return {
        "name": self.name,
        "direction": self.direction,
        "robust_z": self.robust_z,
        "baseline_median": self.baseline_median,
        "z_threshold": self.z_threshold,
        "breached": self.breached,
    }

EarlyWarningEvidence dataclass 

EarlyWarningEvidence(
    detector: str,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window: int,
    step: int,
    persistence: int,
    n_baseline_windows: int,
    warning_triggered: bool,
    warning_window: int | None,
    warning_sample: int | None,
    transition_onset_sample: int | None,
    lead_samples: int | None,
    lead_seconds: float | None,
    lead_is_early: bool,
    indicators: tuple[EarlyWarningIndicator, ...],
    verdict: str,
    framework: str,
    disclaimer: str,
)

A hash-sealed, review-only early-warning assurance record.

Attributes

detector : str Detector family label, e.g. critical_slowing_down, synchronisation, or transition_entropy. observable : str The physical quantity the detector read (a bus-frequency variance, a cross-channel order parameter, a per-channel phase field, ...). signal_source : str Provenance identifier of the screened signal (dataset, event, or channel set). captured_at : str Measurement timestamp of the event, supplied by the caller. sampling_rate_hz : float Sampling rate of the screened signal, in hertz; converts a sample lead into seconds. window, step : int Echoed analysis window length and hop, in samples. persistence : int Echoed number of consecutive breaching windows required to alarm. n_baseline_windows : int Number of leading windows the detector fitted its baseline on. warning_triggered : bool Whether the detector raised a sustained alarm. warning_window : int | None Index of the first window of the triggering run, or None. warning_sample : int | None Sample index of the triggering window, or None. transition_onset_sample : int | None Caller-supplied ground-truth onset sample, or None when unknown. lead_samples : int | None transition_onset_sample - warning_sample when both are known, else None. Positive means an early alarm; non-positive means late or coincident. lead_seconds : float | None lead_samples / sampling_rate_hz when defined, else None. lead_is_early : bool True only when the lead is defined and strictly positive. indicators : tuple[EarlyWarningIndicator, ...] Per-indicator contributions at the reported window. verdict : str :data:EARLY_WARNING_FLAGGED if the detector alarmed, else :data:NO_EARLY_WARNING. framework : str The early-warning framework the record maps to. disclaimer : str The review-only claim boundary. content_hash : str SHA-256 of the canonical record (excluding this field); computed on construction.

Methods:
__post_init__ 
__post_init__() -> None

Compute the content hash from the canonical evidence payload.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
240
241
242
243
244
def __post_init__(self) -> None:
    """Compute the content hash from the canonical evidence payload."""
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the whole sealed record.

Returns

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

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
273
274
275
276
277
278
279
280
281
282
283
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the whole sealed record.

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

Functions:

seal_early_warning 

seal_early_warning(
    *,
    detector: str,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window: int,
    step: int,
    persistence: int,
    n_baseline_windows: int,
    warning_triggered: bool,
    warning_window: int | None,
    warning_sample: int | None,
    indicators: Sequence[EarlyWarningIndicator],
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence

Seal an early-warning alarm into a hash-addressed evidence record.

This is the neutral primitive: it depends only on the alarm decision, the provenance, and a pre-extracted set of indicator contributions, so it seals any detector without importing its internals.

Parameters

detector : str Detector family label. observable : str The physical quantity the detector read. signal_source : str Provenance identifier of the screened signal. captured_at : str Measurement timestamp of the event, supplied by the caller. sampling_rate_hz : float Sampling rate of the screened signal, in hertz (> 0). window, step, persistence : int Echoed analysis parameters; each must be a positive integer. n_baseline_windows : int Number of leading windows the detector fitted its baseline on; must be a positive integer. warning_triggered : bool Whether the detector raised a sustained alarm. warning_window, warning_sample : int | None Triggering window and sample indices; both must be present when warning_triggered is true and absent otherwise. indicators : Sequence[EarlyWarningIndicator] At least one indicator contribution; each direction must be :data:RISE or :data:DROP and each numeric field finite. transition_onset_sample : int | None Caller-supplied ground-truth onset sample; enables the lead computation.

Returns

EarlyWarningEvidence The hash-sealed, review-only early-warning record.

Raises

ValueError If an identifier is empty, a count is not a positive integer, a sample index is negative, the alarm flags are inconsistent, the indicators are empty or malformed, or the sampling rate is not positive.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
def seal_early_warning(
    *,
    detector: str,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window: int,
    step: int,
    persistence: int,
    n_baseline_windows: int,
    warning_triggered: bool,
    warning_window: int | None,
    warning_sample: int | None,
    indicators: Sequence[EarlyWarningIndicator],
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence:
    """Seal an early-warning alarm into a hash-addressed evidence record.

    This is the neutral primitive: it depends only on the alarm decision, the
    provenance, and a pre-extracted set of indicator contributions, so it seals
    any detector without importing its internals.

    Parameters
    ----------
    detector : str
        Detector family label.
    observable : str
        The physical quantity the detector read.
    signal_source : str
        Provenance identifier of the screened signal.
    captured_at : str
        Measurement timestamp of the event, supplied by the caller.
    sampling_rate_hz : float
        Sampling rate of the screened signal, in hertz (``> 0``).
    window, step, persistence : int
        Echoed analysis parameters; each must be a positive integer.
    n_baseline_windows : int
        Number of leading windows the detector fitted its baseline on; must be a
        positive integer.
    warning_triggered : bool
        Whether the detector raised a sustained alarm.
    warning_window, warning_sample : int | None
        Triggering window and sample indices; both must be present when
        ``warning_triggered`` is true and absent otherwise.
    indicators : Sequence[EarlyWarningIndicator]
        At least one indicator contribution; each ``direction`` must be
        :data:`RISE` or :data:`DROP` and each numeric field finite.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample; enables the lead computation.

    Returns
    -------
    EarlyWarningEvidence
        The hash-sealed, review-only early-warning record.

    Raises
    ------
    ValueError
        If an identifier is empty, a count is not a positive integer, a sample
        index is negative, the alarm flags are inconsistent, the indicators are
        empty or malformed, or the sampling rate is not positive.
    """
    detector_label = _non_empty_str(detector, "detector")
    observable_label = _non_empty_str(observable, "observable")
    source = _non_empty_str(signal_source, "signal_source")
    captured = _non_empty_str(captured_at, "captured_at")
    fs = _positive_real(sampling_rate_hz, "sampling_rate_hz")
    window_int = _positive_int(window, "window")
    step_int = _positive_int(step, "step")
    persistence_int = _positive_int(persistence, "persistence")
    n_baseline = _positive_int(n_baseline_windows, "n_baseline_windows")
    triggered = _bool(warning_triggered, "warning_triggered")
    window_idx = _optional_non_negative_int(warning_window, "warning_window")
    sample_idx = _optional_non_negative_int(warning_sample, "warning_sample")
    onset = _optional_non_negative_int(
        transition_onset_sample, "transition_onset_sample"
    )
    if triggered != (window_idx is not None) or triggered != (sample_idx is not None):
        raise ValueError(
            "warning_window and warning_sample must be present exactly when "
            "warning_triggered is true"
        )
    sealed_indicators = _validate_indicators(indicators)

    lead_samples, lead_seconds, lead_is_early = _lead(onset, sample_idx, fs)
    verdict = EARLY_WARNING_FLAGGED if triggered else NO_EARLY_WARNING

    return EarlyWarningEvidence(
        detector=detector_label,
        observable=observable_label,
        signal_source=source,
        captured_at=captured,
        sampling_rate_hz=fs,
        window=window_int,
        step=step_int,
        persistence=persistence_int,
        n_baseline_windows=n_baseline,
        warning_triggered=triggered,
        warning_window=window_idx,
        warning_sample=sample_idx,
        transition_onset_sample=onset,
        lead_samples=lead_samples,
        lead_seconds=lead_seconds,
        lead_is_early=lead_is_early,
        indicators=sealed_indicators,
        verdict=verdict,
        framework=EARLY_WARNING_FRAMEWORK,
        disclaimer=EARLY_WARNING_DISCLAIMER,
    )

seal_critical_slowing_down_alarm 

seal_critical_slowing_down_alarm(
    warning: CriticalSlowingDownWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    transition_onset_sample: int | None = None,
    detector: str = "critical_slowing_down",
) -> EarlyWarningEvidence

Seal a critical-slowing-down alarm, pinning both rising indicators.

The record carries the variance and lag-one autocorrelation contributions — the two second-moment indicators of critical slowing down — at the reported window, so an auditor sees which indicator carried the alarm.

Parameters

warning : CriticalSlowingDownWarning The detector output to seal. observable, signal_source, captured_at : str Provenance of the screened signal, forwarded to :func:seal_early_warning. sampling_rate_hz : float Sampling rate of the screened signal, in hertz. transition_onset_sample : int | None Caller-supplied ground-truth onset sample. detector : str Detector family label to seal; defaults to critical_slowing_down. Pass critical_slowing_down_multiscale when sealing the multi-scale variant so audit records distinguish the two.

Returns

EarlyWarningEvidence The sealed record for the critical-slowing-down alarm.

Raises

ValueError If warning is not a :class:~scpn_phase_orchestrator.monitor.critical_slowing_down.CriticalSlowingDownWarning or the forwarded provenance is invalid.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
def seal_critical_slowing_down_alarm(
    warning: CriticalSlowingDownWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    transition_onset_sample: int | None = None,
    detector: str = "critical_slowing_down",
) -> EarlyWarningEvidence:
    """Seal a critical-slowing-down alarm, pinning both rising indicators.

    The record carries the variance and lag-one autocorrelation contributions —
    the two second-moment indicators of critical slowing down — at the reported
    window, so an auditor sees which indicator carried the alarm.

    Parameters
    ----------
    warning : CriticalSlowingDownWarning
        The detector output to seal.
    observable, signal_source, captured_at : str
        Provenance of the screened signal, forwarded to :func:`seal_early_warning`.
    sampling_rate_hz : float
        Sampling rate of the screened signal, in hertz.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample.
    detector : str
        Detector family label to seal; defaults to ``critical_slowing_down``.
        Pass ``critical_slowing_down_multiscale`` when sealing the multi-scale
        variant so audit records distinguish the two.

    Returns
    -------
    EarlyWarningEvidence
        The sealed record for the critical-slowing-down alarm.

    Raises
    ------
    ValueError
        If ``warning`` is not a
        :class:`~scpn_phase_orchestrator.monitor.critical_slowing_down.CriticalSlowingDownWarning`
        or the forwarded provenance is invalid.
    """
    from scpn_phase_orchestrator.monitor.critical_slowing_down import (
        CriticalSlowingDownWarning,
    )

    if not isinstance(warning, CriticalSlowingDownWarning):
        raise ValueError("warning must be a CriticalSlowingDownWarning")
    report = _report_window(
        warning.combined_z.tolist(),
        RISE,
        warning.n_baseline_windows,
        warning.warning_window,
    )
    indicators = (
        _indicator_at(
            "variance",
            RISE,
            warning.robust_z_variance.tolist(),
            warning.baseline_variance,
            warning.z_threshold,
            report,
        ),
        _indicator_at(
            "lag1_autocorrelation",
            RISE,
            warning.robust_z_autocorrelation.tolist(),
            warning.baseline_autocorrelation,
            warning.z_threshold,
            report,
        ),
    )
    return seal_early_warning(
        detector=detector,
        observable=observable,
        signal_source=signal_source,
        captured_at=captured_at,
        sampling_rate_hz=sampling_rate_hz,
        window=warning.window,
        step=warning.step,
        persistence=warning.persistence,
        n_baseline_windows=warning.n_baseline_windows,
        warning_triggered=warning.warning_triggered,
        warning_window=warning.warning_window,
        warning_sample=warning.warning_sample,
        indicators=indicators,
        transition_onset_sample=transition_onset_sample,
    )

seal_synchronisation_alarm 

seal_synchronisation_alarm(
    warning: SynchronisationWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence

Seal a rising-synchronisation alarm on the Kuramoto order parameter.

Parameters

warning : SynchronisationWarning The detector output to seal. observable, signal_source, captured_at : str Provenance of the screened signal, forwarded to :func:seal_early_warning. sampling_rate_hz : float Sampling rate of the screened signal, in hertz. transition_onset_sample : int | None Caller-supplied ground-truth onset sample.

Returns

EarlyWarningEvidence The sealed record for the synchronisation alarm.

Raises

ValueError If warning is not a :class:~scpn_phase_orchestrator.monitor.synchronisation.SynchronisationWarning or the forwarded provenance is invalid.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
def seal_synchronisation_alarm(
    warning: SynchronisationWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence:
    """Seal a rising-synchronisation alarm on the Kuramoto order parameter.

    Parameters
    ----------
    warning : SynchronisationWarning
        The detector output to seal.
    observable, signal_source, captured_at : str
        Provenance of the screened signal, forwarded to :func:`seal_early_warning`.
    sampling_rate_hz : float
        Sampling rate of the screened signal, in hertz.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample.

    Returns
    -------
    EarlyWarningEvidence
        The sealed record for the synchronisation alarm.

    Raises
    ------
    ValueError
        If ``warning`` is not a
        :class:`~scpn_phase_orchestrator.monitor.synchronisation.SynchronisationWarning`
        or the forwarded provenance is invalid.
    """
    from scpn_phase_orchestrator.monitor.synchronisation import SynchronisationWarning

    if not isinstance(warning, SynchronisationWarning):
        raise ValueError("warning must be a SynchronisationWarning")
    report = _report_window(
        warning.robust_z.tolist(),
        RISE,
        warning.n_baseline_windows,
        warning.warning_window,
    )
    indicators = (
        _indicator_at(
            "order_parameter",
            RISE,
            warning.robust_z.tolist(),
            warning.baseline_median,
            warning.z_threshold,
            report,
        ),
    )
    return seal_early_warning(
        detector="synchronisation",
        observable=observable,
        signal_source=signal_source,
        captured_at=captured_at,
        sampling_rate_hz=sampling_rate_hz,
        window=warning.window,
        step=warning.step,
        persistence=warning.persistence,
        n_baseline_windows=warning.n_baseline_windows,
        warning_triggered=warning.warning_triggered,
        warning_window=warning.warning_window,
        warning_sample=warning.warning_sample,
        indicators=indicators,
        transition_onset_sample=transition_onset_sample,
    )

seal_transition_entropy_alarm 

seal_transition_entropy_alarm(
    warning: ExplosiveSyncWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence

Seal an ordinal-transition-entropy alarm (a regularisation drop).

Parameters

warning : ExplosiveSyncWarning The detector output to seal. observable, signal_source, captured_at : str Provenance of the screened signal, forwarded to :func:seal_early_warning. sampling_rate_hz : float Sampling rate of the screened signal, in hertz. transition_onset_sample : int | None Caller-supplied ground-truth onset sample.

Returns

EarlyWarningEvidence The sealed record for the transition-entropy alarm.

Raises

ValueError If warning is not an :class:~scpn_phase_orchestrator.monitor.explosive_sync.ExplosiveSyncWarning or the forwarded provenance is invalid.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
def seal_transition_entropy_alarm(
    warning: ExplosiveSyncWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence:
    """Seal an ordinal-transition-entropy alarm (a regularisation drop).

    Parameters
    ----------
    warning : ExplosiveSyncWarning
        The detector output to seal.
    observable, signal_source, captured_at : str
        Provenance of the screened signal, forwarded to :func:`seal_early_warning`.
    sampling_rate_hz : float
        Sampling rate of the screened signal, in hertz.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample.

    Returns
    -------
    EarlyWarningEvidence
        The sealed record for the transition-entropy alarm.

    Raises
    ------
    ValueError
        If ``warning`` is not an
        :class:`~scpn_phase_orchestrator.monitor.explosive_sync.ExplosiveSyncWarning`
        or the forwarded provenance is invalid.
    """
    from scpn_phase_orchestrator.monitor.explosive_sync import ExplosiveSyncWarning

    if not isinstance(warning, ExplosiveSyncWarning):
        raise ValueError("warning must be an ExplosiveSyncWarning")
    report = _report_window(
        warning.robust_z.tolist(),
        DROP,
        warning.n_baseline_windows,
        warning.warning_window,
    )
    indicators = (
        _indicator_at(
            "transition_entropy",
            DROP,
            warning.robust_z.tolist(),
            warning.baseline_median,
            warning.z_threshold,
            report,
        ),
    )
    return seal_early_warning(
        detector="transition_entropy",
        observable=observable,
        signal_source=signal_source,
        captured_at=captured_at,
        sampling_rate_hz=sampling_rate_hz,
        window=warning.window,
        step=warning.step,
        persistence=warning.persistence,
        n_baseline_windows=warning.n_baseline_windows,
        warning_triggered=warning.warning_triggered,
        warning_window=warning.warning_window,
        warning_sample=warning.warning_sample,
        indicators=indicators,
        transition_onset_sample=transition_onset_sample,
    )

seal_ensemble_alarm 

seal_ensemble_alarm(
    ensemble: EnsembleWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window: int,
    step: int,
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence

Seal a fused ensemble alarm, pinning every member's contribution.

Each fused member becomes an indicator carrying its native robust z-score at the reported window, so an auditor sees exactly which detectors drove — or failed to drive — the fused decision. The suite is run on one window grid, so window and step are supplied by the caller that ran it.

Parameters

ensemble : EnsembleWarning The fused decision to seal. observable, signal_source, captured_at : str Provenance of the screened signal, forwarded to :func:seal_early_warning. sampling_rate_hz : float Sampling rate of the screened signal, in hertz. window, step : int Analysis window length and hop the suite was run with. transition_onset_sample : int | None Caller-supplied ground-truth onset sample.

Returns

EarlyWarningEvidence The sealed record for the fused ensemble alarm.

Raises

ValueError If ensemble is not an :class:~scpn_phase_orchestrator.monitor.ensemble_warning.EnsembleWarning or the forwarded provenance is invalid.

Source code in src/scpn_phase_orchestrator/assurance/early_warning_evidence.py 
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
def seal_ensemble_alarm(
    ensemble: EnsembleWarning,
    *,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window: int,
    step: int,
    transition_onset_sample: int | None = None,
) -> EarlyWarningEvidence:
    """Seal a fused ensemble alarm, pinning every member's contribution.

    Each fused member becomes an indicator carrying its native robust z-score at
    the reported window, so an auditor sees exactly which detectors drove — or
    failed to drive — the fused decision. The suite is run on one window grid, so
    ``window`` and ``step`` are supplied by the caller that ran it.

    Parameters
    ----------
    ensemble : EnsembleWarning
        The fused decision to seal.
    observable, signal_source, captured_at : str
        Provenance of the screened signal, forwarded to :func:`seal_early_warning`.
    sampling_rate_hz : float
        Sampling rate of the screened signal, in hertz.
    window, step : int
        Analysis window length and hop the suite was run with.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample.

    Returns
    -------
    EarlyWarningEvidence
        The sealed record for the fused ensemble alarm.

    Raises
    ------
    ValueError
        If ``ensemble`` is not an
        :class:`~scpn_phase_orchestrator.monitor.ensemble_warning.EnsembleWarning`
        or the forwarded provenance is invalid.
    """
    from scpn_phase_orchestrator.monitor.ensemble_warning import EnsembleWarning

    if not isinstance(ensemble, EnsembleWarning):
        raise ValueError("ensemble must be an EnsembleWarning")
    indicators = tuple(
        EarlyWarningIndicator(
            name=contribution.name,
            direction=contribution.direction,
            robust_z=contribution.robust_z,
            baseline_median=contribution.baseline_median,
            z_threshold=contribution.z_threshold,
            breached=contribution.breached,
        )
        for contribution in ensemble.contributions
    )
    return seal_early_warning(
        detector=f"ensemble_{ensemble.rule}",
        observable=observable,
        signal_source=signal_source,
        captured_at=captured_at,
        sampling_rate_hz=sampling_rate_hz,
        window=window,
        step=step,
        persistence=ensemble.persistence,
        n_baseline_windows=ensemble.n_baseline_windows,
        warning_triggered=ensemble.warning_triggered,
        warning_window=ensemble.warning_window,
        warning_sample=ensemble.warning_sample,
        indicators=indicators,
        transition_onset_sample=transition_onset_sample,
    )

Grid Early-Warning Advisory

scpn_phase_orchestrator.assurance.grid_early_warning_advisory is the step from a live grid instability alarm to an operator decision surface — the pinnacle the streaming monitor was built toward — done honestly. When the certified streaming monitor raises a StreamAlarm, this module turns it into a claim-bounded advisory record: the growth rate σ that crossed the certified threshold, the most-unstable bus, the alarm time, the certified operating point, and — as a first-class sealed field — the detector's honest recall, so the reader knows how much the detector misses.

The advisory is passive and review-only. It never actuates: every record carries non_actuating = True and actuating = False, the fail-closed stance of the STL runtime actuation gate. The sealed recall is the point: at its certified streaming operating point the detector leads only about a quarter of growing-instability episodes at a matched ten-percent stream false alarm, so an advisory is a reason to look, never a guarantee, and the absence of an advisory is not evidence of stability. seal_grid_early_warning_advisory is the neutral primitive; advise_from_stream_alarm reads the alarm claims and the certified operating point straight off a live monitor. The record is content-addressed with the same canonical-JSON SHA-256 seal, so any later mutation is detectable.

grid_early_warning_advisory

A hash-sealed, review-only operator advisory for a live grid instability alarm.

This is the step from a live alarm to a decision surface — the pinnacle the streaming monitor was built toward — done honestly. When the certified streaming monitor (:class:~scpn_phase_orchestrator.monitor.grid_modal_stream.GridModalStreamMonitor) raises a :class:~scpn_phase_orchestrator.monitor.grid_modal_stream.StreamAlarm, this module turns it into a claim-bounded advisory record an operator can read: the growth rate σ that crossed the certified threshold, the most-unstable bus, the alarm time, the certified operating point, and — as a first-class sealed field — the detector's honest recall, so the reader knows how much the detector misses.

The advisory is passive and review-only. It never actuates: every record carries non_actuating = True and actuating = False, the same fail-closed stance the STL runtime actuation gate takes. It exists to inform a human decision, not to make one. The sealed recall is the point: at its certified streaming operating point the detector leads only about a quarter of growing-instability episodes at a matched ten-percent stream false alarm, so an advisory is a reason to look, never a guarantee, and — critically — the absence of an advisory is not evidence of stability.

The record is content-addressed with the same canonical-JSON SHA-256 seal the early-warning evidence, the assurance-case bundle, and the NERC PRC oscillation evidence use (:func:~scpn_phase_orchestrator.assurance._hashing.canonical_record_hash). The capture timestamp and any ground-truth onset are supplied by the caller (they are properties of the measured event, not wall-clock readings taken here), so the record is deterministic and reproducible; a reported lead is honest, including a non-positive lead when the alarm was coincident with or later than the onset.

:func:seal_grid_early_warning_advisory is the neutral primitive; :func:advise_from_stream_alarm is the thin adapter that reads the alarm claims and the certified operating point straight off a live monitor.

References

  • Kundur 1994, Power System Stability and Control — small-signal (modal) stability: the growth rate σ the advisory surfaces is the dominant mode's eigenvalue.

Classes

GridEarlyWarningAdvisory dataclass 

GridEarlyWarningAdvisory(
    detector: str,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window_seconds: float,
    step_seconds: float,
    persistence: int,
    aggregation: str,
    recency_top: float,
    r2_gate: float,
    warning_sample: int,
    warning_time_s: float,
    growth_rate: float,
    growth_rate_threshold: float,
    most_unstable_bus: int,
    transition_onset_sample: int | None,
    lead_samples: int | None,
    lead_seconds: float | None,
    lead_is_early: bool,
    certified_recall: float,
    certified_false_alarm: float,
    certified_operating_point: str,
    non_actuating: bool,
    actuating: bool,
    verdict: str,
    framework: str,
    disclaimer: str,
)

A hash-sealed, review-only grid early-warning operator advisory.

Attributes

detector : str The detector family label, e.g. grid_modal_growth_stream. observable : str The physical quantity the detector read. signal_source : str Provenance identifier of the live stream (feed, event, or scenario). captured_at : str Measurement timestamp of the alarm, supplied by the caller. sampling_rate_hz : float Stream sampling rate in hertz. window_seconds, step_seconds : float The certified streaming operating point: window length and re-scoring hop. persistence : int Consecutive above-threshold windows required before the alarm fired. aggregation : str The certified aggregation ("focal" or "mean"). recency_top : float The certified recency weighting the growth rate was fitted under. r2_gate : float The certified fit-quality gate; 0.0 when off. warning_sample : int Stream sample index the alarm fired at. warning_time_s : float Alarm time in seconds from the stream start. growth_rate : float The growth rate σ at the alarm window. growth_rate_threshold : float The certified matched-false-alarm threshold σ crossed. most_unstable_bus : int The most-unstable bus, or :data:WHOLE_NETWORK_BUS under the mean aggregation. transition_onset_sample : int | None Caller-supplied ground-truth onset sample, or None when unknown. lead_samples : int | None transition_onset_sample - warning_sample when both are known, else None. lead_seconds : float | None lead_samples / sampling_rate_hz when defined, else None. lead_is_early : bool True only when the lead is defined and strictly positive. certified_recall : float The honest fraction of growing-instability episodes the detector leads at the certified operating point — sealed so the operator sees the miss rate. certified_false_alarm : float The matched stream false-alarm rate the threshold was certified at. certified_operating_point : str Provenance of the certified operating point (the sealed artefact it came from). non_actuating : bool Always True: the advisory never actuates. actuating : bool Always False: no actuation path exists. verdict : str :data:GRID_ADVISORY_RAISED. framework : str The stability framework the advisory maps to. disclaimer : str The review-only claim boundary. content_hash : str SHA-256 of the canonical record (excluding this field); set on construction.

Methods:
__post_init__ 
__post_init__() -> None

Compute the content hash from the canonical advisory payload.

Source code in src/scpn_phase_orchestrator/assurance/grid_early_warning_advisory.py 
195
196
197
198
199
def __post_init__(self) -> None:
    """Compute the content hash from the canonical advisory payload."""
    object.__setattr__(
        self, "content_hash", canonical_record_hash(self._canonical_payload())
    )
to_audit_record 
to_audit_record() -> dict[str, object]

Return a JSON-safe mapping of the whole sealed advisory.

Returns

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

Source code in src/scpn_phase_orchestrator/assurance/grid_early_warning_advisory.py 
234
235
236
237
238
239
240
241
242
243
244
def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe mapping of the whole sealed advisory.

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

Functions:

seal_grid_early_warning_advisory 

seal_grid_early_warning_advisory(
    *,
    detector: str,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window_seconds: float,
    step_seconds: float,
    persistence: int,
    aggregation: str,
    recency_top: float,
    r2_gate: float,
    warning_sample: int,
    warning_time_s: float,
    growth_rate: float,
    growth_rate_threshold: float,
    most_unstable_bus: int,
    certified_recall: float,
    certified_false_alarm: float,
    certified_operating_point: str,
    transition_onset_sample: int | None = None,
) -> GridEarlyWarningAdvisory

Seal a live grid instability alarm into a hash-addressed, review-only advisory.

The neutral primitive: it depends only on the alarm claims and the certified operating point, so it seals any monitor configuration without importing its internals. It hard-wires the non-actuating stance and computes the honest lead.

Parameters

detector, observable, signal_source, captured_at, certified_operating_point : str The detector label, the read quantity, the stream provenance, the caller-supplied timestamp, and the operating-point provenance; each non-empty. sampling_rate_hz, window_seconds, step_seconds : float The stream rate and the operating-point window and hop in seconds; each > 0. persistence : int The certified persistence; a positive integer. aggregation : str "focal" or "mean". recency_top : float The certified recency weighting; a finite number >= 1. r2_gate, certified_recall, certified_false_alarm : float The certified fit-quality gate and the honest recall and false-alarm rate; each a finite number in [0, 1]. warning_sample : int The alarm's stream sample index; non-negative. warning_time_s, growth_rate, growth_rate_threshold : float The alarm time and the growth rate and threshold; each finite. most_unstable_bus : int The most-unstable bus, or :data:WHOLE_NETWORK_BUS under the mean aggregation. transition_onset_sample : int | None Caller-supplied ground-truth onset sample; enables the lead computation.

Returns

GridEarlyWarningAdvisory The hash-sealed, review-only advisory.

Raises

ValueError If an identifier is empty, a rate or window is not positive, persistence is not a positive integer, aggregation is unknown, a bounded rate leaves [0, 1], recency_top is below one, a sample index is negative, the bus is below the whole-network sentinel, or a reported real is not finite.

Source code in src/scpn_phase_orchestrator/assurance/grid_early_warning_advisory.py 
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
def seal_grid_early_warning_advisory(
    *,
    detector: str,
    observable: str,
    signal_source: str,
    captured_at: str,
    sampling_rate_hz: float,
    window_seconds: float,
    step_seconds: float,
    persistence: int,
    aggregation: str,
    recency_top: float,
    r2_gate: float,
    warning_sample: int,
    warning_time_s: float,
    growth_rate: float,
    growth_rate_threshold: float,
    most_unstable_bus: int,
    certified_recall: float,
    certified_false_alarm: float,
    certified_operating_point: str,
    transition_onset_sample: int | None = None,
) -> GridEarlyWarningAdvisory:
    """Seal a live grid instability alarm into a hash-addressed, review-only advisory.

    The neutral primitive: it depends only on the alarm claims and the certified
    operating point, so it seals any monitor configuration without importing its
    internals. It hard-wires the non-actuating stance and computes the honest lead.

    Parameters
    ----------
    detector, observable, signal_source, captured_at, certified_operating_point : str
        The detector label, the read quantity, the stream provenance, the
        caller-supplied timestamp, and the operating-point provenance; each non-empty.
    sampling_rate_hz, window_seconds, step_seconds : float
        The stream rate and the operating-point window and hop in seconds; each ``> 0``.
    persistence : int
        The certified persistence; a positive integer.
    aggregation : str
        ``"focal"`` or ``"mean"``.
    recency_top : float
        The certified recency weighting; a finite number ``>= 1``.
    r2_gate, certified_recall, certified_false_alarm : float
        The certified fit-quality gate and the honest recall and false-alarm rate; each
        a finite number in ``[0, 1]``.
    warning_sample : int
        The alarm's stream sample index; non-negative.
    warning_time_s, growth_rate, growth_rate_threshold : float
        The alarm time and the growth rate and threshold; each finite.
    most_unstable_bus : int
        The most-unstable bus, or :data:`WHOLE_NETWORK_BUS` under the mean aggregation.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample; enables the lead computation.

    Returns
    -------
    GridEarlyWarningAdvisory
        The hash-sealed, review-only advisory.

    Raises
    ------
    ValueError
        If an identifier is empty, a rate or window is not positive, ``persistence`` is
        not a positive integer, ``aggregation`` is unknown, a bounded rate leaves
        ``[0, 1]``, ``recency_top`` is below one, a sample index is negative, the bus is
        below the whole-network sentinel, or a reported real is not finite.
    """
    fs = _positive_real(sampling_rate_hz, "sampling_rate_hz")
    onset = _optional_non_negative_int(
        transition_onset_sample, "transition_onset_sample"
    )
    sample_idx = _non_negative_int(warning_sample, "warning_sample")
    if aggregation not in ("focal", "mean"):
        raise ValueError(f"aggregation must be 'mean' or 'focal', got {aggregation!r}")
    bus = _finite_int(most_unstable_bus, "most_unstable_bus")
    if bus < WHOLE_NETWORK_BUS:
        raise ValueError(f"most_unstable_bus must be >= {WHOLE_NETWORK_BUS}, got {bus}")
    lead_samples = None if onset is None else onset - sample_idx
    lead_seconds = None if lead_samples is None else lead_samples / fs
    lead_is_early = lead_samples is not None and lead_samples > 0

    return GridEarlyWarningAdvisory(
        detector=_non_empty_str(detector, "detector"),
        observable=_non_empty_str(observable, "observable"),
        signal_source=_non_empty_str(signal_source, "signal_source"),
        captured_at=_non_empty_str(captured_at, "captured_at"),
        sampling_rate_hz=fs,
        window_seconds=_positive_real(window_seconds, "window_seconds"),
        step_seconds=_positive_real(step_seconds, "step_seconds"),
        persistence=_positive_int(persistence, "persistence"),
        aggregation=aggregation,
        recency_top=_recency(recency_top),
        r2_gate=_unit_interval(r2_gate, "r2_gate"),
        warning_sample=sample_idx,
        warning_time_s=_finite_real(warning_time_s, "warning_time_s"),
        growth_rate=_finite_real(growth_rate, "growth_rate"),
        growth_rate_threshold=_finite_real(
            growth_rate_threshold, "growth_rate_threshold"
        ),
        most_unstable_bus=bus,
        transition_onset_sample=onset,
        lead_samples=lead_samples,
        lead_seconds=lead_seconds,
        lead_is_early=lead_is_early,
        certified_recall=_unit_interval(certified_recall, "certified_recall"),
        certified_false_alarm=_unit_interval(
            certified_false_alarm, "certified_false_alarm"
        ),
        certified_operating_point=_non_empty_str(
            certified_operating_point, "certified_operating_point"
        ),
        non_actuating=True,
        actuating=False,
        verdict=GRID_ADVISORY_RAISED,
        framework=GRID_EARLY_WARNING_FRAMEWORK,
        disclaimer=GRID_EARLY_WARNING_DISCLAIMER,
    )

advise_from_stream_alarm 

advise_from_stream_alarm(
    alarm: StreamAlarm,
    monitor: GridModalStreamMonitor,
    *,
    signal_source: str,
    captured_at: str,
    certified_recall: float,
    certified_false_alarm: float,
    certified_operating_point: str,
    observable: str = GRID_EARLY_WARNING_OBSERVABLE,
    detector: str = "grid_modal_growth_stream",
    transition_onset_sample: int | None = None,
) -> GridEarlyWarningAdvisory

Seal an advisory straight from a live monitor's alarm and operating point.

Reads the alarm claims (growth rate, threshold, most-unstable bus, sample and time) off the :class:~scpn_phase_orchestrator.monitor.grid_modal_stream.StreamAlarm and the certified operating point (rate, window, step, persistence, aggregation, recency weighting, gate) off the live monitor, so the advisory records exactly what fired with no hand-set constants beyond the caller-supplied provenance and honest rates.

Parameters

alarm : StreamAlarm The lead event the monitor raised. monitor : GridModalStreamMonitor The monitor that raised it, read for its certified operating point. signal_source, captured_at, certified_operating_point : str The stream provenance, the caller-supplied capture timestamp, and the certified-operating-point provenance. certified_recall, certified_false_alarm : float The honest recall and matched false-alarm rate of the certified operating point. observable, detector : str The read-quantity and detector labels sealed into the record. transition_onset_sample : int | None Caller-supplied ground-truth onset sample; enables the lead computation.

Returns

GridEarlyWarningAdvisory The hash-sealed, review-only advisory for the alarm.

Source code in src/scpn_phase_orchestrator/assurance/grid_early_warning_advisory.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
def advise_from_stream_alarm(
    alarm: StreamAlarm,
    monitor: GridModalStreamMonitor,
    *,
    signal_source: str,
    captured_at: str,
    certified_recall: float,
    certified_false_alarm: float,
    certified_operating_point: str,
    observable: str = GRID_EARLY_WARNING_OBSERVABLE,
    detector: str = "grid_modal_growth_stream",
    transition_onset_sample: int | None = None,
) -> GridEarlyWarningAdvisory:
    """Seal an advisory straight from a live monitor's alarm and operating point.

    Reads the alarm claims (growth rate, threshold, most-unstable bus, sample and time)
    off the :class:`~scpn_phase_orchestrator.monitor.grid_modal_stream.StreamAlarm` and
    the certified operating point (rate, window, step, persistence, aggregation, recency
    weighting, gate) off the live monitor, so the advisory records exactly what fired
    with no hand-set constants beyond the caller-supplied provenance and honest rates.

    Parameters
    ----------
    alarm : StreamAlarm
        The lead event the monitor raised.
    monitor : GridModalStreamMonitor
        The monitor that raised it, read for its certified operating point.
    signal_source, captured_at, certified_operating_point : str
        The stream provenance, the caller-supplied capture timestamp, and the
        certified-operating-point provenance.
    certified_recall, certified_false_alarm : float
        The honest recall and matched false-alarm rate of the certified operating point.
    observable, detector : str
        The read-quantity and detector labels sealed into the record.
    transition_onset_sample : int | None
        Caller-supplied ground-truth onset sample; enables the lead computation.

    Returns
    -------
    GridEarlyWarningAdvisory
        The hash-sealed, review-only advisory for the alarm.
    """
    return seal_grid_early_warning_advisory(
        detector=detector,
        observable=observable,
        signal_source=signal_source,
        captured_at=captured_at,
        sampling_rate_hz=monitor.rate,
        window_seconds=monitor.window_seconds,
        step_seconds=monitor.step_seconds,
        persistence=monitor.persistence,
        aggregation=monitor.aggregation,
        recency_top=monitor.recency_top,
        r2_gate=monitor.r2_gate,
        warning_sample=alarm.sample_index,
        warning_time_s=alarm.time_s,
        growth_rate=alarm.score,
        growth_rate_threshold=alarm.threshold,
        most_unstable_bus=alarm.bus,
        certified_recall=certified_recall,
        certified_false_alarm=certified_false_alarm,
        certified_operating_point=certified_operating_point,
        transition_onset_sample=transition_onset_sample,
    )