Monitor — Grid Modal Stream

A causal, real-time streaming monitor that carries the certified grid modal-growth detector online — the step from an offline benchmark to an operational early warning.

What it is for

The offline head-to-head certifies the detector on fixed pre-onset segments. This monitor runs the same detector causally on a live stream: it keeps a sliding window of the most recent per-bus voltage samples and, every step, re-scores that window with the identical primitives the detector uses, so the streaming score on a window is bit-for-bit the offline modal_growth_score on that same window. That identity is what makes the offline-calibrated threshold valid online — the monitor never recalibrates.

Contract

• GridModalStreamMonitor — pushes one per-bus sample at a time (update) and raises a StreamAlarm when the live growth rate crosses the certified threshold, after an optional persistence debounce, latching until the growth falls back below so each instability episode raises one lead event.
• StreamAlarm — a lead event: the sample index and time it fired, the growth rate σ, the threshold crossed, and the most unstable bus.
• GridModalStreamMonitor.from_evidence — builds the monitor straight from a sealed head-to-head artefact (aggregation, recency weighting, and matched-false-alarm threshold from the certification), so the certified detector becomes the live monitor with no hand-set constants.

Each update is transactional: a sample must be one non-empty finite real bus vector, free of text, boolean, complex, and broken-protocol aliases, and its bus count must match the established stream. The monitor copies validated samples before mutating its index or buffer, so a rejected PMU frame cannot poison a later score. Constructor controls and emitted alarm scalars are finite, non-coercive primitives; persistence is an exact positive integer.

The honest live-deployment operating point — a stream is stricter than the pre-onset window, because a damped fault has a transient growth window the continuous monitor also alarms on — is measured and sealed in examples/real_data/psml_modal_growth/.

grid_modal_stream

A causal, real-time streaming monitor for the certified grid modal-growth detector.

The offline head-to-head (bench.grid_modal_head_to_head) certifies the grid modal detector (monitor.grid_modal_growth): on real PMU data, at a matched false alarm, its growth-rate σ leads instability transitions far more than chance. That certification is done on fixed pre-onset segments — the training/validation harness. This module is the step to the pinnacle: it runs the same detector causally on a live stream, so the certified operating point becomes an operational early warning.

The monitor keeps a sliding window of the most recent per-bus voltage samples and, every step, re-scores that window with the identical primitives the detector uses (per_bus_deviation, envelope_growth_rate) — so the streaming score on a window is bit-for-bit the offline modal_growth_score on that same window. That identity is what makes the offline-calibrated threshold valid online: the monitor never recalibrates, it carries the certified threshold and fires when the live σ crosses it (after an optional persistence debounce), latching until σ falls back below so each instability episode raises one lead event.

:meth:GridModalStreamMonitor.from_evidence closes the loop for the offline per-window operating point: it builds the monitor from a sealed head-to-head artefact, taking the aggregation, recency weighting, and matched-false-alarm threshold straight from the certification — the certified detector becomes the live monitor with no hand-set constants.

A live stream is stricter than the pre-onset window, because a damped fault has a transient growth window the continuous monitor also alarms on, so the plain per-window threshold over-alarms online. The r2_gate carries the fit-quality gate (grid_modal_growth.fit_gated_growth_rate) into the live scoring: it rejects a fault's step-like transient and holds the stream false alarm at target. :meth:GridModalStreamMonitor.from_stream_evidence builds the monitor at the streaming winner — window, step, persistence, threshold, and gate — read straight from a sealed streaming operating-point artefact.

Classes

StreamAlarm dataclass

StreamAlarm(
    sample_index: int,
    time_s: float,
    score: float,
    threshold: float,
    bus: int,
)

A lead event raised when the live growth rate σ crosses the threshold.

Attributes

sample_index : int The stream sample index the alarm fired at (the last sample in the window). time_s : float sample_index / rate — the alarm time in seconds from the stream start. score : float The modal growth rate σ at the alarm window. threshold : float The certified matched-false-alarm threshold σ crossed. bus : int The most unstable bus under the focal aggregation (its per-bus σ is the maximum), or :data:WHOLE_NETWORK_BUS under the whole-network aggregation.

GridModalStreamMonitor

GridModalStreamMonitor(
    *,
    rate: float,
    threshold: float,
    window_seconds: float = 2.0,
    step_seconds: float = 0.5,
    aggregation: str = DEFAULT_AGGREGATION,
    recency_top: float = DEFAULT_RECENCY_TOP,
    persistence: int = 1,
    r2_gate: float = 0.0,
)

A causal sliding-window monitor carrying the certified grid detector online.

Parameters

rate : float Sampling rate in Hz; must be positive and finite. threshold : float The certified matched-false-alarm growth-rate threshold; σ at or above it fires. window_seconds : float Sliding-window length in seconds, matching the offline segment length. step_seconds : float How often the window is re-scored, in seconds; must be positive. aggregation : str "focal" (most unstable bus) or "mean" (whole network), as certified. recency_top : float The recency weighting, as certified. persistence : int Consecutive re-scored windows at or above the threshold required before an alarm fires; 1 fires on the first crossing. Must be a positive integer. r2_gate : float Fit-quality gate in [0, 1] applied to each envelope's growth rate before aggregating (see grid_modal_growth.fit_gated_growth_rate). 0.0 (the default) disables the gate, so the live score stays bit-for-bit the offline modal_growth_score; the certified streaming operating point uses a positive gate to reject a fault's step-like transient.

Raises

ValueError If rate/window_seconds/step_seconds are not positive finite, the window is shorter than two samples, persistence is below one, aggregation is neither "focal" nor "mean", or r2_gate is not a finite number in [0, 1].

Source code in src/scpn_phase_orchestrator/monitor/grid_modal_stream.py

def __init__(
    self,
    *,
    rate: float,
    threshold: float,
    window_seconds: float = 2.0,
    step_seconds: float = 0.5,
    aggregation: str = DEFAULT_AGGREGATION,
    recency_top: float = DEFAULT_RECENCY_TOP,
    persistence: int = 1,
    r2_gate: float = 0.0,
) -> None:
    validated_rate = _positive_real(rate, "rate")
    validated_threshold = _finite_real(threshold, "threshold")
    validated_window_seconds = _positive_real(window_seconds, "window_seconds")
    validated_step_seconds = _positive_real(step_seconds, "step_seconds")
    if aggregation not in ("focal", "mean"):
        raise ValueError(
            f"aggregation must be 'mean' or 'focal', got {aggregation!r}"
        )
    validated_persistence = _positive_int(persistence, "persistence")
    validated_recency_top = _finite_real(recency_top, "recency_top")
    if validated_recency_top < 1.0:
        raise ValueError("recency_top must be a finite number at least one")
    if isinstance(r2_gate, (bool, np.bool_)) or not isinstance(r2_gate, Real):
        raise ValueError("r2_gate must be a finite number in [0, 1]")
    validated_r2_gate = float(r2_gate)
    if not np.isfinite(validated_r2_gate) or not 0.0 <= validated_r2_gate <= 1.0:
        raise ValueError("r2_gate must be a finite number in [0, 1]")
    self._rate: float = validated_rate
    self._threshold: float = validated_threshold
    self._window: int = int(round(validated_window_seconds * validated_rate))
    if self._window < 2:
        raise ValueError("window_seconds is too short for the sampling rate")
    self._step: int = max(1, int(round(validated_step_seconds * validated_rate)))
    self._aggregation: str = aggregation
    self._recency_top: float = validated_recency_top
    self._persistence: int = validated_persistence
    self._r2_gate: float = validated_r2_gate
    self._buffer: list[FloatArray] = []
    self._index: int = 0
    self._since_score: int = 0
    self._above: int = 0
    self._latched: bool = False
    self._latest_score: float = float("nan")

Attributes

latest_score property

latest_score: float

The most recent growth rate σ, or NaN before the first score.

rate property

rate: float

The stream sampling rate in Hz the monitor was constructed with.

threshold property

threshold: float

The certified matched-false-alarm threshold σ must reach to alarm.

aggregation property

aggregation: str

The certified aggregation scored under ("focal" or "mean").

r2_gate property

r2_gate: float

The fit-quality gate on the live rate; 0.0 means the gate is off.

window_seconds property

window_seconds: float

The effective window length in seconds (the rounded operating point).

step_seconds property

step_seconds: float

The effective re-scoring hop in seconds (the rounded operating point).

persistence property

persistence: int

Consecutive above-threshold windows required before an alarm fires.

recency_top property

recency_top: float

The recency weighting the growth rate is fitted under.

Methods:

from_evidence classmethod

from_evidence(
    evidence_path: str | Path,
    *,
    rate: float,
    **kwargs: object,
) -> GridModalStreamMonitor

Build a monitor from a sealed head-to-head artefact.

Reads the aggregation, recency weighting, and matched-false-alarm threshold from the certified modal record, so the live monitor carries exactly the certified operating point with no hand-set constants.

Parameters

evidence_path : str or Path Path to a sealed grid_modal_head_to_head.json artefact. rate : float The live stream's sampling rate in Hz. **kwargs : object Extra monitor arguments (window_seconds, step_seconds, persistence).

Returns

GridModalStreamMonitor A monitor at the certified operating point.

Source code in src/scpn_phase_orchestrator/monitor/grid_modal_stream.py

@classmethod
def from_evidence(
    cls, evidence_path: str | Path, *, rate: float, **kwargs: object
) -> GridModalStreamMonitor:
    """Build a monitor from a sealed head-to-head artefact.

    Reads the aggregation, recency weighting, and matched-false-alarm threshold from
    the certified ``modal`` record, so the live monitor carries exactly the
    certified operating point with no hand-set constants.

    Parameters
    ----------
    evidence_path : str or Path
        Path to a sealed ``grid_modal_head_to_head.json`` artefact.
    rate : float
        The live stream's sampling rate in Hz.
    **kwargs : object
        Extra monitor arguments (``window_seconds``, ``step_seconds``,
        ``persistence``).

    Returns
    -------
    GridModalStreamMonitor
        A monitor at the certified operating point.
    """
    payload = json.loads(Path(evidence_path).read_text(encoding="utf-8"))
    modal = payload["modal"]
    return cls(
        rate=rate,
        threshold=float(modal["score_threshold"]),
        aggregation=str(modal["aggregation"]),
        recency_top=float(modal["recency_top"]),
        **kwargs,  # type: ignore[arg-type]  # keys checked by __init__ signature
    )

from_stream_evidence classmethod

from_stream_evidence(
    evidence_path: str | Path,
    *,
    rate: float,
    recency_top: float = DEFAULT_RECENCY_TOP,
    gate_r2: float = _CERTIFIED_STREAM_R2_GATE,
    target_false_alarm: float = 0.1,
) -> GridModalStreamMonitor

Build a monitor at the certified streaming operating point.

Reads a sealed streaming operating-point artefact, selects its winning configuration (:func:_select_stream_operating_point — the development-best that holds the target false alarm out of sample, exactly the sealed verdict's choice), and configures the monitor with that window, step, persistence, and threshold, turning the fit-quality gate on iff the winner is the "r2gate" feature. This deploys the honest streaming winner — not the more permissive per-window operating point of :meth:from_evidence — with no hand-set thresholds.

Parameters

evidence_path : str or Path Path to a sealed grid_modal_stream_operating_point.json artefact. rate : float The live stream's sampling rate in Hz. recency_top : float The recency weighting the search was run at; defaults to the certified :data:~scpn_phase_orchestrator.monitor.grid_modal_growth.DEFAULT_RECENCY_TOP. The artefact does not carry it, so it is set here for transparency. gate_r2 : float The fit-quality gate level the search's "r2gate" feature used; defaults to the certified value. Only applied when the winner is a gated configuration. target_false_alarm : float The matched stream false-alarm target used to select the winner; must match the artefact's target_stream_false_alarm.

Returns

GridModalStreamMonitor A monitor at the certified streaming operating point.

Source code in src/scpn_phase_orchestrator/monitor/grid_modal_stream.py

@classmethod
def from_stream_evidence(
    cls,
    evidence_path: str | Path,
    *,
    rate: float,
    recency_top: float = DEFAULT_RECENCY_TOP,
    gate_r2: float = _CERTIFIED_STREAM_R2_GATE,
    target_false_alarm: float = 0.10,
) -> GridModalStreamMonitor:
    """Build a monitor at the certified *streaming* operating point.

    Reads a sealed streaming operating-point artefact, selects its winning
    configuration (:func:`_select_stream_operating_point` — the development-best
    that holds the target false alarm out of sample, exactly the sealed verdict's
    choice), and configures the monitor with that window, step, persistence, and
    threshold, turning the fit-quality gate on iff the winner is the ``"r2gate"``
    feature. This deploys the honest streaming winner — not the more permissive
    per-window operating point of :meth:`from_evidence` — with no hand-set
    thresholds.

    Parameters
    ----------
    evidence_path : str or Path
        Path to a sealed ``grid_modal_stream_operating_point.json`` artefact.
    rate : float
        The live stream's sampling rate in Hz.
    recency_top : float
        The recency weighting the search was run at; defaults to the certified
        :data:`~scpn_phase_orchestrator.monitor.grid_modal_growth.DEFAULT_RECENCY_TOP`.
        The artefact does not carry it, so it is set here for transparency.
    gate_r2 : float
        The fit-quality gate level the search's ``"r2gate"`` feature used; defaults
        to the certified value. Only applied when the winner is a gated
        configuration.
    target_false_alarm : float
        The matched stream false-alarm target used to select the winner; must match
        the artefact's ``target_stream_false_alarm``.

    Returns
    -------
    GridModalStreamMonitor
        A monitor at the certified streaming operating point.
    """
    payload = json.loads(Path(evidence_path).read_text(encoding="utf-8"))
    rows = cast("list[dict[str, object]]", payload["search"])
    winner = _select_stream_operating_point(
        rows, target_false_alarm=target_false_alarm
    )
    gated = str(winner["feature"]) == "r2gate"
    return cls(
        rate=rate,
        threshold=float(cast("float", winner["threshold"])),
        window_seconds=float(cast("float", winner["window_seconds"])),
        step_seconds=float(cast("float", winner["step_seconds"])),
        aggregation="focal",
        recency_top=recency_top,
        persistence=int(cast("int", winner["persistence"])),
        r2_gate=gate_r2 if gated else 0.0,
    )

reset

reset() -> None

Clear the window and alarm state, as if freshly constructed.

Source code in src/scpn_phase_orchestrator/monitor/grid_modal_stream.py

def reset(self) -> None:
    """Clear the window and alarm state, as if freshly constructed."""
    self._buffer.clear()
    self._index = 0
    self._since_score = 0
    self._above = 0
    self._latched = False
    self._latest_score = float("nan")

update

update(sample: FloatArray) -> StreamAlarm | None

Push one per-bus voltage sample; return an alarm on a threshold crossing.

Parameters

sample : FloatArray The bus-voltage magnitudes at one time step, shape (buses,).

Returns

StreamAlarm or None A :class:StreamAlarm on the sample that fires a fresh lead event, else None (still warming up, between re-scorings, below threshold, or already latched within the same episode).

Raises

ValueError If sample is not a non-empty finite real one-dimensional bus vector, contains coercive aliases, or changes the established bus count.

Source code in src/scpn_phase_orchestrator/monitor/grid_modal_stream.py

def update(self, sample: FloatArray) -> StreamAlarm | None:
    """Push one per-bus voltage sample; return an alarm on a threshold crossing.

    Parameters
    ----------
    sample : FloatArray
        The bus-voltage magnitudes at one time step, shape ``(buses,)``.

    Returns
    -------
    StreamAlarm or None
        A :class:`StreamAlarm` on the sample that fires a fresh lead event, else
        ``None`` (still warming up, between re-scorings, below threshold, or already
        latched within the same episode).

    Raises
    ------
    ValueError
        If ``sample`` is not a non-empty finite real one-dimensional bus vector,
        contains coercive aliases, or changes the established bus count.
    """
    values = _validate_sample(sample)
    if self._buffer and values.shape != self._buffer[0].shape:
        raise ValueError("sample bus count must remain constant")
    self._index += 1
    self._buffer.append(values)
    if len(self._buffer) > self._window:
        self._buffer.pop(0)
    self._since_score += 1
    if len(self._buffer) < self._window or self._since_score < self._step:
        return None
    self._since_score = 0
    score, bus = self._score_window()
    self._latest_score = score
    if score < self._threshold:
        self._above = 0
        self._latched = False
        return None
    self._above += 1
    if self._latched or self._above < self._persistence:
        return None
    self._latched = True
    return StreamAlarm(
        sample_index=self._index,
        time_s=self._index / self._rate,
        score=score,
        threshold=self._threshold,
        bus=bus,
    )

Functions:

Modal Sentinel

ModalSentinel is the bridge-agnostic live wiring for the stream monitor: any runtime bridge that yields one mapping of channel name to a real reading per frame plugs in, the operating point is read only from a sealed evidence artefact (verified before any value is trusted; a tampered artefact is rejected), and every alarm is sealed into a hash-addressed, review-only record carrying the operating-point provenance. Frames are fail-closed: a missing channel, an unknown channel, or a non-finite reading rejects the frame explicitly rather than silently degrading the monitored vector.

modal_sentinel

Bridge-agnostic wiring from live channel observations to sealed alarms.

The sentinel contract: any runtime bridge that yields one Mapping of channel name to a real reading per frame — the MQTT and OPC-UA tag bridges, the C37118 synchrophasor bridge, or a replayed capture — plugs into :class:ModalSentinel, which assembles the fixed channel vector, drives the certified :class:~scpn_phase_orchestrator.monitor.grid_modal_stream.GridModalStreamMonitor, and seals every alarm into a hash-addressed record carrying the operating-point provenance. The operating point is read ONLY from a sealed evidence artefact, verified before any value is trusted; the sentinel is review-only — it observes, records, and never actuates.

Fail-closed observation contract: a frame must carry exactly the declared channels — a missing channel, an unknown channel, or a non-finite reading rejects the frame with an explicit error rather than silently degrading the monitored vector.

Classes

ModalSentinel dataclass

ModalSentinel(
    monitor: GridModalStreamMonitor,
    channels: tuple[str, ...],
    provenance: dict[str, object] = dict(),
)

Review-only live sentinel: channel observations in, sealed alarms out.

Attributes

monitor : GridModalStreamMonitor The causal stream monitor carrying the certified operating point. channels : tuple[str, ...] The declared channel names, in the fixed vector order every frame must satisfy. provenance : dict[str, object] Operating-point provenance copied into every sealed alarm record. non_actuating : bool Always True — the sentinel observes and never drives hardware. execution_disabled : bool Always True — no control action is emitted from this sentinel.

Methods:

from_sealed_evidence classmethod

from_sealed_evidence(
    evidence_path: str | Path,
    *,
    case_id: str,
    rate: float,
    channels: tuple[str, ...],
    persistence: int = 1,
) -> ModalSentinel

Build a sentinel whose operating point comes only from sealed evidence.

The threshold comes from the sealed local calibration, the aggregation and recency weighting from the sealed detector block, and the window from the named case's sealed configuration; the step is a quarter window, as evaluated. The evidence content hash and the calibration's disclosed limits are carried into every alarm's provenance.

Parameters

evidence_path : str | Path Path to a sealed cross-dataset evidence artefact. case_id : str The sealed corpus case whose window configuration to carry. rate : float The live stream's sampling rate in hertz. channels : tuple[str, ...] Declared channel names in fixed vector order. persistence : int Consecutive above-threshold re-scorings before an alarm fires.

Returns

ModalSentinel A sentinel at the sealed operating point.

Raises

ValueError If the seal fails to verify, the payload is not a cross-dataset evidence record, the case is not in the sealed corpus, or its window is not numeric.

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

@classmethod
def from_sealed_evidence(
    cls,
    evidence_path: str | Path,
    *,
    case_id: str,
    rate: float,
    channels: tuple[str, ...],
    persistence: int = 1,
) -> ModalSentinel:
    """Build a sentinel whose operating point comes only from sealed evidence.

    The threshold comes from the sealed local calibration, the aggregation
    and recency weighting from the sealed detector block, and the window
    from the named case's sealed configuration; the step is a quarter
    window, as evaluated. The evidence content hash and the calibration's
    disclosed limits are carried into every alarm's provenance.

    Parameters
    ----------
    evidence_path : str | Path
        Path to a sealed cross-dataset evidence artefact.
    case_id : str
        The sealed corpus case whose window configuration to carry.
    rate : float
        The live stream's sampling rate in hertz.
    channels : tuple[str, ...]
        Declared channel names in fixed vector order.
    persistence : int
        Consecutive above-threshold re-scorings before an alarm fires.

    Returns
    -------
    ModalSentinel
        A sentinel at the sealed operating point.

    Raises
    ------
    ValueError
        If the seal fails to verify, the payload is not a cross-dataset
        evidence record,
        the case is not in the sealed corpus, or its window is not
        numeric.
    """
    payload = load_verified_evidence(evidence_path)
    corpus = payload.get("corpus")
    if not isinstance(corpus, dict) or "transitions" not in corpus:
        raise ValueError("evidence carries no corpus.transitions block")
    entry: dict[str, object] | None = None
    for candidate in corpus["transitions"]:
        if candidate.get("case") == case_id:
            entry = dict(candidate)
            break
    if entry is None:
        raise ValueError(f"case {case_id!r} is not in the sealed corpus")
    detector = payload.get("detector")
    calibration = payload.get("local_calibration")
    if not isinstance(detector, dict) or not isinstance(calibration, dict):
        raise ValueError("evidence carries no detector/local_calibration blocks")
    window_value = entry.get("window_seconds")
    if isinstance(window_value, bool) or not isinstance(window_value, (int, float)):
        raise ValueError("sealed case entry carries no numeric window_seconds")
    window_seconds = float(window_value)
    monitor = GridModalStreamMonitor(
        rate=rate,
        threshold=float(calibration["threshold"]),
        window_seconds=window_seconds,
        step_seconds=window_seconds / 4.0,
        aggregation=str(detector["aggregation"]),
        recency_top=float(detector["recency_top"]),
        persistence=persistence,
    )
    provenance: dict[str, object] = {
        "evidence_content_hash": payload["content_hash"],
        "case": case_id,
        "threshold": calibration["threshold"],
        "calibration_n_null": calibration.get("n_null"),
    }
    return cls(monitor=monitor, channels=channels, provenance=provenance)

observe

observe(
    values: Mapping[str, float],
) -> dict[str, object] | None

Consume one frame of channel readings; return a sealed alarm record.

Parameters

values : Mapping[str, float] One reading per declared channel, keyed by channel name. The frame must carry exactly the declared channels.

Returns

dict[str, object] | None A sealed, hash-addressed alarm record when the monitor raises a fresh alarm on this frame, else None.

Raises

ValueError If the frame misses a declared channel, carries an unknown channel, or any reading is boolean or not a finite real number.

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

def observe(self, values: Mapping[str, float]) -> dict[str, object] | None:
    """Consume one frame of channel readings; return a sealed alarm record.

    Parameters
    ----------
    values : Mapping[str, float]
        One reading per declared channel, keyed by channel name. The frame
        must carry exactly the declared channels.

    Returns
    -------
    dict[str, object] | None
        A sealed, hash-addressed alarm record when the monitor raises a
        fresh alarm on this frame, else ``None``.

    Raises
    ------
    ValueError
        If the frame misses a declared channel, carries an unknown
        channel, or any reading is boolean or not a finite real number.
    """
    unknown = set(values) - set(self.channels)
    if unknown:
        raise ValueError(f"unknown channels in frame: {sorted(unknown)}")
    vector = np.empty(len(self.channels), dtype=np.float64)
    for index, name in enumerate(self.channels):
        if name not in values:
            raise ValueError(f"frame misses declared channel {name!r}")
        reading = values[name]
        if isinstance(reading, bool) or not isinstance(reading, (int, float)):
            raise ValueError(f"channel {name!r} reading must be a real number")
        value = float(reading)
        if not np.isfinite(value):
            raise ValueError(f"channel {name!r} reading must be finite")
        vector[index] = value
    alarm = self.monitor.update(vector)
    if alarm is None:
        return None
    focal_channel = (
        None if alarm.bus == WHOLE_NETWORK_BUS else self.channels[alarm.bus]
    )
    record: dict[str, object] = {
        "kind": "modal_sentinel_alarm",
        "provenance": dict(self.provenance),
        "channels": list(self.channels),
        "sample_index": alarm.sample_index,
        "time_s": alarm.time_s,
        "score": alarm.score,
        "threshold": alarm.threshold,
        "focal_channel": focal_channel,
        "review_only": True,
    }
    record["content_hash"] = canonical_record_hash(record)
    return record

Functions:

load_verified_evidence

load_verified_evidence(
    evidence_path: str | Path,
) -> dict[str, object]

Load a sealed evidence payload and verify its content hash, fail-closed.

Parameters

evidence_path : str | Path Path to a sealed JSON artefact carrying a content_hash field.

Returns

dict[str, object] The verified payload.

Raises

ValueError If the payload is not a JSON object, carries no content_hash, or the hash does not recompute from the record — a tampered artefact must never configure a live sentinel.

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

def load_verified_evidence(evidence_path: str | Path) -> dict[str, object]:
    """Load a sealed evidence payload and verify its content hash, fail-closed.

    Parameters
    ----------
    evidence_path : str | Path
        Path to a sealed JSON artefact carrying a ``content_hash`` field.

    Returns
    -------
    dict[str, object]
        The verified payload.

    Raises
    ------
    ValueError
        If the payload is not a JSON object, carries no ``content_hash``, or
        the hash does not recompute from the record — a tampered artefact
        must never configure a live sentinel.
    """
    payload = json.loads(Path(evidence_path).read_text(encoding="utf-8"))
    if not isinstance(payload, dict):
        raise ValueError("evidence must be a JSON object; refusing to trust it")
    record = copy.deepcopy(payload)
    sealed = record.pop("content_hash", None)
    if not isinstance(sealed, str):
        raise ValueError("evidence carries no content_hash; refusing to trust it")
    if canonical_record_hash(record) != sealed:
        raise ValueError(
            "evidence content_hash does not recompute from the record; "
            "refusing to configure a live sentinel from a tampered artefact"
        )
    return payload