runtime_src_primordial_os_audit_log.py

"""
Primordial OS Runtime — Audit Log (Phase 6 / Phase 9)
Append-only hash-chained audit logging for runtime decisions.

Each event carries a SHA-256 hash of its own content plus the preceding
event's hash, forming a tamper-evident chain. Phase 9 adds ergonomic
chain helpers: create_runtime_audit_event and append_to_audit_chain.

In-memory audit demonstration only. Not clinical software. Not a medical
device. No health-domain inference, treatment, or decision-making
functionality is present or intended. Does not persist protected health
information or personally identifiable information.
"""

from __future__ import annotations

import hashlib
import json
from dataclasses import dataclass
from datetime import datetime, timezone

from primordial_os.runtime import RuntimeDecision


# ---------------------------------------------------------------------------
# Hash computation
# ---------------------------------------------------------------------------

def compute_event_hash(
    timestamp: str,
    event_type: str,
    final_gate_state: str,
    hard_stop: bool,
    resonance: float,
    pressure_adjusted_stability: float,
    max_oam_severity: float,
    summary: str,
    previous_hash: str | None,
) -> str:
    """Return the SHA-256 hex digest of the canonical serialization of event fields."""
    payload = json.dumps(
        {
            "timestamp": timestamp,
            "event_type": event_type,
            "final_gate_state": final_gate_state,
            "hard_stop": hard_stop,
            "resonance": resonance,
            "pressure_adjusted_stability": pressure_adjusted_stability,
            "max_oam_severity": max_oam_severity,
            "summary": summary,
            "previous_hash": previous_hash,
        },
        sort_keys=True,
        separators=(",", ":"),
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Audit event model
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class AuditEvent:
    """
    A single immutable audit record in the hash chain.

    previous_hash is None for the first event in a chain.
    current_hash is SHA-256 of all other fields plus previous_hash.
    """
    timestamp: str
    event_type: str
    final_gate_state: str
    hard_stop: bool
    resonance: float
    pressure_adjusted_stability: float
    max_oam_severity: float
    summary: str
    previous_hash: str | None
    current_hash: str


# ---------------------------------------------------------------------------
# Factory
# ---------------------------------------------------------------------------

def create_audit_event(
    runtime_decision: RuntimeDecision,
    event_type: str = "runtime_evaluation",
    previous_hash: str | None = None,
) -> AuditEvent:
    """Create an AuditEvent from a RuntimeDecision and link it to the chain."""
    timestamp = datetime.now(timezone.utc).isoformat()
    final_gate_state = runtime_decision.final_gate_state.value
    hard_stop = runtime_decision.hard_stop
    resonance = runtime_decision.hir_result.resonance
    pas = runtime_decision.hir_result.pressure_adjusted_stability
    max_oam_severity = runtime_decision.oam_assessment.max_severity
    summary = runtime_decision.summary

    current_hash = compute_event_hash(
        timestamp=timestamp,
        event_type=event_type,
        final_gate_state=final_gate_state,
        hard_stop=hard_stop,
        resonance=resonance,
        pressure_adjusted_stability=pas,
        max_oam_severity=max_oam_severity,
        summary=summary,
        previous_hash=previous_hash,
    )

    return AuditEvent(
        timestamp=timestamp,
        event_type=event_type,
        final_gate_state=final_gate_state,
        hard_stop=hard_stop,
        resonance=resonance,
        pressure_adjusted_stability=pas,
        max_oam_severity=max_oam_severity,
        summary=summary,
        previous_hash=previous_hash,
        current_hash=current_hash,
    )


# ---------------------------------------------------------------------------
# Chain verification
# ---------------------------------------------------------------------------

def create_runtime_audit_event(
    runtime_decision: RuntimeDecision,
    event_type: str = "runtime_evaluation",
) -> AuditEvent:
    """Create a standalone AuditEvent with previous_hash=None."""
    return create_audit_event(runtime_decision, event_type=event_type, previous_hash=None)


def append_to_audit_chain(
    events: tuple[AuditEvent, ...] | list[AuditEvent],
    runtime_decision: RuntimeDecision,
    event_type: str = "runtime_evaluation",
) -> tuple[AuditEvent, ...]:
    """
    Append a new AuditEvent to an existing chain and return the new chain as a tuple.

    Does not mutate the input chain. previous_hash is None when the chain is
    empty; otherwise it is the current_hash of the last event in the chain.
    """
    previous_hash = events[-1].current_hash if events else None
    new_event = create_audit_event(
        runtime_decision,
        event_type=event_type,
        previous_hash=previous_hash,
    )
    return tuple(events) + (new_event,)


# ---------------------------------------------------------------------------
# Chain verification
# ---------------------------------------------------------------------------

def verify_audit_chain(events: tuple[AuditEvent, ...] | list[AuditEvent]) -> bool:
    """
    Return True if every event's hash is internally consistent and each
    event is correctly linked to its predecessor.

    An empty list is valid. A single event is valid if its hash is correct.
    """
    for i, event in enumerate(events):
        expected = compute_event_hash(
            timestamp=event.timestamp,
            event_type=event.event_type,
            final_gate_state=event.final_gate_state,
            hard_stop=event.hard_stop,
            resonance=event.resonance,
            pressure_adjusted_stability=event.pressure_adjusted_stability,
            max_oam_severity=event.max_oam_severity,
            summary=event.summary,
            previous_hash=event.previous_hash,
        )
        if event.current_hash != expected:
            return False
        if i > 0 and event.previous_hash != events[i - 1].current_hash:
            return False

    return True