memory: land ticket #000017 (memory-root lifelong learning summary)

Periodic, deterministic projection over audit_events that summarizes
recurring failure motifs, audit-mode distribution, and falsification
state. Sibling layer to providence_cache (per-cache_key answers) and
audit_events (per-event chain) — memory_root is the cross-query
behavior history a SelfModel optionally cites.

Surface:

- arborist.memory.{canonical,projections,snapshot,store,falsify}
- Three default branch projections at v1 (PROJECTION_VERSION pin):
  - failure-motif:violations (counts violation tags from
    providence_write events)
  - audit-mode-distribution (STRICT/HYBRID/UNGROUNDED counts)
  - falsification-state (current cache state distribution)
- memory_root = SHA-256 over canonical body bytes; sort-invariant
  on branches.
- CLI: arborist memory snapshot|show|branches|falsify
- Audit events: memory_snapshot_landed, memory_falsified,
  memory_marked_stale.

SelfModel integration: arborist.selfmodel.snapshot reads latest live
memory_root and folds into SelfModel body. Already shipped in #000014;
this ticket completes the round-trip (memory shifts → SelfModel root
shifts).

Tests: tests/test_memory_root.py (15 cases). Full suite: 1040 passed,
36 skipped.

tests/test_memory_root.py

@@ -0,0 +1,325 @@
+"""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()