Audit
July 1, 2026 · View on GitHub
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()
::: scpn_phase_orchestrator.runtime.audit_logger
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.
::: scpn_phase_orchestrator.runtime.audit_signing
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}")
::: scpn_phase_orchestrator.runtime.replay
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.
::: scpn_phase_orchestrator.runtime.audit_stream
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_headerand 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