arborist/tests/test_session_integration.py

"""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
    # All five 5R sub-batteries (Phase 2 of #000021).
    for sub in ("react", "rearrange", "restore", "replicate", "resonate"):
        assert ("5r", 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 complete Dav1DPrometheus suite executes 662
    deterministic tasks across 21 sub-batteries (5S+5T+5F+5R).

    History:
    - Phase 1a baseline: 462 tasks.
    - Phase 1c (#000025, 2026-05-09): 5F synthetic side expanded
      10 → 30; +100 → 562.
    - Phase 1d (#000025, 2026-05-09): 5F synthetic 30 → 50; +100 → 662.
      (Live side ALSO went 30 → 50 but lives in *-live-v1 files
      that the default-fixture-set doesn't load — those run via
      the dedicated Makefile targets.)
    """
    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 == 662


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