Audit¶

SHA256-chained audit logging, protobuf event streaming, and deterministic replay for regulatory compliance, debugging, and formal verification. Every supervisor decision, regime transition, and actuation command can be recorded with tamper-evident JSONL records and a parallel event-sourced protobuf stream.

Motivation¶

SPO is designed for safety-critical applications (power grids, plasma control, medical devices). These domains require:

Traceability — every control decision must be attributable to a specific input state and policy rule
Tamper evidence — audit records must detect insertion, deletion, or modification after the fact
Reproducibility — given the same inputs and code version, the system must produce identical outputs

The audit subsystem provides all three via hash-chained logging, event-sourced streaming, and deterministic replay.

Hash Chain Structure¶

Each audit record contains:

Field	Description
`step`	Monotonic step counter
`timestamp`	ISO 8601 UTC timestamp
`event_type`	`regime_transition`, `actuation`, `boundary_breach`, etc.
`payload`	Event-specific data (JSON-serialisable)
`prev_hash`	SHA256 of the previous record
`hash`	SHA256 of this record (step + event_type + payload + prev_hash)

The first record uses prev_hash = "0" * 64. To verify the chain, recompute each hash and check it matches the stored value and the next record's prev_hash.

The spo run --audit header includes the resolved binding summary under binding_config and binding_summary. For N-channel domainpacks this summary includes channel_algebra, covering required and optional channels, derived channels, runtime evidence channels, group membership, coupling participants, and missing required channel evidence.

Keyed Audit Signatures¶

Set SPO_AUDIT_KEY to enable HMAC-SHA256 signatures on JSONL audit records. Audit records include canonical payload metadata for replay verification, with HMAC fields added when signing is enabled:

Field	Description
`_audit_mode`	`hmac-signed` or `unsigned-development`
`_audit_schema_version`	Signature metadata schema version
`_audit_stream_id`	Logical JSONL stream id
`_audit_sequence`	Monotonic record sequence
`_audit_timestamp_unix_ns`	Signing timestamp in Unix nanoseconds
`_previous_hash`	Previous JSONL record hash used by the signature
`_payload_hash`	SHA256 of the canonical payload without audit metadata
`_signature`	HMAC algorithm, key id, and signature value

The raw key is never written to the audit file. The stored key id is sha256(key)[:16], which lets replay choose the right verification key without logging secret material.

When SPO_AUDIT_KEY is configured, ReplayEngine.verify_integrity() and spo replay --verify fail closed if a record is unsigned, malformed, signed by an unknown key, or modified after signing. Without SPO_AUDIT_KEY, legacy unsigned development logs remain readable and hash-chain verification keeps its previous behaviour.

The protobuf event stream uses the same environment policy. When event_stream is enabled on AuditLogger, every envelope records its audit mode, signature algorithm, key id, and HMAC value alongside the existing sequence, payload hash, previous hash, and event hash. verify_event_stream_integrity() and spo watch reject unsigned or signature-invalid envelopes whenever SPO_AUDIT_KEY or SPO_AUDIT_KEYRING is configured.

When simulate() receives an AuditLogger with a protobuf event stream, it flushes the stream after the final event and stores the whole-stream integrity summary on SimulationResult.audit_event_stream_integrity. This is a run-end integrity check, not a per-step action gate; append-time hash chaining remains the live tamper-evidence during the run.

Unsigned logs and streams are allowed only when no audit key is configured. They are marked explicitly as unsigned-development and still carry canonical payload hashes so reviewers can distinguish local development traces from operational signed evidence without losing deterministic payload evidence.

For key rotation, keep historical keys only in the operator environment and pass them as a JSON object through SPO_AUDIT_KEYRING:

export SPO_AUDIT_KEY="$(openssl rand -hex 32)"
export SPO_AUDIT_KEYRING='{
  "<sha256-old-secret-prefix>": "<old-generated-secret>",
  "<sha256-new-secret-prefix>": "<new-generated-secret>"
}'
spo replay audit.jsonl --verify

Each keyring object key must match sha256(secret)[:16]; mismatches fail closed. Do not commit these environment values, include them in diagnostics, or store them in audit artefacts.

Audit Logger¶

Appends timestamped, SHA256-chained records to a JSONL audit trail. When event_stream is supplied, the same stored records are also appended to a length-delimited protobuf stream.

The compatibility facade scpn_phase_orchestrator.audit exposes the audit logger, replay engine, and event-stream helpers lazily. Its __all__ and dir() output list the same public exports before import-time resolution, so interactive tools and documentation generators see the facade contract without eagerly importing the runtime audit stack.

from scpn_phase_orchestrator.runtime.audit_logger import AuditLogger

logger = AuditLogger("audit.jsonl", event_stream="audit.spoa")
logger.log_event("regime_transition", {"from": "nominal", "to": "degraded"})
integrity = logger.verify_event_stream_integrity()
assert integrity is not None and integrity.ok
logger.close()

audit_logger ¶

Append-only JSONL logger for replayable SPO runs.

AuditLogger records headers, simulation steps, supervisor actions, and named events with a SHA-256 hash chain. When SPO_AUDIT_KEY is configured, existing unsigned streams are rejected and new records carry HMAC metadata so downstream replay and reporting can verify provenance before trusting an audit trail.

Classes¶

AuditStreamIntegrityResult `dataclass` ¶

AuditStreamIntegrityResult(
    event_stream_path: str, ok: bool, verified_events: int
)

Close-time integrity summary for a protobuf audit event stream.

Parameters¶

event_stream_path : str Filesystem path to the verified protobuf audit event stream. ok : bool Whether payload digests, sequence numbers, hash links, and required signatures verified. verified_events : int Number of consecutive events verified before the first failure, or all events when ok is True.

Methods:¶

to_audit_record ¶

to_audit_record() -> dict[str, object]

Return a JSON-safe audit-stream integrity record.

Returns¶

dict[str, object] A JSON-safe integrity summary for downstream reports.

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

def to_audit_record(self) -> dict[str, object]:
    """Return a JSON-safe audit-stream integrity record.

    Returns
    -------
    dict[str, object]
        A JSON-safe integrity summary for downstream reports.
    """
    return {
        "event_stream_path": self.event_stream_path,
        "ok": self.ok,
        "verified_events": self.verified_events,
    }

AuditLogger ¶

AuditLogger(
    path: str | Path,
    *,
    event_stream: str | Path | None = None,
)

Append-only JSONL audit log for UPDE simulation steps.

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

def __init__(self, path: str | Path, *, event_stream: str | Path | None = None):
    if not isinstance(path, (str, Path)):
        raise AuditError(f"audit path must be str or Path, got {path!r}")
    if event_stream is not None and not isinstance(event_stream, (str, Path)):
        raise AuditError(
            f"event_stream path must be str, Path, or None, got {event_stream!r}"
        )
    if not str(path).strip():
        raise AuditError("audit path must be a non-empty path")
    if event_stream is not None and not str(event_stream).strip():
        raise AuditError("event_stream path must be non-empty when provided")
    self._path = Path(path)
    if self._path.exists() and self._path.is_dir():
        raise AuditError(
            f"audit path must be a file path, got directory {self._path}"
        )
    self._prev_hash, self._sequence = self._load_previous_state()
    self._audit_key = os.environ.get("SPO_AUDIT_KEY")
    self._stream_id = _DEFAULT_STREAM_ID
    if self._audit_key is not None and self._audit_key == "":
        msg = "SPO_AUDIT_KEY must not be empty"
        raise AuditError(msg)
    if self._audit_key is None and is_production_mode("SPO_AUDIT"):
        raise AuditError(
            "SPO_AUDIT_KEY is required in production mode "
            "(SPO_AUDIT_ENV/SPO_AUDIT_PROFILE or SPO_ENV/SPO_PROFILE = "
            "'production'): refusing to write an unsigned, unverifiable audit "
            "trail. Set SPO_AUDIT_KEY to enable HMAC signing."
        )
    if self._audit_key is not None and self._sequence > 0:
        self._ensure_existing_stream_is_signed()
    self._fh = self._path.open("a", encoding="utf-8", buffering=1)
    self._event_stream = (
        EventStreamWriter(event_stream) if event_stream is not None else None
    )
    self._event_stream_integrity: AuditStreamIntegrityResult | None = None

Attributes¶

event_stream_integrity `property` ¶

event_stream_integrity: AuditStreamIntegrityResult | None

Return the most recent event-stream integrity summary, if any.

Returns¶

AuditStreamIntegrityResult | None The last computed event-stream integrity result, or None when no protobuf stream has been verified.

Methods:¶

log_header ¶

log_header(
    *,
    n_oscillators: int,
    dt: float,
    method: str = "euler",
    seed: int | None = None,
    amplitude_mode: bool = False,
    control_mode: str = "supervisor_policy",
    binding_config: dict[str, object] | None = None,
    binding_summary: dict[str, object] | None = None,
) -> None

Engine configuration record for replay reconstruction.

Parameters¶

n_oscillators : int Number of oscillators in the system. dt : float Integration step size. method : str Integration method (euler, rk4, or rk45). seed : int | None Seed for the deterministic RNG, or None. amplitude_mode : bool Whether the engine runs in Stuart-Landau amplitude mode. control_mode : str Live control surface used by the simulation core. binding_config : dict[str, object] | None Resolved binding configuration, or None. binding_summary : dict[str, object] | None Resolved binding summary, or None.

Raises¶

AuditError If the audit log cannot be written.

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

def log_header(
    self,
    *,
    n_oscillators: int,
    dt: float,
    method: str = "euler",
    seed: int | None = None,
    amplitude_mode: bool = False,
    control_mode: str = "supervisor_policy",
    binding_config: dict[str, object] | None = None,
    binding_summary: dict[str, object] | None = None,
) -> None:
    """Engine configuration record for replay reconstruction.

    Parameters
    ----------
    n_oscillators : int
        Number of oscillators in the system.
    dt : float
        Integration step size.
    method : str
        Integration method (``euler``, ``rk4``, or ``rk45``).
    seed : int | None
        Seed for the deterministic RNG, or ``None``.
    amplitude_mode : bool
        Whether the engine runs in Stuart-Landau amplitude mode.
    control_mode : str
        Live control surface used by the simulation core.
    binding_config : dict[str, object] | None
        Resolved binding configuration, or ``None``.
    binding_summary : dict[str, object] | None
        Resolved binding summary, or ``None``.

    Raises
    ------
    AuditError
        If the audit log cannot be written.
    """
    if isinstance(n_oscillators, bool) or not isinstance(n_oscillators, int):
        raise AuditError(
            f"n_oscillators must be a positive integer, got {n_oscillators!r}"
        )
    if n_oscillators <= 0:
        raise AuditError(
            f"n_oscillators must be a positive integer, got {n_oscillators!r}"
        )
    if (
        isinstance(dt, bool)
        or not isinstance(dt, (int, float))
        or not isfinite(float(dt))
    ):
        raise AuditError(f"dt must be a finite positive real, got {dt!r}")
    if float(dt) <= 0.0:
        raise AuditError(f"dt must be a finite positive real, got {dt!r}")
    if not isinstance(method, str) or not method.strip():
        raise AuditError(f"method must be a non-empty string, got {method!r}")
    if seed is not None and (isinstance(seed, bool) or not isinstance(seed, int)):
        raise AuditError(f"seed must be integer or None, got {seed!r}")
    if isinstance(amplitude_mode, bool) is False:
        raise AuditError(f"amplitude_mode must be bool, got {amplitude_mode!r}")
    if not isinstance(control_mode, str) or not control_mode.strip():
        raise AuditError(
            f"control_mode must be a non-empty string, got {control_mode!r}"
        )
    if binding_config is not None and not isinstance(binding_config, dict):
        raise AuditError(
            "binding_config must be dict[str, object] or None, "
            f"got {binding_config!r}"
        )
    if binding_summary is not None and not isinstance(binding_summary, dict):
        raise AuditError(
            "binding_summary must be dict[str, object] or None, "
            f"got {binding_summary!r}"
        )
    record: dict[str, Any] = {
        "header": True,
        "n_oscillators": n_oscillators,
        "dt": dt,
        "method": method,
    }
    if seed is not None:
        record["seed"] = seed
    if amplitude_mode:
        record["amplitude_mode"] = True
    record["control_mode"] = control_mode

    if binding_summary is None:
        binding_summary = binding_config
    if binding_summary is not None:
        record["binding_summary"] = binding_summary
    if binding_config is not None:
        record["binding_config"] = binding_config
    self._write_record(record)

log_step ¶

log_step(
    step: int,
    upde_state: UPDEState,
    actions: list[ControlAction],
    *,
    phases: FloatArray | None = None,
    omegas: FloatArray | None = None,
    knm: FloatArray | None = None,
    alpha: FloatArray | None = None,
    zeta: float = 0.0,
    psi_drive: float = 0.0,
    amplitudes: FloatArray | None = None,
    mu: FloatArray | None = None,
    knm_r: FloatArray | None = None,
    epsilon: float | None = None,
    channel_runtime: dict[str, object] | None = None,
) -> None

Write one simulation step to the audit log with optional full state.

Parameters¶

step : int Zero-based simulation step index. upde_state : UPDEState The UPDE state to record or export. actions : list[ControlAction] The control actions recorded for the step. phases : FloatArray | None Oscillator phases in radians, shape (N,). omegas : FloatArray | None Natural frequencies in rad/s, shape (N,). knm : FloatArray | None Coupling matrix K_nm, shape (N, N). alpha : FloatArray | None Phase-lag matrix in radians, shape (N, N), or None for no lag. zeta : float External drive strength ζ. psi_drive : float External drive reference phase in radians. amplitudes : FloatArray | None Oscillator amplitudes, shape (N,), or None. mu : FloatArray | None Per-oscillator linear growth parameters, or None. knm_r : FloatArray | None Amplitude coupling matrix, shape (N, N), or None. epsilon : float | None Stuart-Landau amplitude coupling factor, or None. channel_runtime : dict[str, object] | None N-channel runtime evidence, or None.

Raises¶

AuditError If the audit log cannot be written.

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

def log_step(
    self,
    step: int,
    upde_state: UPDEState,
    actions: list[ControlAction],
    *,
    phases: FloatArray | None = None,
    omegas: FloatArray | None = None,
    knm: FloatArray | None = None,
    alpha: FloatArray | None = None,
    zeta: float = 0.0,
    psi_drive: float = 0.0,
    amplitudes: FloatArray | None = None,
    mu: FloatArray | None = None,
    knm_r: FloatArray | None = None,
    epsilon: float | None = None,
    channel_runtime: dict[str, object] | None = None,
) -> None:
    """Write one simulation step to the audit log with optional full state.

    Parameters
    ----------
    step : int
        Zero-based simulation step index.
    upde_state : UPDEState
        The UPDE state to record or export.
    actions : list[ControlAction]
        The control actions recorded for the step.
    phases : FloatArray | None
        Oscillator phases in radians, shape ``(N,)``.
    omegas : FloatArray | None
        Natural frequencies in rad/s, shape ``(N,)``.
    knm : FloatArray | None
        Coupling matrix ``K_nm``, shape ``(N, N)``.
    alpha : FloatArray | None
        Phase-lag matrix in radians, shape ``(N, N)``, or ``None`` for no lag.
    zeta : float
        External drive strength ``ζ``.
    psi_drive : float
        External drive reference phase in radians.
    amplitudes : FloatArray | None
        Oscillator amplitudes, shape ``(N,)``, or ``None``.
    mu : FloatArray | None
        Per-oscillator linear growth parameters, or ``None``.
    knm_r : FloatArray | None
        Amplitude coupling matrix, shape ``(N, N)``, or ``None``.
    epsilon : float | None
        Stuart-Landau amplitude coupling factor, or ``None``.
    channel_runtime : dict[str, object] | None
        N-channel runtime evidence, or ``None``.

    Raises
    ------
    AuditError
        If the audit log cannot be written.
    """
    if isinstance(step, bool) or not isinstance(step, int):
        raise AuditError(f"step must be a non-negative integer, got {step!r}")
    if step < 0:
        raise AuditError(f"step must be a non-negative integer, got {step!r}")
    if not isinstance(upde_state, UPDEState):
        raise AuditError(f"upde_state must be UPDEState, got {upde_state!r}")
    if not isinstance(actions, list):
        raise AuditError(f"actions must be list[ControlAction], got {actions!r}")
    for idx, action in enumerate(actions):
        if not isinstance(action, ControlAction):
            raise AuditError(
                f"actions[{idx}] must be ControlAction, got {action!r}"
            )
    record = {
        "ts": time.time(),
        "step": step,
        "regime": upde_state.regime_id,
        "stability": upde_state.stability_proxy,
        "layers": [{"R": ls.R, "psi": ls.psi} for ls in upde_state.layers],
        "actions": [
            {
                "knob": a.knob,
                "scope": a.scope,
                "value": a.value,
                "ttl_s": a.ttl_s,
                "justification": a.justification,
            }
            for a in actions
        ],
    }
    if phases is not None:
        if omegas is None or knm is None or alpha is None:
            msg = "omegas, knm, alpha required when phases is provided"
            raise AuditError(msg)
        for name, arr in (
            ("phases", phases),
            ("omegas", omegas),
            ("knm", knm),
            ("alpha", alpha),
        ):
            if not np.isfinite(arr).all():
                raise AuditError(f"{name} must contain only finite values")
        record["phases"] = phases.tolist()
        record["omegas"] = omegas.tolist()
        record["knm"] = knm.tolist()
        record["alpha"] = alpha.tolist()
        record["zeta"] = zeta
        record["psi_drive"] = psi_drive
    if amplitudes is not None:
        record["amplitudes"] = amplitudes.tolist()
    if mu is not None:
        record["mu"] = mu.tolist()
    if knm_r is not None:
        record["knm_r"] = knm_r.tolist()
    if epsilon is not None:
        if isinstance(epsilon, bool) or not isinstance(epsilon, (int, float)):
            raise AuditError(f"epsilon must be finite real, got {epsilon!r}")
        if not isfinite(float(epsilon)):
            raise AuditError(f"epsilon must be finite real, got {epsilon!r}")
        record["epsilon"] = epsilon
    if channel_runtime is not None:
        if not isinstance(channel_runtime, dict):
            raise AuditError(
                "channel_runtime must be dict[str, object] when provided"
            )
        record["channel_runtime"] = channel_runtime
    self._write_record(record)

log_event ¶

log_event(event_type: str, data: dict[str, Any]) -> None

Write a named event with arbitrary data to the audit log.

Parameters¶

event_type : str Named event type, or None. data : dict[str, Any] Arbitrary JSON-safe event payload.

Raises¶

AuditError If the audit log cannot be written.

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

def log_event(self, event_type: str, data: dict[str, Any]) -> None:
    """Write a named event with arbitrary data to the audit log.

    Parameters
    ----------
    event_type : str
        Named event type, or ``None``.
    data : dict[str, Any]
        Arbitrary JSON-safe event payload.

    Raises
    ------
    AuditError
        If the audit log cannot be written.
    """
    if not isinstance(event_type, str) or not event_type.strip():
        raise AuditError(
            f"event_type must be a non-empty string, got {event_type!r}"
        )
    if not isinstance(data, dict):
        raise AuditError(f"data must be dict[str, object], got {data!r}")
    record = {"ts": time.time(), "event": event_type, **data}
    self._write_record(record)

close ¶

close() -> None

Flush and close the audit log file handle.

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

def close(self) -> None:
    """Flush and close the audit log file handle."""
    self._fh.flush()
    if self._event_stream is not None:
        if self._event_stream_integrity is None:
            self.verify_event_stream_integrity()
        self._event_stream.close()
    self._fh.close()

verify_event_stream_integrity ¶

verify_event_stream_integrity() -> (
    AuditStreamIntegrityResult | None
)

Verify the configured protobuf event stream after flushing writes.

Returns¶

AuditStreamIntegrityResult | None The integrity summary when this logger owns an event stream, else None for JSONL-only audit logs.

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

def verify_event_stream_integrity(self) -> AuditStreamIntegrityResult | None:
    """Verify the configured protobuf event stream after flushing writes.

    Returns
    -------
    AuditStreamIntegrityResult | None
        The integrity summary when this logger owns an event stream, else
        ``None`` for JSONL-only audit logs.
    """
    if self._event_stream is None:
        return None
    self._fh.flush()
    self._event_stream.flush()
    path = self._event_stream.path
    events = read_event_stream(path)
    ok, verified_events = verify_event_stream_events(events)
    result = AuditStreamIntegrityResult(
        event_stream_path=str(path),
        ok=ok,
        verified_events=verified_events,
    )
    self._event_stream_integrity = result
    return result

Functions:¶

Audit Signing Helpers¶

scpn_phase_orchestrator.runtime.audit_signing centralises audit signature constants, key identifiers, and verification key loading. It is used by JSONL audit replay and protobuf event-stream verification so keyring validation stays consistent across both audit transports.

audit_signing ¶

HMAC key discovery helpers for signed audit verification.

The module derives stable non-secret key identifiers and loads current or historical verification keys from SPO_AUDIT_KEY and SPO_AUDIT_KEYRING. Empty, malformed, or mismatched key material raises ValueError so signed audit verification never falls back to an ambiguous trust state.

Functions:¶

key_id_for_secret ¶

key_id_for_secret(key_material: str) -> str

Return the audit key identifier stored in signed audit metadata.

Parameters¶

key_material : str The audit signing-key material.

Returns¶

str The audit key identifier stored in signed audit metadata.

Raises¶

ValueError If the inputs are invalid or inconsistent.

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

def key_id_for_secret(key_material: str) -> str:
    """Return the audit key identifier stored in signed audit metadata.

    Parameters
    ----------
    key_material : str
        The audit signing-key material.

    Returns
    -------
    str
        The audit key identifier stored in signed audit metadata.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if key_material == "":
        raise ValueError("audit signing key must not be empty")
    return hashlib.sha256(key_material.encode()).hexdigest()[:16]

audit_verification_keys ¶

audit_verification_keys() -> dict[str, str]

Load current and historical audit verification keys from the environment.

SPO_AUDIT_KEY supplies the current operational key. SPO_AUDIT_KEYRING supplies historical keys as a JSON object mapping sha256(secret)[:16] to the corresponding secret. Invalid or mismatched keyrings fail closed by raising ValueError.

Returns¶

dict[str, str] Load current and historical audit verification keys from the environment.

Raises¶

ValueError If the inputs are invalid or inconsistent.

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

def audit_verification_keys() -> dict[str, str]:
    """Load current and historical audit verification keys from the environment.

    `SPO_AUDIT_KEY` supplies the current operational key. `SPO_AUDIT_KEYRING`
    supplies historical keys as a JSON object mapping `sha256(secret)[:16]` to
    the corresponding secret. Invalid or mismatched keyrings fail closed by
    raising `ValueError`.

    Returns
    -------
    dict[str, str]
        Load current and historical audit verification keys from the environment.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    keys: dict[str, str] = {}
    current_key = os.environ.get("SPO_AUDIT_KEY")
    if current_key == "":
        raise ValueError("SPO_AUDIT_KEY must not be empty")
    if current_key is not None:
        keys[key_id_for_secret(current_key)] = current_key

    keyring_json = os.environ.get("SPO_AUDIT_KEYRING")
    if keyring_json is None:
        return keys
    if keyring_json == "":
        raise ValueError("SPO_AUDIT_KEYRING must not be empty")
    try:
        loaded = json.loads(keyring_json, parse_constant=_reject_json_constant)
    except json.JSONDecodeError as exc:
        raise ValueError("SPO_AUDIT_KEYRING must be JSON") from exc
    except ValueError as exc:
        raise ValueError("SPO_AUDIT_KEYRING must contain only finite JSON") from exc
    if not isinstance(loaded, dict):
        raise ValueError("SPO_AUDIT_KEYRING must be a JSON object")
    for key_id, key_value in loaded.items():
        if not isinstance(key_id, str) or key_id == "":
            raise ValueError("SPO_AUDIT_KEYRING key ids must be non-empty strings")
        if not isinstance(key_value, str) or key_value == "":
            raise ValueError("SPO_AUDIT_KEYRING keys must be non-empty strings")
        if key_id != key_id_for_secret(key_value):
            raise ValueError("SPO_AUDIT_KEYRING key id does not match key material")
        keys[key_id] = key_value
    return keys

Deterministic Replay¶

Replays an audit trail against a fresh SPO instance to reproduce the exact sequence of states. Replay verifies that the same inputs produce the same outputs — detecting non-determinism, floating-point platform differences, or code regressions.

Replay guarantees:

Same binding spec + same audit trail → identical phase trajectories
Any divergence is flagged with the step number and magnitude
Platform-specific float differences (x86 vs ARM extended precision) are handled via configurable tolerance
JSONL parsing rejects non-finite constants, duplicate object keys, and non-object lines before replay or hash verification

from scpn_phase_orchestrator.runtime.replay import ReplayEngine

replay = ReplayEngine("audit.jsonl")
entries = replay.load()
header = replay.load_header(entries)

if header is not None:
    engine = replay.build_engine(header)
    ok, verified = replay.verify_determinism_chained(engine, entries)
    print(f"verified={verified} ok={ok}")

replay ¶

Replay and integrity verification for SPO audit logs.

The replay engine reconstructs UPDE or Stuart-Landau state from audit JSONL, checks hash-chain and optional HMAC integrity, and reruns chained state transitions against logged next-step phases. Malformed headers, unsupported methods, invalid signatures, and non-replayable records fail closed instead of silently accepting unverifiable provenance.

Classes¶

ReplayEngine ¶

ReplayEngine(log_path: str | Path)

Replay and verify determinism of JSONL audit logs.

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

def __init__(self, log_path: str | Path):
    self._log_path = Path(log_path)

Methods:¶

load ¶

load() -> list[dict[str, Any]]

Read and parse all JSONL entries from the audit log file.

Returns¶

list[dict[str, Any]] Read and parse all JSONL entries from the audit log file.

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

def load(self) -> list[dict[str, Any]]:
    """Read and parse all JSONL entries from the audit log file.

    Returns
    -------
    list[dict[str, Any]]
        Read and parse all JSONL entries from the audit log file.
    """
    entries = []
    with self._log_path.open(encoding="utf-8") as fh:
        for line in fh:
            line = line.strip()
            if line:
                entries.append(_parse_audit_json(line))
    return entries

replay_step ¶

replay_step(step_data: dict[str, Any]) -> UPDEState

Reconstruct UPDEState from a log entry.

Parameters¶

step_data : dict[str, Any] A single audit-log entry to reconstruct.

Returns¶

UPDEState The reconstructed UPDE state.

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

def replay_step(self, step_data: dict[str, Any]) -> UPDEState:
    """Reconstruct UPDEState from a log entry.

    Parameters
    ----------
    step_data : dict[str, Any]
        A single audit-log entry to reconstruct.

    Returns
    -------
    UPDEState
        The reconstructed UPDE state.
    """
    layers = [
        LayerState(
            R=_numeric_value(ld.get("R", 0.0)),
            psi=_numeric_value(ld.get("psi", 0.0)),
        )
        for ld in _layer_records(step_data)
    ]
    return UPDEState(
        layers=layers,
        cross_layer_alignment=np.zeros((len(layers), len(layers))),
        stability_proxy=_numeric_value(step_data.get("stability", 0.0)),
        regime_id=step_data.get("regime", "unknown"),
    )

load_header ¶

load_header(
    entries: list[dict[str, Any]],
) -> dict[str, Any] | None

Extract the header record (engine config) if present.

Parameters¶

entries : list[dict[str, Any]] Audit-log entries to operate on.

Returns¶

dict[str, Any] | None The header record, or None if absent.

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

def load_header(self, entries: list[dict[str, Any]]) -> dict[str, Any] | None:
    """Extract the header record (engine config) if present.

    Parameters
    ----------
    entries : list[dict[str, Any]]
        Audit-log entries to operate on.

    Returns
    -------
    dict[str, Any] | None
        The header record, or ``None`` if absent.
    """
    for entry in entries:
        if entry.get("header"):
            return entry
    return None

step_entries ¶

step_entries(
    entries: list[dict[str, Any]],
) -> list[dict[str, Any]]

Filter to entries with full UPDE state (replayable).

Parameters¶

entries : list[dict[str, Any]] Audit-log entries to operate on.

Returns¶

list[dict[str, Any]] The replayable entries with full UPDE state.

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

def step_entries(self, entries: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Filter to entries with full UPDE state (replayable).

    Parameters
    ----------
    entries : list[dict[str, Any]]
        Audit-log entries to operate on.

    Returns
    -------
    list[dict[str, Any]]
        The replayable entries with full UPDE state.
    """
    return [e for e in entries if "phases" in e]

build_engine ¶

build_engine(
    header: dict[str, Any],
) -> UPDEEngine | StuartLandauEngine

Construct engine from header (UPDE or Stuart-Landau).

Parameters¶

header : dict[str, Any] The audit header record (engine config).

Returns¶

UPDEEngine | StuartLandauEngine The engine reconstructed from the header.

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

def build_engine(self, header: dict[str, Any]) -> UPDEEngine | StuartLandauEngine:
    """Construct engine from header (UPDE or Stuart-Landau).

    Parameters
    ----------
    header : dict[str, Any]
        The audit header record (engine config).

    Returns
    -------
    UPDEEngine | StuartLandauEngine
        The engine reconstructed from the header.
    """
    n_oscillators = _required_header_int(header, "n_oscillators")
    dt = _required_header_float(header, "dt")
    method = _header_method(header)
    if _header_amplitude_mode(header):
        return StuartLandauEngine(
            n_oscillators=n_oscillators,
            dt=dt,
            method=method,
        )
    return UPDEEngine(
        n_oscillators=n_oscillators,
        dt=dt,
        method=method,
    )

verify_determinism_chained ¶

verify_determinism_chained(
    engine: UPDEEngine,
    entries: list[dict[str, Any]],
    atol: float = 1e-06,
) -> tuple[bool, int]

Chained multi-step replay: output of step N must match input of step N+1.

Returns (passed, n_verified).

Parameters¶

engine : UPDEEngine The engine used to replay logged steps. entries : list[dict[str, Any]] Audit-log entries to operate on. atol : float Absolute comparison tolerance.

Returns¶

tuple[bool, int] A (passed, n_verified) pair.

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

def verify_determinism_chained(
    self,
    engine: UPDEEngine,
    entries: list[dict[str, Any]],
    atol: float = 1e-6,
) -> tuple[bool, int]:
    """Chained multi-step replay: output of step N must match input of step N+1.

    Returns (passed, n_verified).

    Parameters
    ----------
    engine : UPDEEngine
        The engine used to replay logged steps.
    entries : list[dict[str, Any]]
        Audit-log entries to operate on.
    atol : float
        Absolute comparison tolerance.

    Returns
    -------
    tuple[bool, int]
        A ``(passed, n_verified)`` pair.
    """
    replayable = self.step_entries(entries)
    if len(replayable) < 2:
        return True, 0

    verified = 0
    for i in range(len(replayable) - 1):
        curr = replayable[i]
        nxt = replayable[i + 1]
        if not _has_fields(curr, _UPDE_REPLAY_FIELDS) or "phases" not in nxt:
            return False, verified
        try:
            phases = np.asarray(curr["phases"])
            omegas = np.asarray(curr["omegas"])
            knm_arr = np.asarray(curr["knm"])
            alpha_arr = np.asarray(curr["alpha"])
            zeta = curr.get("zeta", 0.0)
            psi_drive = curr.get("psi_drive", 0.0)

            computed = engine.step(
                phases, omegas, knm_arr, zeta, psi_drive, alpha_arr
            )
            logged_next = np.asarray(nxt["phases"])
        except (TypeError, ValueError):
            return False, verified

        if not np.allclose(computed, logged_next, atol=atol):
            return False, verified
        verified += 1

    return True, verified

verify_integrity `staticmethod` ¶

verify_integrity(
    entries: list[dict[str, Any]],
) -> tuple[bool, int]

Verify the SHA256 hash chain of audit log entries.

Returns (all_valid, n_verified). Legacy logs without _hash fields return (True, 0) unless SPO_AUDIT_KEY is configured.

Parameters¶

entries : list[dict[str, Any]] Audit-log entries to operate on.

Returns¶

tuple[bool, int] A (all_valid, n_verified) pair for the hash chain.

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

@staticmethod
def verify_integrity(entries: list[dict[str, Any]]) -> tuple[bool, int]:
    """Verify the SHA256 hash chain of audit log entries.

    Returns (all_valid, n_verified).  Legacy logs without ``_hash``
    fields return (True, 0) unless ``SPO_AUDIT_KEY`` is configured.

    Parameters
    ----------
    entries : list[dict[str, Any]]
        Audit-log entries to operate on.

    Returns
    -------
    tuple[bool, int]
        A ``(all_valid, n_verified)`` pair for the hash chain.
    """
    try:
        audit_keys = audit_verification_keys()
    except ValueError:
        return False, 0
    require_signature = bool(audit_keys)
    prev = _ZERO_HASH
    verified = 0
    expected_sequence = 1
    for entry in entries:
        stored = entry.get("_hash")
        if stored is None:
            if require_signature:
                return False, verified
            continue
        without_hash = {k: v for k, v in entry.items() if k != "_hash"}
        try:
            json_line = _canonical_audit_json(without_hash)
        except ValueError:
            return False, verified
        expected = hashlib.sha256((prev + json_line).encode()).hexdigest()
        if expected != stored:
            return False, verified
        if require_signature and not _verify_hmac_signature(
            entry,
            audit_keys,
            expected_previous_hash=prev,
            expected_sequence=expected_sequence,
        ):
            return False, verified
        prev = stored
        verified += 1
        expected_sequence += 1
    return True, verified

verify_determinism_sl_chained ¶

verify_determinism_sl_chained(
    engine: StuartLandauEngine,
    entries: list[dict[str, Any]],
    atol: float = 1e-06,
) -> tuple[bool, int]

Chained multi-step replay for Stuart-Landau engine.

Supports two log formats: - New format: separate 'phases' (N) + 'amplitudes' (N) fields. - Legacy format: 'phases' holds the full SL state (2N) with 'mu' present. When neither mu nor amplitudes are present, skips with a warning. Returns (passed, n_verified).

Parameters¶

engine : StuartLandauEngine The engine used to replay logged steps. entries : list[dict[str, Any]] Audit-log entries to operate on. atol : float Absolute comparison tolerance.

Returns¶

tuple[bool, int] A (passed, n_verified) pair for the Stuart-Landau replay.

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

def verify_determinism_sl_chained(
    self,
    engine: StuartLandauEngine,
    entries: list[dict[str, Any]],
    atol: float = 1e-6,
) -> tuple[bool, int]:
    """Chained multi-step replay for Stuart-Landau engine.

    Supports two log formats:
    - New format: separate 'phases' (N) + 'amplitudes' (N) fields.
    - Legacy format: 'phases' holds the full SL state (2N) with 'mu' present.
    When neither mu nor amplitudes are present, skips with a warning.
    Returns (passed, n_verified).

    Parameters
    ----------
    engine : StuartLandauEngine
        The engine used to replay logged steps.
    entries : list[dict[str, Any]]
        Audit-log entries to operate on.
    atol : float
        Absolute comparison tolerance.

    Returns
    -------
    tuple[bool, int]
        A ``(passed, n_verified)`` pair for the Stuart-Landau replay.
    """
    replayable = self.step_entries(entries)
    if len(replayable) < 2:
        return True, 0

    verified = 0
    for i in range(len(replayable) - 1):
        curr = replayable[i]
        nxt = replayable[i + 1]
        if not _has_fields(curr, _SL_REPLAY_FIELDS) or "phases" not in nxt:
            return False, verified

        try:
            omegas = np.asarray(curr["omegas"])
            n = len(omegas)

            if "amplitudes" in curr:
                state = np.concatenate(
                    [np.asarray(curr["phases"]), np.asarray(curr["amplitudes"])]
                )
            elif "mu" in curr:
                # Legacy: full SL state [theta; r] stored in 'phases'
                state = np.asarray(curr["phases"])
            else:
                _log.warning(
                    "SL replay step %d: amplitude fields missing, skipping", i
                )
                continue

            knm_flat = np.asarray(curr["knm"])
            alpha_flat = np.asarray(curr["alpha"])
            zeta = curr.get("zeta", 0.0)
            psi_drive = curr.get("psi_drive", 0.0)

            knm_arr = knm_flat.reshape(n, n) if knm_flat.ndim == 1 else knm_flat
            alpha_arr = (
                alpha_flat.reshape(n, n) if alpha_flat.ndim == 1 else alpha_flat
            )

            mu = np.asarray(curr.get("mu", np.zeros(n)))
            knm_r_flat = np.asarray(curr.get("knm_r", np.zeros(n * n)))
            knm_r = knm_r_flat.reshape(n, n) if knm_r_flat.ndim == 1 else knm_r_flat
            epsilon = curr.get("epsilon", 1.0)

            computed = engine.step(
                state,
                omegas,
                mu,
                knm_arr,
                knm_r,
                zeta,
                psi_drive,
                alpha_arr,
                epsilon=epsilon,
            )
        except (TypeError, ValueError):
            return False, verified

        try:
            if "amplitudes" in nxt:
                logged_next = np.concatenate(
                    [np.asarray(nxt["phases"]), np.asarray(nxt["amplitudes"])]
                )
            else:
                logged_next = np.asarray(nxt["phases"])

            if not np.allclose(computed, logged_next, atol=atol):
                return False, verified
        except (TypeError, ValueError):
            return False, verified
        verified += 1

    return True, verified

verify_determinism ¶

verify_determinism(
    engine: UPDEEngine, steps: list[dict[str, Any]]
) -> bool

Re-run logged steps and compare global order parameter R.

Requires steps to include 'phases', 'omegas', 'knm', 'zeta', 'psi', 'alpha' fields for full replay. Compares replayed global R against logged 'R' (or 'r_global') field.

Parameters¶

engine : UPDEEngine The engine used to replay logged steps. steps : list[dict[str, Any]] Number of simulation steps to run.

Returns¶

bool True when the replayed order parameter matches the log.

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

def verify_determinism(
    self, engine: UPDEEngine, steps: list[dict[str, Any]]
) -> bool:
    """Re-run logged steps and compare global order parameter R.

    Requires steps to include 'phases', 'omegas', 'knm', 'zeta', 'psi',
    'alpha' fields for full replay. Compares replayed global R against
    logged 'R' (or 'r_global') field.

    Parameters
    ----------
    engine : UPDEEngine
        The engine used to replay logged steps.
    steps : list[dict[str, Any]]
        Number of simulation steps to run.

    Returns
    -------
    bool
        ``True`` when the replayed order parameter matches the log.
    """
    for entry in steps:
        if "phases" not in entry:
            continue
        phases = np.asarray(entry["phases"])
        omegas = np.asarray(entry["omegas"])
        knm = np.asarray(entry["knm"])
        alpha = np.asarray(entry["alpha"])
        zeta = entry.get("zeta", 0.0)
        psi_drive = entry.get("psi_drive", 0.0)

        new_phases = engine.step(phases, omegas, knm, zeta, psi_drive, alpha)
        r_actual, _ = engine.compute_order_parameter(new_phases)

        # Compare against logged global R (same quantity as compute_order_parameter)
        r_logged = entry.get("R") or entry.get("r_global")
        if r_logged is not None and abs(r_actual - r_logged) > 1e-6:
            return False
    return True

Functions:¶

Protobuf Event Stream¶

scpn_phase_orchestrator.runtime.audit_stream provides the event-sourced stream layer. The schema is tracked in proto/audit.proto and packaged as scpn_phase_orchestrator/audit/audit.proto.

from scpn_phase_orchestrator.runtime.audit_stream import (
    read_event_stream,
    verify_event_stream_integrity,
)

events = read_event_stream("audit.spoa")
ok, verified = verify_event_stream_integrity(events)

The stream is not a replacement for deterministic replay; it is the live transport for the same audit records. The JSONL file remains the compatibility format for existing reports and replay tooling.

audit_stream ¶

Event-sourced audit stream backed by length-delimited protobuf envelopes.

Classes¶

AuditStreamEvent `dataclass` ¶

AuditStreamEvent(
    schema_version: int,
    stream_id: str,
    sequence: int,
    event_type: str,
    recorded_at_unix_ns: int,
    source: str,
    previous_hash: str,
    payload_json: str,
    payload_sha256: str,
    event_hash: str,
    signature_algorithm: str,
    signature_key_id: str,
    signature: str,
    audit_mode: str,
    payload: Payload,
)

Decoded audit event envelope with parsed JSON payload.

EventStreamWriter ¶

EventStreamWriter(
    path: str | Path, *, stream_id: str = "spo-audit"
)

Append length-delimited protobuf audit events to a stream file.

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

def __init__(self, path: str | Path, *, stream_id: str = "spo-audit") -> None:
    self._path = Path(path)
    self._stream_id = _validate_stream_id(stream_id)
    self._sequence = 0
    self._previous_hash = ZERO_HASH
    self._audit_key = os.environ.get("SPO_AUDIT_KEY")
    if self._audit_key == "":
        raise ValueError("SPO_AUDIT_KEY must not be empty")
    is_new = not self._path.exists() or self._path.stat().st_size == 0
    self._fh = self._path.open("ab", buffering=0)
    if is_new:
        self._fh.write(STREAM_MAGIC)
    else:
        events = read_event_stream(self._path)
        if events:
            last = events[-1]
            self._sequence = last.sequence
            self._previous_hash = last.event_hash

Attributes¶

path `property` ¶

path: Path

Return the stream file path written by this writer.

Returns¶

Path The protobuf audit stream path.

Methods:¶

write ¶

write(
    payload: Payload, *, event_type: str | None = None
) -> None

Append one payload as a hashed and optionally signed audit event.

Parameters¶

payload : Payload The event or wire payload. event_type : str | None Named event type, or None.

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

def write(self, payload: Payload, *, event_type: str | None = None) -> None:
    """Append one payload as a hashed and optionally signed audit event.

    Parameters
    ----------
    payload : Payload
        The event or wire payload.
    event_type : str | None
        Named event type, or ``None``.
    """
    canonical_payload = _canonical_json(payload)
    payload_sha256 = hashlib.sha256(canonical_payload.encode()).hexdigest()
    self._sequence += 1
    now_ns = time.time_ns()
    resolved_type = event_type or _event_type_for_payload(payload)
    event_hash = _event_hash(
        stream_id=self._stream_id,
        sequence=self._sequence,
        event_type=resolved_type,
        recorded_at_unix_ns=now_ns,
        source="spo",
        previous_hash=self._previous_hash,
        payload_sha256=payload_sha256,
        schema_version=SCHEMA_VERSION,
    )
    signature = ""
    signature_key_id = ""
    signature_algorithm = ""
    audit_mode = "unsigned-development"
    if self._audit_key is not None:
        audit_mode = "hmac-signed"
        signature_algorithm = SIGNATURE_ALGORITHM
        signature_key_id = key_id_for_secret(self._audit_key)
        signature = hmac.new(
            self._audit_key.encode(),
            _signature_material(
                audit_mode=audit_mode,
                stream_id=self._stream_id,
                sequence=self._sequence,
                event_type=resolved_type,
                recorded_at_unix_ns=now_ns,
                source="spo",
                previous_hash=self._previous_hash,
                payload_sha256=payload_sha256,
                event_hash=event_hash,
                schema_version=SCHEMA_VERSION,
                key_id=signature_key_id,
            ).encode(),
            hashlib.sha256,
        ).hexdigest()
    message = cast("Any", _AuditEnvelope())
    message.schema_version = SCHEMA_VERSION
    message.stream_id = self._stream_id
    message.sequence = self._sequence
    message.event_type = resolved_type
    recorded_at = message.recorded_at
    recorded_at.seconds = now_ns // 1_000_000_000
    recorded_at.nanos = now_ns % 1_000_000_000
    message.source = "spo"
    message.previous_hash = self._previous_hash
    message.payload_json = canonical_payload
    message.payload_sha256 = payload_sha256
    message.event_hash = event_hash
    message.signature_algorithm = signature_algorithm
    message.signature_key_id = signature_key_id
    message.signature = signature
    message.audit_mode = audit_mode
    raw = message.SerializeToString(deterministic=True)
    self._fh.write(_encode_varint(len(raw)))
    self._fh.write(raw)
    self._previous_hash = event_hash

flush ¶

flush() -> None

Flush buffered audit bytes without closing the stream handle.

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

def flush(self) -> None:
    """Flush buffered audit bytes without closing the stream handle."""
    self._fh.flush()

close ¶

close() -> None

Flush buffered audit bytes and close the underlying stream handle.

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

def close(self) -> None:
    """Flush buffered audit bytes and close the underlying stream handle."""
    self._fh.flush()
    self._fh.close()

Functions:¶

read_event_stream ¶

read_event_stream(
    path: str | Path,
) -> list[AuditStreamEvent]

Read all protobuf events from an SPO audit stream.

Parameters¶

path : str | Path Filesystem path to the target file.

Returns¶

list[AuditStreamEvent] The decoded audit stream events.

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

def read_event_stream(path: str | Path) -> list[AuditStreamEvent]:
    """Read all protobuf events from an SPO audit stream.

    Parameters
    ----------
    path : str | Path
        Filesystem path to the target file.

    Returns
    -------
    list[AuditStreamEvent]
        The decoded audit stream events.
    """
    with Path(path).open("rb") as fh:
        return _read_events_from_handle(fh)

iter_event_stream ¶

iter_event_stream(
    path: str | Path,
    *,
    from_start: bool = False,
    poll_interval_s: float = 0.2,
) -> Iterator[AuditStreamEvent]

Yield existing and newly appended stream events in order.

Parameters¶

path : str | Path Filesystem path to the target file. from_start : bool Whether to replay from the start of the stream. poll_interval_s : float Poll interval in seconds.

Returns¶

Iterator[AuditStreamEvent] An iterator over existing and newly appended stream events.

Raises¶

FileNotFoundError If the stream file does not exist.

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

def iter_event_stream(
    path: str | Path,
    *,
    from_start: bool = False,
    poll_interval_s: float = 0.2,
) -> Iterator[AuditStreamEvent]:
    """Yield existing and newly appended stream events in order.

    Parameters
    ----------
    path : str | Path
        Filesystem path to the target file.
    from_start : bool
        Whether to replay from the start of the stream.
    poll_interval_s : float
        Poll interval in seconds.

    Returns
    -------
    Iterator[AuditStreamEvent]
        An iterator over existing and newly appended stream events.

    Raises
    ------
    FileNotFoundError
        If the stream file does not exist.
    """
    poll_interval = _validate_poll_interval_s(poll_interval_s)
    path_obj = Path(path)
    if not from_start and not path_obj.exists():
        raise FileNotFoundError(f"audit event stream path does not exist: {path_obj}")
    offset = 0
    if not from_start and path_obj.exists():
        offset = len(read_event_stream(path_obj))
    while True:
        current = read_event_stream(path_obj) if path_obj.exists() else []
        yield from current[offset:]
        offset = len(current)
        time.sleep(poll_interval)

tail_event_stream ¶

tail_event_stream(
    path: str | Path,
    *,
    from_start: bool = False,
    max_events: int | None = None,
    poll_interval_s: float = 0.2,
) -> list[AuditStreamEvent]

Tail a stream file until max_events decoded events are available.

Use :func:iter_event_stream for unbounded live streaming.

Parameters¶

path : str | Path Filesystem path to the target file. from_start : bool Whether to replay from the start of the stream. max_events : int | None Maximum number of events to read, or None. poll_interval_s : float Poll interval in seconds.

Returns¶

list[AuditStreamEvent] The decoded events, up to max_events.

Raises¶

ValueError If the inputs are invalid or inconsistent.

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

def tail_event_stream(
    path: str | Path,
    *,
    from_start: bool = False,
    max_events: int | None = None,
    poll_interval_s: float = 0.2,
) -> list[AuditStreamEvent]:
    """Tail a stream file until ``max_events`` decoded events are available.

    Use :func:`iter_event_stream` for unbounded live streaming.

    Parameters
    ----------
    path : str | Path
        Filesystem path to the target file.
    from_start : bool
        Whether to replay from the start of the stream.
    max_events : int | None
        Maximum number of events to read, or ``None``.
    poll_interval_s : float
        Poll interval in seconds.

    Returns
    -------
    list[AuditStreamEvent]
        The decoded events, up to ``max_events``.

    Raises
    ------
    ValueError
        If the inputs are invalid or inconsistent.
    """
    if max_events is None:
        raise ValueError("max_events is required for bounded tail_event_stream")
    if (
        isinstance(max_events, bool)
        or not isinstance(max_events, int)
        or max_events <= 0
    ):
        raise ValueError("max_events must be a positive integer")
    poll_interval = _validate_poll_interval_s(poll_interval_s)
    events: list[AuditStreamEvent] = []
    iterator = iter_event_stream(
        path,
        from_start=from_start,
        poll_interval_s=poll_interval,
    )
    while len(events) < max_events:
        events.append(next(iterator))
    return events

verify_event_stream_integrity ¶

verify_event_stream_integrity(
    events: list[AuditStreamEvent],
) -> tuple[bool, int]

Verify payload digests, sequence continuity, and event hash chaining.

Parameters¶

events : list[AuditStreamEvent] The decoded audit stream events.

Returns¶

tuple[bool, int] A (ok, count) pair: integrity flag and verified event count.

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

def verify_event_stream_integrity(
    events: list[AuditStreamEvent],
) -> tuple[bool, int]:
    """Verify payload digests, sequence continuity, and event hash chaining.

    Parameters
    ----------
    events : list[AuditStreamEvent]
        The decoded audit stream events.

    Returns
    -------
    tuple[bool, int]
        A ``(ok, count)`` pair: integrity flag and verified event count.
    """
    try:
        audit_keys = audit_verification_keys()
    except ValueError:
        return False, 0
    require_signature = bool(audit_keys)
    previous_hash = ZERO_HASH
    expected_sequence = 1
    verified = 0
    for event in events:
        canonical_payload = _canonical_json(event.payload)
        payload_sha256 = hashlib.sha256(canonical_payload.encode()).hexdigest()
        expected_hash = _event_hash(
            stream_id=event.stream_id,
            sequence=event.sequence,
            event_type=event.event_type,
            recorded_at_unix_ns=event.recorded_at_unix_ns,
            source=event.source,
            previous_hash=previous_hash,
            payload_sha256=payload_sha256,
            schema_version=event.schema_version,
        )
        if event.sequence != expected_sequence:
            return False, verified
        if event.previous_hash != previous_hash:
            return False, verified
        if event.payload_sha256 != payload_sha256:
            return False, verified
        if event.event_hash != expected_hash:
            return False, verified
        if require_signature and not _verify_event_signature(
            event,
            audit_keys,
            expected_hash=expected_hash,
            expected_previous_hash=previous_hash,
        ):
            return False, verified
        previous_hash = event.event_hash
        expected_sequence += 1
        verified += 1
    return True, verified

Pipeline integration¶

The audit logger sits at the output of the supervisor loop:

SupervisorPolicy.decide() ──→ list[ControlAction]
                                      │
                               ┌──────┼──────┐
                               ↓      ↓      ↓
                          Actuator  Audit   EventBus
                                   Logger
                                     │
                              audit.jsonl (append)
                              audit.spoa (append)
                                     │
                              SHA-256 chain

Every regime transition, actuation command, and boundary violation is recorded. The audit trail is the authoritative record of what the system did and why.

AuditLogger API¶

AuditLogger(log_path: str | Path, *, event_stream: str | Path | None = None)

Method	Signature	Description
`log_header`	`(n_oscillators, dt, method, seed, amplitude_mode)`	Engine config record
`log_step`	`(step, upde_state, actions, *, phases, omegas, knm, alpha, zeta, psi_drive, amplitudes, mu, knm_r, epsilon)`	Full simulation step
`log_event`	`(event_type: str, data: dict)`	Named event with arbitrary data
`close`	`()`	Flush and close file handle

Supports context manager (with AuditLogger(...) as logger:).

log_step optionally records full engine state (phases, omegas, knm, alpha) for deterministic replay. When phases is provided, omegas, knm, and alpha are required (raises AuditError otherwise). Stuart-Landau fields (amplitudes, mu, knm_r, epsilon) are optional.

Audit in operational workflows¶

Audit records are used in three recurring workflows:

Incident reconstruction: replay the same binding spec and input state, then compare verify_determinism() against the original trace.
Release review: use log_header and per-step metadata to confirm that backend choice, precision mode, and adapter set were constant across candidate runs.
Policy validation: compare regime transitions, action frequency, and boundary hits before promoting rules to wider environments.

Use signed mode when operational evidence is required (SPO_AUDIT_KEY or SPO_AUDIT_KEYRING). Unsigned logs remain available for local development, but they are explicitly flagged and must not be used as production evidence.

ReplayEngine API¶

ReplayEngine(log_path: str | Path)

Method	Signature	Description
`load`	`() → list[dict]`	Parse all JSONL entries
`load_header`	`(entries) → dict \\| None`	Extract engine config
`step_entries`	`(entries) → list[dict]`	Filter to replayable steps
`build_engine`	`(header) → UPDEEngine \\| StuartLandauEngine`	Reconstruct engine from header
`verify_integrity`	`(entries) → (bool, int)`	Verify SHA-256 chain
`verify_determinism`	`(engine, steps) → bool`	Compare replayed R to logged R
`verify_determinism_chained`	`(engine, entries, atol) → (bool, int)`	Multi-step replay: output N = input N+1
`verify_determinism_sl_chained`	`(engine, entries, atol) → (bool, int)`	Stuart-Landau chained replay

Hash chain verification algorithm¶

prev_hash = "0" * 64  # genesis hash
for record in records:
    stored_hash = record.pop("_hash")
    content = json.dumps(record, separators=(",", ":"))
    expected = sha256((prev_hash + content).encode()).hexdigest()
    assert stored_hash == expected  # tamper detection
    prev_hash = stored_hash

Compliance references¶

NIST SP 800-92: Guide to Computer Security Log Management
IEC 62443: Industrial communication networks security
ISO 27001 A.12.4: Logging and monitoring

Audit¶

Motivation¶

Hash Chain Structure¶

Keyed Audit Signatures¶

Audit Logger¶

audit_logger ¶

Classes¶

AuditStreamIntegrityResult dataclass ¶

Parameters¶

Methods:¶

to_audit_record ¶

Returns¶

AuditLogger ¶

Attributes¶

event_stream_integrity property ¶

Returns¶

Methods:¶

log_header ¶

Parameters¶

Raises¶

log_step ¶

Parameters¶

Raises¶

log_event ¶

Parameters¶

Raises¶

close ¶

verify_event_stream_integrity ¶

Returns¶

Functions:¶

Audit Signing Helpers¶

audit_signing ¶

Functions:¶

key_id_for_secret ¶

Parameters¶

Returns¶

Raises¶

audit_verification_keys ¶

Returns¶

Raises¶

Deterministic Replay¶

replay ¶

Classes¶

ReplayEngine ¶

Methods:¶

load ¶

Returns¶

replay_step ¶

Parameters¶

Returns¶

load_header ¶

Parameters¶

Returns¶

step_entries ¶

Parameters¶

Returns¶

build_engine ¶

Parameters¶

Returns¶

verify_determinism_chained ¶

Parameters¶

Returns¶

verify_integrity staticmethod ¶

Parameters¶

Returns¶

verify_determinism_sl_chained ¶

Parameters¶

Returns¶

verify_determinism ¶

Parameters¶

Returns¶

Functions:¶

Protobuf Event Stream¶

audit_stream ¶

Classes¶

AuditStreamEvent dataclass ¶

EventStreamWriter ¶

Attributes¶

path property ¶

Returns¶

AuditStreamIntegrityResult `dataclass` ¶

event_stream_integrity `property` ¶

verify_integrity `staticmethod` ¶

AuditStreamEvent `dataclass` ¶

path `property` ¶