runtime_src_primordial_os_audit_store.py

"""
Primordial OS Runtime — Audit Store (Phase 10)
Safe JSONL file writer and reader for hash-chained audit events.

Writes one JSON object per line (JSONL). Parent directory must exist before
writing. Hidden file paths are rejected. File is created on first append;
existing content is preserved on subsequent appends.

No raw prompts, user input, PHI, PII, or clinical content is written —
only the structural runtime fields that already exist in AuditEvent.

Pre-validation architecture. Not clinical software. Not a medical device.
No health-domain inference, treatment, or decision-making functionality
is present or intended.
"""

from __future__ import annotations

import json
from pathlib import Path

from primordial_os.audit_log import AuditEvent, verify_audit_chain


# ---------------------------------------------------------------------------
# Serialization
# ---------------------------------------------------------------------------

def audit_event_to_dict(event: AuditEvent) -> dict:
    """Return a plain dict of structural runtime fields only.

    Explicit field enumeration ensures no unexpected keys are written.
    No raw prompts, user input, PHI, PII, or clinical content is included.
    """
    return {
        "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,
        "current_hash": event.current_hash,
    }


def audit_event_from_dict(data: dict) -> AuditEvent:
    """Reconstruct an AuditEvent from a plain dict (e.g. parsed from JSONL)."""
    return AuditEvent(
        timestamp=data["timestamp"],
        event_type=data["event_type"],
        final_gate_state=data["final_gate_state"],
        hard_stop=data["hard_stop"],
        resonance=data["resonance"],
        pressure_adjusted_stability=data["pressure_adjusted_stability"],
        max_oam_severity=data["max_oam_severity"],
        summary=data["summary"],
        previous_hash=data["previous_hash"],
        current_hash=data["current_hash"],
    )


# ---------------------------------------------------------------------------
# Path safety
# ---------------------------------------------------------------------------

def _validate_write_path(path: Path) -> None:
    """Raise for unsafe or invalid write targets.

    Checks:
      - hidden file paths (name starts with '.')
      - parent directory does not exist
    """
    if path.name.startswith("."):
        raise ValueError(f"Hidden file paths are not permitted: {path}")
    if not path.parent.exists():
        raise FileNotFoundError(
            f"Parent directory does not exist: {path.parent}"
        )


# ---------------------------------------------------------------------------
# JSONL I/O
# ---------------------------------------------------------------------------

def append_audit_event_jsonl(path: str | Path, event: AuditEvent) -> None:
    """Append one AuditEvent as a single JSON line to the JSONL file at path.

    Safety requirements:
    - Parent directory must exist; will not be created automatically.
    - Hidden file paths (name starts with '.') are rejected.
    - File is created if absent; existing lines are never modified.
    - Encoding: UTF-8.
    - Format: one compact JSON object per line, keys sorted, no trailing spaces.
    - No raw prompts, PHI, PII, or clinical content is written.
    """
    p = Path(path)
    _validate_write_path(p)
    line = json.dumps(
        audit_event_to_dict(event),
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
    )
    with p.open("a", encoding="utf-8") as f:
        f.write(line + "\n")


def read_audit_events_jsonl(path: str | Path) -> tuple[AuditEvent, ...]:
    """Read all AuditEvents from a JSONL file and return them as a tuple.

    Events are returned in file order (append order).

    Raises:
      FileNotFoundError  — if the file does not exist.
      ValueError         — if any line contains malformed JSON.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"Audit log file not found: {p}")
    events: list[AuditEvent] = []
    with p.open("r", encoding="utf-8") as f:
        for line_num, raw_line in enumerate(f, start=1):
            line = raw_line.strip()
            if not line:
                continue
            try:
                data = json.loads(line)
            except json.JSONDecodeError as exc:
                raise ValueError(
                    f"Malformed JSON on line {line_num} of {p}: {exc}"
                ) from exc
            events.append(audit_event_from_dict(data))
    return tuple(events)


def verify_audit_jsonl(path: str | Path) -> bool:
    """Return True if the JSONL file at path contains a valid, unbroken audit chain.

    Returns False — without raising — when:
    - the file is missing
    - the chain hash linkage is broken
    - any event hash does not match its stored fields

    An empty file is treated as a valid (empty) chain and returns True.
    """
    p = Path(path)
    try:
        events = read_audit_events_jsonl(p)
    except FileNotFoundError:
        return False
    return verify_audit_chain(events)