acdl/core/metrics/decision_ledger.py

"""Nova Decision Ledger — SQLite append-only hash-chain (REQ-188, D-121).

Extends outbox_writer.py to emit to a SQLite append-only table with a hash
chain (prev_hash + own hash, SHA-256). Stores ai.decision.made events
(decision_id=run_id, chosen_action=band, confidence=score,
alternatives=perInput, human_override=HITL block) with outcome backfill
from apply.completed. Also stores attestation.recorded events (D-132).

Honors D-083 (no S3 Object Lock/JWS — local SQLite hash-chain only).
D-120: Nova-native (SQLite, no QLDB).
D-128: metrics/ at repo root.
"""

import datetime
import hashlib
import json
import os
import sqlite3
import sys

_LEDGER_PATH = os.path.join(
    os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))),
    "metrics", "decision_ledger.db",
)

_GENESIS_HASH = "GENESIS"


def _iso8601_now():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _canonical_hash(event):
    """SHA-256 over canonical JSON (sort_keys, compact separators)."""
    canonical = json.dumps(event, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _init_db(db_path=None):
    """Create the ledger table if it doesn't exist."""
    if db_path is None:
        db_path = _LEDGER_PATH
    os.makedirs(os.path.dirname(db_path), exist_ok=True)
    conn = sqlite3.connect(db_path)
    conn.execute("""
        CREATE TABLE IF NOT EXISTS decision_ledger (
            seq INTEGER PRIMARY KEY AUTOINCREMENT,
            event_id TEXT NOT NULL,
            event_type TEXT NOT NULL,
            run_id TEXT NOT NULL,
            contract_id TEXT,
            environment TEXT,
            event_time TEXT NOT NULL,
            payload TEXT NOT NULL,
            prev_hash TEXT NOT NULL,
            hash TEXT NOT NULL
        )
    """)
    conn.execute("CREATE INDEX IF NOT EXISTS idx_run_id ON decision_ledger(run_id)")
    conn.execute("CREATE INDEX IF NOT EXISTS idx_event_type ON decision_ledger(event_type)")
    conn.commit()
    conn.close()


def _get_last_hash(db_path=None):
    """Get the hash of the last row in the ledger (or GENESIS if empty)."""
    if db_path is None:
        db_path = _LEDGER_PATH
    conn = sqlite3.connect(db_path)
    row = conn.execute("SELECT hash FROM decision_ledger ORDER BY seq DESC LIMIT 1").fetchone()
    conn.close()
    return row[0] if row else _GENESIS_HASH


def append(event, db_path=None):
    """Append an event to the Decision Ledger with hash-chain integrity.

    Args:
        event: a CloudEvents 1.0 envelope dict (from event_envelope.make_event)
        db_path: path to the SQLite ledger

    Returns:
        The row dict (seq, event_id, event_type, run_id, hash, prev_hash).
    """
    if db_path is None:
        db_path = _LEDGER_PATH
    _init_db(db_path)
    prev_hash = _get_last_hash(db_path)
    event_hash = _canonical_hash(event)
    platform = event.get("platform", {})
    data = event.get("data", {})

    conn = sqlite3.connect(db_path)
    conn.execute("BEGIN IMMEDIATE")
    cursor = conn.execute(
        """INSERT INTO decision_ledger
           (event_id, event_type, run_id, contract_id, environment, event_time, payload, prev_hash, hash)
           VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
        (
            event.get("id", ""),
            event.get("type", ""),
            platform.get("run_id", ""),
            platform.get("contract_id", ""),
            platform.get("environment", ""),
            event.get("time", _iso8601_now()),
            json.dumps(event, sort_keys=True),
            prev_hash,
            event_hash,
        ),
    )
    seq = cursor.lastrowid
    conn.commit()
    conn.close()
    return {"seq": seq, "event_id": event.get("id", ""), "event_type": event.get("type", ""),
            "run_id": platform.get("run_id", ""), "hash": event_hash, "prev_hash": prev_hash}


def verify_chain(db_path=None):
    """Verify the hash chain integrity. Returns (ok, broken_count, details).

    Recomputes each row's hash from its payload and checks:
    1. The stored hash matches the recomputed hash.
    2. The prev_hash matches the previous row's hash.
    """
    if db_path is None:
        db_path = _LEDGER_PATH
    _init_db(db_path)
    conn = sqlite3.connect(db_path)
    rows = conn.execute("SELECT seq, hash, prev_hash, payload FROM decision_ledger ORDER BY seq").fetchall()
    conn.close()
    if not rows:
        return True, 0, "empty ledger"

    broken = 0
    details = []
    prev_hash = _GENESIS_HASH
    for seq, stored_hash, stored_prev, payload_json in rows:
        event = json.loads(payload_json)
        recomputed = _canonical_hash(event)
        if recomputed != stored_hash:
            broken += 1
            details.append(f"seq={seq}: hash mismatch (stored={stored_hash[:12]}... recomputed={recomputed[:12]}...)")
        if stored_prev != prev_hash:
            broken += 1
            details.append(f"seq={seq}: prev_hash mismatch (expected={prev_hash[:12]}... got={stored_prev[:12]}...)")
        prev_hash = stored_hash
    return broken == 0, broken, "; ".join(details) if details else "chain intact"


def query_by_run(run_id, db_path=None):
    """Query all ledger entries for a given run_id."""
    if db_path is None:
        db_path = _LEDGER_PATH
    _init_db(db_path)
    conn = sqlite3.connect(db_path)
    rows = conn.execute(
        "SELECT seq, event_type, event_time, payload FROM decision_ledger WHERE run_id = ? ORDER BY seq",
        (run_id,),
    ).fetchall()
    conn.close()
    return [{"seq": r[0], "event_type": r[1], "event_time": r[2], "payload": json.loads(r[3])} for r in rows]


def stats(db_path=None):
    """Return ledger statistics."""
    if db_path is None:
        db_path = _LEDGER_PATH
    _init_db(db_path)
    conn = sqlite3.connect(db_path)
    total = conn.execute("SELECT COUNT(*) FROM decision_ledger").fetchone()[0]
    by_type = conn.execute("SELECT event_type, COUNT(*) FROM decision_ledger GROUP BY event_type").fetchall()
    by_env = conn.execute("SELECT environment, COUNT(*) FROM decision_ledger GROUP BY environment").fetchall()
    conn.close()
    return {
        "total": total,
        "by_event_type": dict(by_type),
        "by_environment": dict(by_env),
    }


def export_since(since_iso, fmt="json", db_path=None):
    """Export ledger entries since a given ISO8601 timestamp."""
    if db_path is None:
        db_path = _LEDGER_PATH
    _init_db(db_path)
    conn = sqlite3.connect(db_path)
    rows = conn.execute(
        "SELECT seq, event_type, run_id, event_time, payload FROM decision_ledger WHERE event_time >= ? ORDER BY seq",
        (since_iso,),
    ).fetchall()
    conn.close()
    entries = [{"seq": r[0], "event_type": r[1], "run_id": r[2], "event_time": r[3], "payload": json.loads(r[4])} for r in rows]
    if fmt == "csv":
        import csv
        import io
        buf = io.StringIO()
        writer = csv.DictWriter(buf, fieldnames=["seq", "event_type", "run_id", "event_time", "payload"])
        writer.writeheader()
        for e in entries:
            e["payload"] = json.dumps(e["payload"])
            writer.writerow(e)
        return buf.getvalue()
    return json.dumps(entries, indent=2)


def replay_run(run_id, db_path=None):
    """Reconstruct a run's full event sequence from the ledger.

    Prints the ordered event sequence (run.started -> policy.evaluated ->
    confidence.computed -> ai.decision.made -> attestation.recorded ->
    run.completed/failed) with the decision's confidence, alternatives,
    and outcome.
    """
    if db_path is None:
        db_path = _LEDGER_PATH
    entries = query_by_run(run_id, db_path)
    if not entries:
        return f"no events found for run_id={run_id}"
    lines = [f"=== Replay: run_id={run_id} ({len(entries)} events) ==="]
    for e in entries:
        payload = e["payload"]
        data = payload.get("data", {})
        etype = e["event_type"]
        line = f"  [{e['seq']}] {e['event_time']} {etype}"
        if etype == "nova.ai.decision.made":
            line += f" confidence={data.get('confidence', '?')} band={data.get('chosen_action', '?')} override={data.get('human_override', '?')}"
            if data.get("escalation_reason"):
                line += f" escalation_reason={data.get('escalation_reason')}"
        elif etype == "nova.attestation.recorded":
            line += f" env={data.get('environment', '?')} approver={data.get('approver', '?')} result={data.get('result', '?')}"
        elif etype == "nova.run.completed":
            line += f" exit={data.get('exit_code', '?')} outcome={data.get('outcome', '?')}"
        elif etype == "nova.run.failed":
            line += f" exit={data.get('exit_code', '?')} outcome=failed"
        elif etype == "nova.outcome.backfilled":
            line += f" prev={data.get('previous_outcome', '?')} new={data.get('new_outcome', '?')} at={data.get('backfilled_at', '?')}"
        lines.append(line)
    lines.append("=== End replay ===")
    return "\n".join(lines)


if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("usage: decision_ledger.py <verify-chain|stats|query|export|replay> [args]", file=sys.stderr)
        sys.exit(2)
    cmd = sys.argv[1]
    if cmd == "verify-chain":
        ok, broken, details = verify_chain()
        print(f"chain_ok={ok} broken={broken} details={details}")
        sys.exit(0 if ok else 1)
    elif cmd == "stats":
        print(json.dumps(stats(), indent=2))
    elif cmd == "query" and len(sys.argv) >= 3:
        print(json.dumps(query_by_run(sys.argv[2]), indent=2))
    elif cmd == "export" and len(sys.argv) >= 3:
        print(export_since(sys.argv[2]))
    elif cmd == "replay" and len(sys.argv) >= 3:
        print(replay_run(sys.argv[2]))
    else:
        print(f"unknown command: {cmd}", file=sys.stderr)
        sys.exit(2)