""" 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