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