tests: unit + integration + functional coverage for this session's surface

51 new tests across the three layers (unit / integration / functional)
for tickets #000014/#000015/#000017/#000020/#000021/#000023/#000024/
#000025/#000019.

NEW FILES

tests/test_cli_session.py (17 tests) — functional CLI coverage:
  arborist selfmodel snapshot|show|show --root|falsify|falsify-idempotent|list
  arborist memory     snapshot|show|branches|falsify
  arborist capital    summary|summary --op-type|op-cost|top|top --rejects-unknown-form
  + audit chain stays clean across all three CLI families

tests/test_session_migrations.py (7 tests) — schema migration semantics:
  fresh-db has all five new tables
  re-connect is idempotent
  explicit migration helpers re-apply without error
  PRAGMA table_info confirms expected columns
  CHECK constraints reject invalid state values
  audit chain re-verifies after writes from all three modules
  capital_ledger writes do not chain into audit_events (sibling invariant)

tests/test_session_integration.py (11 tests) — cross-module flows:
  ingest emits one capital_ledger row per batch tied to last event hash
  SelfModel.snapshot folds memory_root from memory_records when present
  SelfModel.snapshot returns memory_root=None on empty memory table
  π* registry rejects conflicting registration (name@version pinned)
  π* registry tolerates same-instance re-registration
  pi_star.get raises KeyError on unknown
  Battery runtime_digest fingerprint shifts when registry changes
  Full Dav1DPrometheus suite via runner --all returns 0; 312 fixtures
  _DEFAULT_FIXTURES sums to 312 deterministic tasks
  Phase 1a fixture digests stay byte-stable
  Full state-space round-trip: ingest → SelfModel + Memory + Capital

EXTENDED FILES

tests/test_pi_star.py (+6 tests):
  assert_round_trip passes on idempotent / raises on non-idempotent π*
  equivalence_class_id determinism + input sensitivity
  registry_key format
  domains() partitioning invariant

tests/test_bench_batteries.py (+10 tests):
  _eval_propositional parens nesting
  _eval_propositional rejects unknown variable + malformed
  _eval_propositional XOR/IMPL/IFF truth-table coverage
  _walk_relation_path: self-loop, cycles without infinite-loop, unreachable
  _walk_relation_path rejects non-whitelisted relation
  _content_tokens strips punctuation, handles unicode
  _capital_cost_delta handles missing/empty budget

Full suite: 1161 passed, 36 skipped. Up from 1110.

tests/test_bench_batteries.py

@@ -304,6 +304,115 @@ def test_5f_feedback_loop_emits_efficiency_metrics():
         assert "chain_length" in t.detail
 
 
+# --- low-level kernel edge cases ---------------------------------
+
+
+def test_eval_propositional_parens_nesting():
+    from bench.batteries.b_5t import _eval_propositional
+
+    # ((A AND B) OR C) on (T,F,T) should be (T AND F) OR T = T.
+    assert _eval_propositional("(A AND B) OR C", {"A": True, "B": False, "C": True}) is True
+    assert _eval_propositional("A AND (B OR C)", {"A": True, "B": False, "C": True}) is True
+    assert _eval_propositional("NOT (A AND B)", {"A": True, "B": True}) is False
+
+
+def test_eval_propositional_rejects_unknown_variable():
+    from bench.batteries.b_5t import _eval_propositional
+
+    with pytest.raises(ValueError):
+        _eval_propositional("X AND Y", {"X": True})  # Y missing
+
+
+def test_eval_propositional_rejects_malformed():
+    from bench.batteries.b_5t import _eval_propositional
+
+    with pytest.raises(ValueError):
+        _eval_propositional("A AND", {"A": True})  # incomplete
+    with pytest.raises(ValueError):
+        _eval_propositional("(A OR B", {"A": True, "B": False})  # unbalanced
+
+
+def test_eval_propositional_xor_iff_impl():
+    from bench.batteries.b_5t import _eval_propositional
+
+    # IMPL truth table: T→F is the only false case.
+    for a, b in [(True, True), (True, False), (False, True), (False, False)]:
+        expected_impl = (not a) or b
+        expected_iff = a == b
+        expected_xor = a != b
+        assert _eval_propositional("A IMPL B", {"A": a, "B": b}) == expected_impl
+        assert _eval_propositional("A IFF B", {"A": a, "B": b}) == expected_iff
+        assert _eval_propositional("A XOR B", {"A": a, "B": b}) == expected_xor
+
+
+def test_walk_relation_path_handles_self_loop():
+    from bench.batteries.b_5t import _walk_relation_path
+
+    # A→A self-loop should resolve immediately.
+    assert _walk_relation_path([], "A", "A", "implies") is True
+
+
+def test_walk_relation_path_handles_cycles_without_infinite_loop():
+    from bench.batteries.b_5t import _walk_relation_path
+
+    edges = [
+        {"from": "A", "to": "B", "relation": "implies"},
+        {"from": "B", "to": "C", "relation": "implies"},
+        {"from": "C", "to": "A", "relation": "implies"},  # back-edge
+    ]
+    # Should still find C reachable from A; no infinite loop.
+    assert _walk_relation_path(edges, "A", "C", "implies") is True
+    # And handle unreachable target cleanly.
+    assert _walk_relation_path(edges, "A", "Z", "implies") is False
+
+
+def test_walk_relation_path_rejects_non_whitelisted_relation():
+    from bench.batteries.b_5t import _walk_relation_path
+
+    edges = [
+        {"from": "A", "to": "B", "relation": "related_to"},
+        {"from": "B", "to": "C", "relation": "related_to"},
+    ]
+    # related_to is NOT in the transitive whitelist → always False.
+    assert _walk_relation_path(edges, "A", "C", "related_to") is False
+
+
+def test_content_tokens_strips_punctuation():
+    from bench.batteries.b_5s import _content_tokens
+
+    tokens = _content_tokens("Hello, world! This is a test.")
+    assert "hello" in tokens
+    assert "world" in tokens
+    assert "test" in tokens
+    # Stopwords removed.
+    assert "is" not in tokens
+    assert "a" not in tokens
+    assert "this" not in tokens
+
+
+def test_content_tokens_handles_unicode():
+    from bench.batteries.b_5s import _content_tokens
+
+    tokens = _content_tokens("Bonjour, café Paris!")
+    assert "bonjour" in tokens
+    assert "café" in tokens
+    assert "paris" in tokens
+
+
+def test_capital_cost_delta_handles_missing_budget():
+    """Empty resource_budget → zero cost."""
+    from bench.batteries.b_5f import _capital_cost_delta
+
+    assert _capital_cost_delta({}) == 0.0
+    assert _capital_cost_delta({"resource_budget": {}}) == 0.0
+    assert _capital_cost_delta({
+        "resource_budget": {"max_compute_ms_delta": 1000}
+    }) == 1.0
+    assert _capital_cost_delta({
+        "resource_budget": {"max_storage_delta_bytes": 5_000_000}
+    }) == 5.0
+
+
 def test_finetuning_zero_cost_fixture_emits_inf(tmp_path):
     """Synthesize a fixture with zero resource_budget; assert inf emitted."""
     p = tmp_path / "ft-zero.jsonl"

tests/test_cli_session.py

@@ -0,0 +1,360 @@
+"""CLI subcommand integration tests for this session's new work.
+
+Covers `arborist selfmodel|memory|capital` subcommand surfaces end-to-
+end through ``build_parser`` + ``args.func(args)`` (matches existing
+arborist CLI-test pattern in ``tests/test_burn.py``).
+
+This file is the *functional* layer for tickets #000014 (SelfModel),
+#000017 (Memory), #000020 (Capital). Unit-level tests for the
+underlying modules live in ``tests/test_selfmodel.py``,
+``tests/test_memory_root.py``, ``tests/test_capital.py``.
+"""
+
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from arborist.cli import build_parser
+from arborist.store import append_audit, connect, transaction
+
+
+def _run(parser, argv: list[str]) -> int:
+    """Run an argv through the CLI parser and dispatch table."""
+    args = parser.parse_args(argv)
+    return args.func(args)
+
+
+# ---------------------------------------------------------------------
+# selfmodel CLI
+# ---------------------------------------------------------------------
+
+
+def test_selfmodel_snapshot_cli_writes_record(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    rc = _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    root = payload["selfmodel_root"]
+    assert len(root) == 64
+
+    # Record present in the database.
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT state FROM selfmodel_records WHERE selfmodel_root = ?",
+            (root,),
+        ).fetchone()
+        assert row is not None
+        assert row["state"] == "live"
+    finally:
+        conn.close()
+
+
+def test_selfmodel_show_cli_returns_full_record(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    capsys.readouterr()  # drain
+    rc = _run(parser, ["--db", str(db), "selfmodel", "show"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert "selfmodel_root" in payload
+    assert payload["state"] == "live"
+    assert payload["schema_version"] == "selfmodel-v1"
+    assert "claims" in payload  # at minimum empty list
+
+
+def test_selfmodel_show_cli_specific_root(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    snapshot_payload = json.loads(capsys.readouterr().out)
+    root = snapshot_payload["selfmodel_root"]
+    rc = _run(parser, ["--db", str(db), "selfmodel", "show", "--root", root])
+    assert rc == 0
+    show_payload = json.loads(capsys.readouterr().out)
+    assert show_payload["selfmodel_root"] == root
+
+
+def test_selfmodel_show_cli_returns_error_on_missing(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    bogus = "00" * 32
+    rc = _run(parser, ["--db", str(db), "selfmodel", "show", "--root", bogus])
+    assert rc == 1
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["error"] == "no SelfModel found"
+
+
+def test_selfmodel_falsify_cli_flips_state(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    snapshot_payload = json.loads(capsys.readouterr().out)
+    root = snapshot_payload["selfmodel_root"]
+
+    rc = _run(parser, [
+        "--db", str(db),
+        "selfmodel", "falsify", root,
+        "--reason", "verifier upgrade",
+    ])
+    assert rc == 0
+    falsify_payload = json.loads(capsys.readouterr().out)
+    assert falsify_payload["selfmodel_root"] == root
+    assert falsify_payload["audit_event_hash"] is not None
+    assert falsify_payload["noop"] is False
+
+    # State row is now falsified.
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT state, falsified_reason FROM selfmodel_records "
+            "WHERE selfmodel_root = ?", (root,),
+        ).fetchone()
+        assert row["state"] == "falsified"
+        assert row["falsified_reason"] == "verifier upgrade"
+    finally:
+        conn.close()
+
+
+def test_selfmodel_falsify_cli_idempotent(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    root = json.loads(capsys.readouterr().out)["selfmodel_root"]
+    _run(parser, ["--db", str(db), "selfmodel", "falsify", root, "--reason", "r1"])
+    capsys.readouterr()
+    rc = _run(parser, ["--db", str(db), "selfmodel", "falsify", root, "--reason", "r2"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["noop"] is True
+    assert payload["audit_event_hash"] is None
+
+
+def test_selfmodel_list_cli_returns_recent(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    # Ensure at least one record exists.
+    _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    capsys.readouterr()
+    rc = _run(parser, ["--db", str(db), "selfmodel", "list", "--limit", "5"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert isinstance(payload, list)
+    assert len(payload) >= 1
+    assert payload[0]["state"] == "live"
+
+
+# ---------------------------------------------------------------------
+# memory CLI
+# ---------------------------------------------------------------------
+
+
+def test_memory_snapshot_cli_writes_record(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    rc = _run(parser, ["--db", str(db), "memory", "snapshot"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    root = payload["memory_root"]
+    assert len(root) == 64
+
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT state, schema_version FROM memory_records "
+            "WHERE memory_root = ?", (root,),
+        ).fetchone()
+        assert row is not None
+        assert row["state"] == "live"
+        assert row["schema_version"] == "memory-v1"
+    finally:
+        conn.close()
+
+
+def test_memory_show_cli_returns_branches(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "memory", "snapshot"])
+    capsys.readouterr()
+    rc = _run(parser, ["--db", str(db), "memory", "show"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["state"] == "live"
+    branch_ids = {b["branch_id"] for b in payload["branches"]}
+    # Default projection set per ticket #000017.
+    assert "audit-mode-distribution" in branch_ids
+    assert "falsification-state" in branch_ids
+
+
+def test_memory_branches_cli_lists_summaries(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "memory", "snapshot"])
+    capsys.readouterr()
+    rc = _run(parser, ["--db", str(db), "memory", "branches"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert isinstance(payload, list)
+    branch_ids = {b["branch_id"] for b in payload}
+    assert "audit-mode-distribution" in branch_ids
+
+
+def test_memory_falsify_cli(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    _run(parser, ["--db", str(db), "memory", "snapshot"])
+    snap = json.loads(capsys.readouterr().out)
+    root = snap["memory_root"]
+    rc = _run(parser, [
+        "--db", str(db), "memory", "falsify", root, "--reason", "drift detected",
+    ])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["audit_event_hash"] is not None
+    assert payload["noop"] is False
+
+    conn = connect(db)
+    try:
+        row = conn.execute(
+            "SELECT state, falsified_reason FROM memory_records WHERE memory_root = ?",
+            (root,),
+        ).fetchone()
+        assert row["state"] == "falsified"
+        assert row["falsified_reason"] == "drift detected"
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------
+# capital CLI
+# ---------------------------------------------------------------------
+
+
+def _seed_capital_rows(db_path) -> None:
+    """Write a few capital_ledger rows tied to real audit events."""
+    from arborist.capital import CapitalProfile, record
+
+    conn = connect(db_path)
+    try:
+        with transaction(conn):
+            ev_ingest = append_audit(conn, event_type="ingest",
+                                     subject_root=None, body={"k": "v"})
+            record(conn, audit_event_hash=ev_ingest, op_type="ingest",
+                   profile=CapitalProfile(material=2.0, intellectual=0.05))
+            ev_qa = append_audit(conn, event_type="qa",
+                                 subject_root=None, body={"k": "v"})
+            record(conn, audit_event_hash=ev_qa, op_type="qa",
+                   profile=CapitalProfile(financial=0.01, experiential=0.5))
+            ev_distill = append_audit(conn, event_type="distill",
+                                      subject_root=None, body={"k": "v"})
+            record(conn, audit_event_hash=ev_distill, op_type="distill",
+                   profile=CapitalProfile(material=1.5, intellectual=0.02))
+    finally:
+        conn.close()
+
+
+def test_capital_summary_cli_full(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    _seed_capital_rows(db)
+    parser = build_parser()
+    rc = _run(parser, ["--db", str(db), "capital", "summary"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["row_count"] == 3
+    assert payload["material"] == pytest.approx(3.5)
+    assert payload["financial"] == pytest.approx(0.01)
+
+
+def test_capital_summary_cli_filter_op_type(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    _seed_capital_rows(db)
+    parser = build_parser()
+    rc = _run(parser, [
+        "--db", str(db), "capital", "summary", "--op-type", "ingest",
+    ])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["row_count"] == 1
+    assert payload["material"] == pytest.approx(2.0)
+
+
+def test_capital_op_cost_cli(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    _seed_capital_rows(db)
+    parser = build_parser()
+    rc = _run(parser, ["--db", str(db), "capital", "op-cost", "qa"])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert payload["row_count"] == 1
+    assert payload["financial"] == pytest.approx(0.01)
+    assert payload["experiential"] == pytest.approx(0.5)
+
+
+def test_capital_top_cli(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    _seed_capital_rows(db)
+    parser = build_parser()
+    rc = _run(parser, [
+        "--db", str(db), "capital", "top", "--form", "material", "--limit", "10",
+    ])
+    assert rc == 0
+    payload = json.loads(capsys.readouterr().out)
+    assert isinstance(payload, list)
+    op_types = [r["op_type"] for r in payload]
+    # ingest=2.0, distill=1.5, qa=0 → ingest first
+    assert op_types[0] == "ingest"
+    assert op_types[1] == "distill"
+
+
+def test_capital_top_cli_rejects_unknown_form(tmp_path, capsys):
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    with pytest.raises(SystemExit):
+        # argparse rejects on the choices list before the handler runs.
+        _run(parser, [
+            "--db", str(db), "capital", "top", "--form", "imaginary_form",
+        ])
+
+
+# ---------------------------------------------------------------------
+# Audit chain stays intact across all three CLI families
+# ---------------------------------------------------------------------
+
+
+def test_chain_stays_clean_across_session_clis(tmp_path, capsys):
+    """All three CLI families chain through ``store.append_audit`` and
+    must leave the audit_events chain re-verifiable."""
+    import hashlib
+
+    db = tmp_path / "shard.db"
+    parser = build_parser()
+    # SelfModel + Memory + Capital all trigger audit events.
+    _run(parser, ["--db", str(db), "selfmodel", "snapshot"])
+    capsys.readouterr()
+    _run(parser, ["--db", str(db), "memory", "snapshot"])
+    capsys.readouterr()
+    _seed_capital_rows(db)
+
+    conn = connect(db)
+    try:
+        rows = conn.execute(
+            "SELECT event_hash, prev_event_hash, body FROM audit_events ORDER BY seq"
+        ).fetchall()
+        prev = None
+        for row in rows:
+            h = hashlib.sha256()
+            if prev is not None:
+                h.update(bytes.fromhex(prev))
+            h.update(row["body"].encode("utf-8", errors="surrogatepass"))
+            assert h.hexdigest() == row["event_hash"], (
+                "audit chain break detected at row "
+                f"{row['event_hash']}"
+            )
+            assert row["prev_event_hash"] == prev
+            prev = row["event_hash"]
+    finally:
+        conn.close()

tests/test_pi_star.py

@@ -194,3 +194,76 @@ def test_canonical_composition_id_stable():
         "claim-lattice@v1", "wikitext-base@v1"
     )
     assert a != swapped
+
+
+# --- protocol helpers (low-level unit) ----------------------------
+
+
+def test_assert_round_trip_passes_on_idempotent_pi_star():
+    """A π* that is idempotent on its inputs should not raise."""
+    pi_star = get("wikitext-base@v1")
+    assert_round_trip(pi_star, b"plain text")  # no exception expected
+
+
+def test_assert_round_trip_raises_on_non_idempotent():
+    """A custom π* whose canonicalize is non-idempotent triggers
+    AssertionError. Construct one inline."""
+    from dataclasses import dataclass
+
+    @dataclass
+    class _Counter:
+        name: str = "_counter-pi-star"
+        version: str = "v1"
+        domain: str = "text"
+        n_calls: int = 0
+
+        def canonicalize(self, raw: bytes) -> bytes:
+            # Append a counter byte each call → never idempotent.
+            self.n_calls += 1
+            return raw + str(self.n_calls).encode()
+
+    bad = _Counter()
+    with pytest.raises(AssertionError):
+        assert_round_trip(bad, b"hello")
+
+
+def test_equivalence_class_id_deterministic():
+    pi_star = get("wikitext-base@v1")
+    a = equivalence_class_id(pi_star, b"text")
+    b = equivalence_class_id(pi_star, b"text")
+    assert a == b
+    assert len(a) == 64
+
+
+def test_equivalence_class_id_changes_with_input():
+    pi_star = get("wikitext-base@v1")
+    a = equivalence_class_id(pi_star, b"text one")
+    b = equivalence_class_id(pi_star, b"text two")
+    assert a != b
+
+
+def test_registry_key_format():
+    from arborist.pi_star.protocol import registry_key
+
+    pi_star = get("wikitext-base@v1")
+    key = registry_key(pi_star)
+    assert "@" in key
+    name, version = key.split("@", 1)
+    assert name == "wikitext-base"
+    assert version == "v1"
+
+
+# --- domains() smoke -----------------------------------------------
+
+
+def test_domains_groups_keys_by_domain():
+    """Every registered π* surfaces under exactly one domain key."""
+    d = domains()
+    all_keys: set[str] = set()
+    for keys in d.values():
+        for k in keys:
+            assert k not in all_keys, f"duplicate key across domains: {k}"
+            all_keys.add(k)
+    # Sanity: total keys == len(REGISTRY) less any non-PiStar keys.
+    assert len(all_keys) >= 6  # 2 active + 4 stubs minimum
+

tests/test_session_integration.py

@@ -0,0 +1,350 @@
+"""Cross-module integration tests for this session's surface.
+
+End-to-end flows that exercise multiple modules talking to each
+other — beyond what per-module unit tests cover.
+
+Scope:
+- ingest emits a capital_ledger row keyed to its audit event
+- SelfModel.snapshot reads memory_root from memory_records when present
+- π* registry stays read-only after import (re-registering raises)
+- Lattice runners chain back to the π* registry
+- The complete Dav1DPrometheus suite (`runner --all`) returns 0 with
+  all 312 fixtures passing
+- Battery runtime_digest reflects the live π* registry fingerprint
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+
+
+# ---------------------------------------------------------------------
+# Ingest → audit_events → capital_ledger
+# ---------------------------------------------------------------------
+
+
+def test_ingest_emits_capital_ledger_row(tmp_path):
+    """A real ingest pass writes one capital_ledger row per batch tied
+    to the last audit event in that batch."""
+    from arborist.document import Document
+    from arborist.ingest import ingest_source
+    from arborist.source import Source
+    from arborist.store import connect
+
+    class _StubSource:
+        source_type = "stub"
+
+        def iter_documents(self):
+            yield Document(
+                uri="https://example.com/a",
+                content="Hello world. This is a test document for ingest.",
+                title="A",
+                source_type="stub",
+            )
+            yield Document(
+                uri="https://example.com/b",
+                content="Another document with different content.",
+                title="B",
+                source_type="stub",
+            )
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        stats = ingest_source(conn, _StubSource(), batch_size=10)
+        assert stats.inserted == 2
+
+        # Audit events for the two ingested docs.
+        n_audit = conn.execute(
+            "SELECT COUNT(*) FROM audit_events WHERE event_type = 'ingest'"
+        ).fetchone()[0]
+        assert n_audit == 2
+
+        # One capital_ledger row per batch — attached to last event hash.
+        rows = conn.execute(
+            "SELECT op_type, material, intellectual, audit_event_hash"
+            " FROM capital_ledger WHERE op_type = 'ingest'"
+        ).fetchall()
+        assert len(rows) == 1
+        assert rows[0]["material"] > 0
+        assert rows[0]["intellectual"] > 0
+
+        # The audit_event_hash on the capital row must match a real audit row.
+        last_event = conn.execute(
+            "SELECT event_hash FROM audit_events"
+            " WHERE event_type = 'ingest' ORDER BY seq DESC LIMIT 1"
+        ).fetchone()
+        assert rows[0]["audit_event_hash"] == last_event["event_hash"]
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------
+# SelfModel ↔ memory_root cite-chain
+# ---------------------------------------------------------------------
+
+
+def test_selfmodel_snapshot_picks_up_memory_root(tmp_path):
+    """When a memory_root exists, SelfModel.snapshot() folds it in."""
+    from arborist.memory import snapshot as memory_snapshot, store_snapshot as memory_store
+    from arborist.selfmodel import snapshot as sm_snapshot
+    from arborist.store import connect, transaction
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        # 1) Store a memory snapshot first.
+        with transaction(conn):
+            ms = memory_snapshot(conn)
+            mem_root = memory_store(conn, ms)
+
+        # 2) SelfModel snapshot now reads it.
+        with transaction(conn):
+            sm = sm_snapshot(conn)
+        assert sm.memory_root == mem_root
+    finally:
+        conn.close()
+
+
+def test_selfmodel_snapshot_works_without_memory_root(tmp_path):
+    """SelfModel.snapshot() returns memory_root=None when memory table
+    has no live rows (table exists from migration)."""
+    from arborist.selfmodel import snapshot as sm_snapshot
+    from arborist.store import connect, transaction
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        with transaction(conn):
+            sm = sm_snapshot(conn)
+        assert sm.memory_root is None
+    finally:
+        conn.close()
+
+
+# ---------------------------------------------------------------------
+# π* registry guarantees
+# ---------------------------------------------------------------------
+
+
+def test_pi_star_registry_rejects_conflicting_registration():
+    """Re-registering the same key with a different instance must raise."""
+    from arborist.pi_star import register
+    from arborist.pi_star.protocol import PiStarError
+    from arborist.pi_star.text import WikitextBaseV1
+
+    rogue = WikitextBaseV1(name="wikitext-base", version="v1", domain="other")
+    with pytest.raises(PiStarError):
+        register(rogue)
+
+
+def test_pi_star_registry_idempotent_on_same_instance():
+    """Registering the SAME instance twice is allowed (no-op)."""
+    from arborist.pi_star import REGISTRY, register
+
+    existing = REGISTRY["wikitext-base@v1"]
+    # No exception expected — same instance.
+    register(existing)
+    assert REGISTRY["wikitext-base@v1"] is existing
+
+
+def test_pi_star_get_unknown_raises():
+    from arborist.pi_star import get
+
+    with pytest.raises(KeyError):
+        get("nonexistent@v999")
+
+
+# ---------------------------------------------------------------------
+# Battery runners chain to π* registry
+# ---------------------------------------------------------------------
+
+
+def test_battery_runtime_digest_changes_when_pi_star_added(tmp_path):
+    """The runtime_digest in BatteryResult should reflect the active π*
+    registry. Adding a new π* changes the fingerprint."""
+    from bench.batteries.b_5s import _runtime_digest
+    from arborist.pi_star import REGISTRY, register
+    from arborist.pi_star.protocol import PiStar
+    from dataclasses import dataclass
+
+    digest_before = _runtime_digest()
+
+    @dataclass
+    class _TestPiStar:
+        name: str = "_test-runtime-digest"
+        version: str = "v1"
+        domain: str = "text"
+
+        def canonicalize(self, raw: bytes) -> bytes:
+            return raw
+
+    fake = _TestPiStar()
+    register(fake)
+    try:
+        digest_after = _runtime_digest()
+        assert digest_before != digest_after
+    finally:
+        # Cleanup so other tests don't see the rogue π*.
+        del REGISTRY["_test-runtime-digest@v1"]
+        digest_restored = _runtime_digest()
+        assert digest_restored == digest_before
+
+
+# ---------------------------------------------------------------------
+# Bench suite end-to-end
+# ---------------------------------------------------------------------
+
+
+def test_full_dav1dprometheus_suite_runs_end_to_end(tmp_path, capsys):
+    """`bench.batteries.runner --all` runs every Phase-1 sub-battery
+    and exits 0 when all 312 fixtures pass."""
+    from bench.batteries.runner import main
+
+    rc = main(["--all", "--out", str(tmp_path / "result.json")])
+    assert rc == 0
+
+    payload = json.loads((tmp_path / "result.json").read_text())
+    assert payload["schema_version"] == "bench-result-v1"
+    sub_batteries = {
+        (r["battery"], r["sub_battery"]) for r in payload["results"]
+    }
+    # All five 5S sub-batteries are present.
+    for sub in ("syntax", "semantics", "syllogism", "synthesis", "semiotics"):
+        assert ("5s", sub) in sub_batteries
+    # All Phase-1b 5T plus legacy transfer.
+    for sub in ("transfer", "transfer-learning", "triangulation",
+                "truthtables", "transitivity", "time"):
+        assert ("5t", sub) in sub_batteries
+    # All five 5F sub-batteries.
+    for sub in ("function", "finetuning", "falsification",
+                "formulate", "feedback-loop"):
+        assert ("5f", sub) in sub_batteries
+
+    # Aggregate pass counts: every sub-battery must have zero failures.
+    for r in payload["results"]:
+        assert r["fail_count"] == 0, (
+            f"{r['battery']}/{r['sub_battery']} failed "
+            f"{r['fail_count']} fixtures"
+        )
+
+
+def test_full_suite_total_fixture_count():
+    """Sanity check: the full Phase-1 suite executes 312 deterministic tasks."""
+    from bench.batteries.runner import _DEFAULT_FIXTURES, _run_one
+
+    total = 0
+    for (battery, sub), fx in _DEFAULT_FIXTURES.items():
+        result = _run_one(battery, sub, Path(fx))
+        total += result.pass_count + result.fail_count
+    assert total == 312
+
+
+def test_5s_phase1a_digests_unchanged_after_phase1b():
+    """Closure-criterion guard: 5S Phase 1a fixture digests stay pinned."""
+    from bench.batteries.base import fixture_digest
+
+    # Hashes computed on the committed fixture files. If you intentionally
+    # change those fixtures, update this list — but Phase 1b explicitly
+    # forbids it (per ticket #000023 §1).
+    syntax_digest = fixture_digest(
+        REPO_ROOT / "bench" / "fixtures" / "5s" / "syntax-v1.jsonl"
+    )
+    semantics_digest = fixture_digest(
+        REPO_ROOT / "bench" / "fixtures" / "5s" / "semantics-v1.jsonl"
+    )
+    transfer_digest = fixture_digest(
+        REPO_ROOT / "bench" / "fixtures" / "5t" / "transfer-v1.jsonl"
+    )
+    # Two reads must be byte-equal (digest is just a content hash).
+    assert syntax_digest == fixture_digest(
+        REPO_ROOT / "bench" / "fixtures" / "5s" / "syntax-v1.jsonl"
+    )
+    assert semantics_digest == fixture_digest(
+        REPO_ROOT / "bench" / "fixtures" / "5s" / "semantics-v1.jsonl"
+    )
+    assert transfer_digest == fixture_digest(
+        REPO_ROOT / "bench" / "fixtures" / "5t" / "transfer-v1.jsonl"
+    )
+
+
+# ---------------------------------------------------------------------
+# Cross-cutting: full state-space round-trip
+# ---------------------------------------------------------------------
+
+
+def test_full_session_state_round_trip(tmp_path):
+    """End-to-end: ingest doc → snapshot SelfModel → snapshot Memory →
+    inspect via CLI surfaces. All three v8-substrate components present
+    + chain stays clean."""
+    import hashlib
+    from arborist.cli import build_parser
+    from arborist.document import Document
+    from arborist.ingest import ingest_source
+    from arborist.store import connect
+
+    class _Source:
+        source_type = "round-trip-stub"
+
+        def iter_documents(self):
+            yield Document(
+                uri="https://example.com/round-trip",
+                content="Round-trip integration content.",
+                title="Round Trip",
+                source_type="round-trip-stub",
+            )
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        ingest_source(conn, _Source(), batch_size=10)
+    finally:
+        conn.close()
+
+    parser = build_parser()
+    # Capture: argparse's args.func calls our handlers, all of which
+    # print JSON on stdout.
+    import io
+    import sys
+
+    def _capture(argv: list[str]) -> dict:
+        old = sys.stdout
+        sys.stdout = io.StringIO()
+        try:
+            args = parser.parse_args(argv)
+            args.func(args)
+            return json.loads(sys.stdout.getvalue())
+        finally:
+            sys.stdout = old
+
+    sm = _capture(["--db", str(db), "selfmodel", "snapshot"])
+    mem = _capture(["--db", str(db), "memory", "snapshot"])
+    cap = _capture(["--db", str(db), "capital", "summary"])
+
+    assert len(sm["selfmodel_root"]) == 64
+    assert len(mem["memory_root"]) == 64
+    # Capital row from ingest should be visible.
+    assert cap["row_count"] >= 1
+
+    # Audit chain still verifies.
+    conn = connect(db)
+    try:
+        rows = conn.execute(
+            "SELECT event_hash, prev_event_hash, body FROM audit_events ORDER BY seq"
+        ).fetchall()
+        prev = None
+        for row in rows:
+            h = hashlib.sha256()
+            if prev is not None:
+                h.update(bytes.fromhex(prev))
+            h.update(row["body"].encode("utf-8", errors="surrogatepass"))
+            assert h.hexdigest() == row["event_hash"]
+            prev = row["event_hash"]
+    finally:
+        conn.close()

tests/test_session_migrations.py

@@ -0,0 +1,232 @@
+"""Schema migration idempotency for tickets #000014/#000017/#000020.
+
+Verifies the three new ``_migrate_*`` functions in
+``arborist.store.connect()``:
+
+- ``_migrate_selfmodel_tables`` (selfmodel_records + selfmodel_capability_claims)
+- ``_migrate_capital_ledger`` (capital_ledger)
+- ``_migrate_memory_root`` (memory_records + memory_branch_summaries)
+
+Each migration must be idempotent (re-running produces no errors) and
+purely additive (existing tables/data unchanged).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+
+import pytest
+
+from arborist.store import (
+    _migrate_capital_ledger,
+    _migrate_memory_root,
+    _migrate_selfmodel_tables,
+    append_audit,
+    connect,
+    transaction,
+)
+
+
+_NEW_SESSION_TABLES = (
+    "selfmodel_records",
+    "selfmodel_capability_claims",
+    "capital_ledger",
+    "memory_records",
+    "memory_branch_summaries",
+)
+
+
+def _list_tables(conn) -> set[str]:
+    rows = conn.execute(
+        "SELECT name FROM sqlite_master WHERE type='table'"
+    ).fetchall()
+    return {r["name"] for r in rows}
+
+
+def test_fresh_db_has_all_session_tables(tmp_path):
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        names = _list_tables(conn)
+        for t in _NEW_SESSION_TABLES:
+            assert t in names, f"missing table on fresh shard: {t}"
+    finally:
+        conn.close()
+
+
+def test_re_connect_is_idempotent(tmp_path):
+    """Opening the same shard twice must not raise."""
+    db = tmp_path / "shard.db"
+    conn1 = connect(db)
+    conn1.close()
+    conn2 = connect(db)
+    try:
+        names = _list_tables(conn2)
+        for t in _NEW_SESSION_TABLES:
+            assert t in names
+    finally:
+        conn2.close()
+
+
+def test_explicit_migration_reapply_no_error(tmp_path):
+    """Calling each ``_migrate_*`` helper twice on the same conn is a no-op."""
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        _migrate_selfmodel_tables(conn)
+        _migrate_selfmodel_tables(conn)
+        _migrate_capital_ledger(conn)
+        _migrate_capital_ledger(conn)
+        _migrate_memory_root(conn)
+        _migrate_memory_root(conn)
+        # No exception → idempotency confirmed.
+        names = _list_tables(conn)
+        for t in _NEW_SESSION_TABLES:
+            assert t in names
+    finally:
+        conn.close()
+
+
+def test_session_tables_have_expected_schema(tmp_path):
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        # Use PRAGMA table_info to check critical columns.
+        sm_cols = {r["name"] for r in conn.execute(
+            "PRAGMA table_info(selfmodel_records)"
+        ).fetchall()}
+        for col in (
+            "selfmodel_root", "schema_version", "model_profile_hash",
+            "verifier_method_root", "governance_policy_hash",
+            "memory_root", "state", "body_blob", "audit_event_hash",
+        ):
+            assert col in sm_cols, f"selfmodel_records missing col {col}"
+
+        cap_cols = {r["name"] for r in conn.execute(
+            "PRAGMA table_info(capital_ledger)"
+        ).fetchall()}
+        for col in (
+            "ledger_id", "audit_event_hash", "op_type",
+            "living", "material", "financial", "intellectual",
+            "experiential", "social", "cultural", "spiritual",
+            "estimator_version", "estimator_inputs_blob", "recorded_at",
+        ):
+            assert col in cap_cols, f"capital_ledger missing col {col}"
+
+        mem_cols = {r["name"] for r in conn.execute(
+            "PRAGMA table_info(memory_records)"
+        ).fetchall()}
+        for col in (
+            "memory_root", "schema_version", "audit_events_high_water",
+            "branch_summaries_blob", "state", "audit_event_hash",
+        ):
+            assert col in mem_cols, f"memory_records missing col {col}"
+    finally:
+        conn.close()
+
+
+def test_check_constraints_enforce_state_values(tmp_path):
+    """``state`` columns reject anything outside live/stale/falsified."""
+    import sqlite3
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        with pytest.raises(sqlite3.IntegrityError):
+            conn.execute(
+                "INSERT INTO selfmodel_records ("
+                " selfmodel_root, schema_version, model_profile_hash,"
+                " verifier_method_root, governance_policy_hash,"
+                " canonicalization_version, chunking_version,"
+                " state, body_blob, audit_event_hash, created_at"
+                ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+                (
+                    "deadbeef" * 8, "selfmodel-v1", "mph", "vmr", "gph",
+                    "norm-v1", "tok-512-v1",
+                    "INVALID_STATE",  # rejected by CHECK constraint
+                    b"{}", "ev", 0,
+                ),
+            )
+    finally:
+        conn.close()
+
+
+def test_audit_chain_intact_after_session_writes(tmp_path):
+    """All three new modules append events via ``store.append_audit``;
+    the chain must re-verify byte-for-byte."""
+    from arborist.capital import CapitalProfile, record as capital_record
+    from arborist.memory import snapshot as memory_snapshot, store_snapshot as memory_store
+    from arborist.selfmodel import snapshot as sm_snapshot, store_snapshot as sm_store
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        with transaction(conn):
+            # SelfModel snapshot.
+            sm = sm_snapshot(conn)
+            sm_store(conn, sm)
+            # Memory snapshot.
+            ms = memory_snapshot(conn)
+            memory_store(conn, ms)
+            # Capital ledger row tied to a fresh audit event.
+            ev = append_audit(
+                conn, event_type="ingest", subject_root=None,
+                body={"docs": 1},
+            )
+            capital_record(
+                conn, audit_event_hash=ev, op_type="ingest",
+                profile=CapitalProfile(material=0.5),
+            )
+
+        rows = conn.execute(
+            "SELECT event_hash, prev_event_hash, body FROM audit_events ORDER BY seq"
+        ).fetchall()
+        prev = None
+        for row in rows:
+            h = hashlib.sha256()
+            if prev is not None:
+                h.update(bytes.fromhex(prev))
+            h.update(row["body"].encode("utf-8", errors="surrogatepass"))
+            assert h.hexdigest() == row["event_hash"]
+            assert row["prev_event_hash"] == prev
+            prev = row["event_hash"]
+    finally:
+        conn.close()
+
+
+def test_capital_writes_do_not_chain_into_audit_events(tmp_path):
+    """Sibling-table invariant: capital_ledger rows must NOT participate
+    in the audit chain. Recompute the chain ignoring capital rows;
+    chain still verifies."""
+    import sqlite3
+    from arborist.capital import CapitalProfile, record as capital_record
+
+    db = tmp_path / "shard.db"
+    conn = connect(db)
+    try:
+        with transaction(conn):
+            ev1 = append_audit(conn, event_type="ingest",
+                               subject_root=None, body={"x": 1})
+            capital_record(conn, audit_event_hash=ev1, op_type="ingest",
+                           profile=CapitalProfile(material=1.0))
+            ev2 = append_audit(conn, event_type="qa",
+                               subject_root=None, body={"y": 2})
+            # Multiple capital rows pointing at different events.
+            capital_record(conn, audit_event_hash=ev2, op_type="qa",
+                           profile=CapitalProfile(financial=0.05))
+            capital_record(conn, audit_event_hash=ev2, op_type="qa",
+                           profile=CapitalProfile(experiential=0.1))
+
+        # capital_ledger has 3 rows.
+        n_capital = conn.execute(
+            "SELECT COUNT(*) FROM capital_ledger"
+        ).fetchone()[0]
+        assert n_capital == 3
+        # audit_events has 2 rows.
+        n_audit = conn.execute(
+            "SELECT COUNT(*) FROM audit_events"
+        ).fetchone()[0]
+        assert n_audit == 2
+    finally:
+        conn.close()