arborist/tests/test_memory_root.py

"""Memory-root tests (ticket #000017).

Covers:
- canonical body and memory_root SHA-256 derivation
- branch_digest stability under repeated input
- branch sort-order invariance
- snapshot determinism on same store state
- store_snapshot idempotency
- falsify and mark_stale audit events + state transitions
- audit chain stays clean across memory ops
"""

from __future__ import annotations

import hashlib

import pytest

from arborist.memory import (
    BranchSummary,
    MemorySnapshot,
    branch_digest,
    branches_for,
    canonical_branches_blob,
    falsify,
    latest,
    load,
    mark_stale,
    memory_root,
    snapshot,
    store_snapshot,
)
from arborist.memory.canonical import SCHEMA_VERSION
from arborist.memory.projections import (
    DEFAULT_BRANCHES,
    PROJECTION_VERSION,
    project_failure_motif_violations,
)
from arborist.store import append_audit, connect, latest_event_hash, transaction


# --- canonical ---------------------------------------------------------


def _branch(branch_id: str, body: bytes, count: int) -> BranchSummary:
    return BranchSummary(
        branch_id=branch_id, summary_blob=body, count=count
    )


def test_branch_digest_stable():
    a = _branch("b", b"hello", 3)
    b = _branch("b", b"hello", 3)
    assert branch_digest(a) == branch_digest(b)


def test_branch_digest_changes_on_content():
    a = _branch("b", b"hello", 3)
    b = _branch("b", b"helloo", 3)
    assert branch_digest(a) != branch_digest(b)


def test_memory_root_invariant_under_branch_order():
    snap_a = MemorySnapshot(
        schema_version=SCHEMA_VERSION,
        parent_memory_root=None,
        audit_events_high_water="genesis",
        branches=(_branch("a", b"x", 1), _branch("b", b"y", 2)),
    )
    snap_b = MemorySnapshot(
        schema_version=SCHEMA_VERSION,
        parent_memory_root=None,
        audit_events_high_water="genesis",
        branches=(_branch("b", b"y", 2), _branch("a", b"x", 1)),
    )
    assert memory_root(snap_a) == memory_root(snap_b)


def test_memory_root_changes_when_high_water_changes():
    snap_a = MemorySnapshot(
        schema_version=SCHEMA_VERSION,
        parent_memory_root=None,
        audit_events_high_water="0xa",
        branches=(),
    )
    snap_b = MemorySnapshot(
        schema_version=SCHEMA_VERSION,
        parent_memory_root=None,
        audit_events_high_water="0xb",
        branches=(),
    )
    assert memory_root(snap_a) != memory_root(snap_b)


# --- snapshot ----------------------------------------------------------


def test_snapshot_builds_default_branches_on_empty(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
        assert ms.schema_version == SCHEMA_VERSION
        # Default branch set produces three summaries.
        ids = sorted(b.branch_id for b in ms.branches)
        assert "audit-mode-distribution" in ids
        assert "failure-motif:violations" in ids
        assert "falsification-state" in ids
    finally:
        conn.close()


def test_snapshot_deterministic_same_state(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            a = snapshot(conn)
            b = snapshot(conn)
        assert memory_root(a) == memory_root(b)
    finally:
        conn.close()


def test_snapshot_picks_up_new_audit_events(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            a = snapshot(conn)
            r1 = memory_root(a)
            append_audit(
                conn,
                event_type="providence_write",
                subject_root="00" * 32,
                body={"audit_mode": "STRICT", "violations": []},
            )
            b = snapshot(conn)
            r2 = memory_root(b)
        assert r1 != r2  # high-water shifted
    finally:
        conn.close()


# --- store_snapshot ---------------------------------------------------


def test_store_snapshot_persists_and_emits_audit(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
            root = store_snapshot(conn, ms)
        loaded = load(conn, root)
        assert loaded is not None
        assert loaded["state"] == "live"

        ev = conn.execute(
            "SELECT 1 FROM audit_events "
            "WHERE event_type='memory_snapshot_landed' AND subject_root = ?",
            (root,),
        ).fetchone()
        assert ev is not None

        # Branch rows present.
        rows = branches_for(conn, root)
        ids = {r["branch_id"] for r in rows}
        assert "audit-mode-distribution" in ids
        assert "falsification-state" in ids
    finally:
        conn.close()


def test_store_snapshot_idempotent(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
            r1 = store_snapshot(conn, ms)
            r2 = store_snapshot(conn, ms)
        assert r1 == r2
        n = conn.execute(
            "SELECT COUNT(*) FROM audit_events "
            "WHERE event_type='memory_snapshot_landed' AND subject_root = ?",
            (r1,),
        ).fetchone()[0]
        assert n == 1
    finally:
        conn.close()


def test_latest_returns_most_recent(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms1 = snapshot(conn)
            r1 = store_snapshot(conn, ms1, ts=1700000000)
            # Force a different snapshot via an audit event then re-snap.
            append_audit(
                conn,
                event_type="providence_write",
                subject_root="00" * 32,
                body={"audit_mode": "HYBRID", "violations": ["X"]},
            )
            ms2 = snapshot(conn)
            r2 = store_snapshot(conn, ms2, ts=1700000100)
        live = latest(conn)
        assert live["memory_root"] == r2
        assert r1 != r2
    finally:
        conn.close()


# --- falsify / mark_stale ---------------------------------------------


def test_falsify_flips_state(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
            root = store_snapshot(conn, ms)
            event_hash = falsify(
                conn,
                root,
                reason="motif drift",
                triggering_branch_id="failure-motif:violations",
            )
        loaded = load(conn, root)
        assert loaded["state"] == "falsified"
        assert event_hash != ""
    finally:
        conn.close()


def test_falsify_idempotent(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
            root = store_snapshot(conn, ms)
            falsify(conn, root, reason="r1")
            second = falsify(conn, root, reason="r2")
        assert second == ""
    finally:
        conn.close()


def test_mark_stale_terminal_under_falsified(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
            root = store_snapshot(conn, ms)
            falsify(conn, root, reason="hard")
            second = mark_stale(conn, root, reason="soft")
        assert second == ""
        assert load(conn, root)["state"] == "falsified"
    finally:
        conn.close()


# --- projection determinism -------------------------------------------


def test_projection_violations_deterministic(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            for tag in ["TITLE_MISMATCH", "TITLE_MISMATCH", "WARRANT_MISSING"]:
                append_audit(
                    conn,
                    event_type="providence_write",
                    subject_root=None,
                    body={"audit_mode": "HYBRID", "violations": [tag]},
                )
        s1, _ = project_failure_motif_violations(conn)
        s2, _ = project_failure_motif_violations(conn)
        assert s1.summary_blob == s2.summary_blob
        assert s1.count == s2.count == 3
        # Decoded body has expected counts.
        import json

        body = json.loads(s1.summary_blob.decode("utf-8"))
        assert body["projection_version"] == PROJECTION_VERSION
        assert body["tag_counts"]["TITLE_MISMATCH"] == 2
        assert body["tag_counts"]["WARRANT_MISSING"] == 1
    finally:
        conn.close()


# --- audit chain ------------------------------------------------------


def test_audit_chain_stays_clean(tmp_path):
    db = tmp_path / "shard.db"
    conn = connect(db)
    try:
        with transaction(conn):
            ms = snapshot(conn)
            root = store_snapshot(conn, ms)
            falsify(conn, root, reason="r")
        rows = conn.execute(
            "SELECT event_hash, prev_event_hash, body "
            "FROM audit_events ORDER BY seq"
        ).fetchall()
        prev = None
        for r in rows:
            h = hashlib.sha256()
            if prev is not None:
                h.update(bytes.fromhex(prev))
            h.update(r["body"].encode("utf-8", errors="surrogatepass"))
            assert h.hexdigest() == r["event_hash"]
            prev = r["event_hash"]
        assert prev == latest_event_hash(conn)
    finally:
        conn.close()