arborist/tests/test_burn_doc.py

"""Burn — delete a document or core leaf with no children.

Mirrors test_burn.py for cache_key burns. Same kindergarten rule:
  delete a leaf only if it has no children, or use --force.

A document/core has children when:
  - derivations.src_root  = root  (downstream cores derived from it)
  - edges.dst_root        = root  (other docs link to it)
  - providence_cache.source_root = root  (Q&A grounded in it)

Cores additionally never evict (per CLAUDE.md), but burn IS allowed —
burn is operator-driven, evict is the automated cold-tier compression.
"""

from __future__ import annotations

import json
import time
from typing import Iterator

from arborist.cli import _burn_core_root, _burn_document_root
from arborist.distill import FirstSentenceDistiller
from arborist.distill.runner import distill_existing
from arborist.document import Document
from arborist.ingest import ingest_source
from arborist.source import Source
from arborist.store import append_audit, connect, transaction


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


class _FakeSource(Source):
    """Minimal in-memory source so we can ingest deterministic content."""

    source_type = "html"

    def __init__(self, docs: list[Document]):
        self.docs = docs

    def iter_documents(self) -> Iterator[Document]:
        yield from self.docs


def _doc(uri: str, content: str, *, edges=None) -> Document:
    return Document(
        uri=uri,
        content=content,
        source_type="html",
        title=uri,
        edges=edges or [],
    )


# Long enough that the default tok-512-v1 chunker emits multiple chunks.
_LONG = (
    "Arborist tends trees and forests of cross-linked information. " * 40
    + "\n\n"
    + "Burn is the kindergarten leaf removal — children gate enforced. " * 40
)


def _ingest_one(db_path, uri: str, content: str = _LONG, edges=None) -> str:
    """Ingest a single document and return its document_root."""
    conn = connect(db_path)
    try:
        ingest_source(conn, _FakeSource([_doc(uri, content, edges=edges)]))
        row = conn.execute(
            "SELECT document_root FROM documents WHERE document_uri = ?",
            (uri,),
        ).fetchone()
        assert row is not None, f"ingest of {uri} did not create a document row"
        return row["document_root"]
    finally:
        conn.close()


def _seed_providence_pointing_at(db_path, *, source_root: str, cache_key: str) -> None:
    """Insert a providence_cache row whose source_root = the doc we want
    to protect, so the children gate fires."""
    conn = connect(db_path)
    try:
        with transaction(conn):
            event_hash = append_audit(
                conn,
                event_type="providence_query",
                subject_root=cache_key,
                body={"cache_key": cache_key, "source_root": source_root},
            )
            conn.execute(
                "INSERT INTO providence_cache "
                "(cache_key, source_root, document_uri, question_hash, question_text, "
                " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
                " governance_policy_hash, schema_version, canonicalization_version, "
                " chunking_version, falsification_state, chain, audit_event_hash, "
                " created_at, hit_count, audit_mode, n_quotes, n_verified, "
                " unverified_quotes, verifier_method) "
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', 'private', ?, ?, 0, ?, 1, 1, NULL, 'quote')",
                (
                    cache_key,
                    source_root,
                    "https://example.com/doc",
                    "qh",
                    "q?",
                    "answer",
                    json.dumps({}),
                    "mp",
                    "ch",
                    "gh",
                    "v9.8.0",
                    "norm-v1",
                    "tok-512-v1",
                    event_hash,
                    int(time.time()),
                    "STRICT",
                ),
            )
    finally:
        conn.close()


def _seed_incoming_edge(db_path, *, dst_root: str) -> None:
    """Insert a single edge pointing at dst_root — a child for burn's gate."""
    conn = connect(db_path)
    try:
        conn.execute(
            "INSERT OR IGNORE INTO edges (src_root, dst_root, dst_uri, edge_type, anchor) "
            "VALUES (?, ?, ?, 'wikilink', '')",
            ("ff" * 32, dst_root, ""),
        )
    finally:
        conn.close()


# ---------------------------------------------------------------------------
# Document burn — happy path
# ---------------------------------------------------------------------------


def test_burn_document_no_children_removes_row_and_chunks(tmp_path):
    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://leaf-1")

    # Sanity: chunks + merkle_nodes + fts rows exist before burn.
    conn = connect(db)
    try:
        chunks_before = conn.execute(
            "SELECT COUNT(*) FROM chunks WHERE document_root = ?", (root,)
        ).fetchone()[0]
        merkle_before = conn.execute(
            "SELECT COUNT(*) FROM merkle_nodes WHERE document_root = ?", (root,)
        ).fetchone()[0]
    finally:
        conn.close()
    assert chunks_before > 0

    result = _burn_document_root(
        root, reason="kindergarten cleanup", by_actor="alice",
        shards_dir=None, db_path=db,
    )
    assert result["status"] == "burned"
    assert result["document_root"] == root
    assert result["kind"] == "surface"
    assert result["burned_chunk_count"] == chunks_before
    assert result["child_counts_at_burn"] == {
        "derivations_downstream": 0,
        "incoming_edges": 0,
        "providence_refs": 0,
    }
    assert isinstance(result["audit_event_hash"], str) and len(result["audit_event_hash"]) == 64

    conn = connect(db)
    try:
        gone = conn.execute(
            "SELECT 1 FROM documents WHERE document_root = ?", (root,)
        ).fetchone()
        chunks_after = conn.execute(
            "SELECT COUNT(*) FROM chunks WHERE document_root = ?", (root,)
        ).fetchone()[0]
        merkle_after = conn.execute(
            "SELECT COUNT(*) FROM merkle_nodes WHERE document_root = ?", (root,)
        ).fetchone()[0]
        last = conn.execute(
            "SELECT event_type, body, subject_root FROM audit_events "
            "ORDER BY seq DESC LIMIT 1"
        ).fetchone()
    finally:
        conn.close()
    assert gone is None
    # FK CASCADE swept chunks + merkle_nodes.
    assert chunks_after == 0
    assert merkle_after == 0
    assert last["event_type"] == "document_burn"
    assert last["subject_root"] == root
    body = json.loads(last["body"])
    assert body["document_root"] == root
    assert body["kind"] == "surface"
    assert body["forced"] is False
    assert body["reason"] == "kindergarten cleanup"
    assert body["by_actor"] == "alice"


# ---------------------------------------------------------------------------
# Document burn — children gates
# ---------------------------------------------------------------------------


def test_burn_document_refuses_with_derivation_downstream(tmp_path):
    """A document with a derived core is not a leaf — refuse without --force."""
    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://has-core")
    # Distill a core from this surface.
    conn = connect(db)
    try:
        result = distill_existing(conn, FirstSentenceDistiller(), kind="surface", limit=10)
    finally:
        conn.close()
    assert result["distilled"] >= 1

    conn = connect(db)
    try:
        events_before = conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0]
    finally:
        conn.close()

    result = _burn_document_root(
        root, reason="should refuse", by_actor="alice",
        shards_dir=None, db_path=db,
    )
    assert result["status"] == "refused_has_children"
    assert result["derivations_downstream"] >= 1
    assert "hint" in result

    conn = connect(db)
    try:
        still_there = conn.execute(
            "SELECT 1 FROM documents WHERE document_root = ?", (root,)
        ).fetchone()
        events_after = conn.execute("SELECT COUNT(*) FROM audit_events").fetchone()[0]
    finally:
        conn.close()
    assert still_there is not None
    # Refusal must NOT write an audit event.
    assert events_after == events_before


def test_burn_document_refuses_with_providence_pointing_at_it(tmp_path):
    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://qa-grounded")
    _seed_providence_pointing_at(db, source_root=root, cache_key="aa" * 32)

    result = _burn_document_root(
        root, reason="should refuse", by_actor="alice",
        shards_dir=None, db_path=db,
    )
    assert result["status"] == "refused_has_children"
    assert result["providence_refs"] == 1

    conn = connect(db)
    try:
        still_there = conn.execute(
            "SELECT 1 FROM documents WHERE document_root = ?", (root,)
        ).fetchone()
    finally:
        conn.close()
    assert still_there is not None


def test_burn_document_refuses_with_incoming_edge(tmp_path):
    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://linked-to")
    _seed_incoming_edge(db, dst_root=root)

    result = _burn_document_root(
        root, reason="should refuse", by_actor="alice",
        shards_dir=None, db_path=db,
    )
    assert result["status"] == "refused_has_children"
    assert result["incoming_edges"] == 1


# ---------------------------------------------------------------------------
# Document burn — --force
# ---------------------------------------------------------------------------


def test_burn_document_force_succeeds_with_children(tmp_path):
    """--force burns past the gate; audit body records counts + forced=True."""
    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://forced")
    _seed_providence_pointing_at(db, source_root=root, cache_key="bb" * 32)
    _seed_incoming_edge(db, dst_root=root)

    result = _burn_document_root(
        root, reason="forced cleanup", by_actor="alice",
        shards_dir=None, db_path=db, force=True,
    )
    assert result["status"] == "burned"

    conn = connect(db)
    try:
        gone = conn.execute(
            "SELECT 1 FROM documents WHERE document_root = ?", (root,)
        ).fetchone()
        last = conn.execute(
            "SELECT body FROM audit_events ORDER BY seq DESC LIMIT 1"
        ).fetchone()
    finally:
        conn.close()
    assert gone is None
    body = json.loads(last["body"])
    assert body["forced"] is True
    counts = body["child_counts_at_burn"]
    assert counts["providence_refs"] == 1
    assert counts["incoming_edges"] == 1


# ---------------------------------------------------------------------------
# Core burn
# ---------------------------------------------------------------------------


def _ingest_and_distill(db_path, uri: str) -> tuple[str, str]:
    """Ingest a surface + distill a core. Return (surface_root, core_root)."""
    surface_root = _ingest_one(db_path, uri)
    conn = connect(db_path)
    try:
        result = distill_existing(conn, FirstSentenceDistiller(), kind="surface", limit=10)
        assert result["distilled"] >= 1
        core = conn.execute(
            "SELECT document_root FROM documents WHERE kind = 'core' "
            "ORDER BY ingest_ts DESC LIMIT 1"
        ).fetchone()
        assert core is not None
        return surface_root, core["document_root"]
    finally:
        conn.close()


def test_burn_core_no_children_removes_row(tmp_path):
    db = tmp_path / "burn.db"
    surface_root, core_root = _ingest_and_distill(db, "html://to-distill")

    # The core is a leaf (no further derivations from it).
    result = _burn_core_root(
        core_root, reason="prune scratch core", by_actor="alice",
        shards_dir=None, db_path=db,
    )
    assert result["status"] == "burned"
    assert result["kind"] == "core"
    # Inbound derivations (core_root = this) cascade with the core.
    assert result["burned_inbound_derivations"] >= 1

    conn = connect(db)
    try:
        gone = conn.execute(
            "SELECT 1 FROM documents WHERE document_root = ?", (core_root,)
        ).fetchone()
        # Surface is untouched — burning a core does not touch its parents.
        surface_still = conn.execute(
            "SELECT 1 FROM documents WHERE document_root = ?", (surface_root,)
        ).fetchone()
        last = conn.execute(
            "SELECT event_type FROM audit_events ORDER BY seq DESC LIMIT 1"
        ).fetchone()
    finally:
        conn.close()
    assert gone is None
    assert surface_still is not None
    assert last["event_type"] == "core_burn"


def test_burn_core_refuses_with_downstream_derivation(tmp_path):
    """A core that some deeper core was distilled FROM is not a leaf."""
    db = tmp_path / "burn.db"
    surface_root, core_root = _ingest_and_distill(db, "html://recursive")
    # Run distill again with kind='core' — this distills FROM the core,
    # producing a depth=2 core whose src_root points back at core_root.
    conn = connect(db)
    try:
        deeper = distill_existing(conn, FirstSentenceDistiller(), kind="core", limit=10)
    finally:
        conn.close()
    if deeper["distilled"] == 0:
        # Recursive distill is content-dependent; if our short core doesn't
        # produce a deeper one, simulate the relationship by inserting a
        # synthetic derivation row keyed off the core. Both paths exercise
        # the gate via derivations.src_root > 0.
        deeper_root = "cc" * 32
        conn = connect(db)
        try:
            with transaction(conn):
                conn.execute(
                    "INSERT INTO documents "
                    "(document_root, document_uri, source_type, kind, "
                    " compression_depth, title, chunking_version, "
                    " canonicalization_version, schema_version, ingest_ts) "
                    "VALUES (?, 'core://deeper', 'html', 'core', 2, 'deeper', "
                    "        'tok-512-v1', 'norm-v1', 'v9.8.0', ?)",
                    (deeper_root, int(time.time())),
                )
                conn.execute(
                    "INSERT INTO derivations "
                    "(core_root, src_root, proof_blob, process_id, distilled_at) "
                    "VALUES (?, ?, '{}', 'test', ?)",
                    (deeper_root, core_root, int(time.time())),
                )
        finally:
            conn.close()

    # Confirm there is at least one downstream derivation now.
    conn = connect(db)
    try:
        downstream = conn.execute(
            "SELECT COUNT(*) FROM derivations WHERE src_root = ?", (core_root,)
        ).fetchone()[0]
    finally:
        conn.close()
    assert downstream >= 1

    result = _burn_core_root(
        core_root, reason="should refuse", by_actor="alice",
        shards_dir=None, db_path=db,
    )
    assert result["status"] == "refused_has_children"
    assert result["derivations_downstream"] >= 1


# ---------------------------------------------------------------------------
# Audit chain integrity
# ---------------------------------------------------------------------------


def test_burn_document_preserves_audit_chain(tmp_path):
    """Same property `make chain-check` enforces — no dangling prev_event_hash."""
    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://chain-check")

    _burn_document_root(root, reason="t", by_actor="a", shards_dir=None, db_path=db)

    conn = connect(db)
    try:
        breaks = conn.execute(
            """
            SELECT COUNT(*) FROM audit_events a1
            LEFT JOIN audit_events a2 ON a2.event_hash = a1.prev_event_hash
            WHERE a1.prev_event_hash IS NOT NULL AND a2.event_hash IS NULL
            """
        ).fetchone()[0]
    finally:
        conn.close()
    assert breaks == 0


# ---------------------------------------------------------------------------
# CLI integration
# ---------------------------------------------------------------------------


def test_burn_cli_document_invocation(tmp_path, capsys):
    from arborist.cli import build_parser

    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://cli-doc")

    parser = build_parser()
    args = parser.parse_args([
        "--db", str(db),
        "burn",
        "--kind", "document",
        "--root", root,
        "--reason", "via cli",
        "--by-actor", "alice",
    ])
    rc = args.func(args)
    assert rc == 0

    payload = json.loads(capsys.readouterr().out)
    assert payload["status"] == "burned"
    assert payload["kind"] == "surface"
    assert payload["document_root"] == root


def test_burn_cli_core_invocation(tmp_path, capsys):
    from arborist.cli import build_parser

    db = tmp_path / "burn.db"
    _, core_root = _ingest_and_distill(db, "html://cli-core")

    parser = build_parser()
    args = parser.parse_args([
        "--db", str(db),
        "burn",
        "--kind", "core",
        "--root", core_root,
        "--reason", "via cli",
    ])
    rc = args.func(args)
    assert rc == 0

    payload = json.loads(capsys.readouterr().out)
    assert payload["status"] == "burned"
    assert payload["kind"] == "core"


def test_burn_cli_default_kind_is_providence(tmp_path, capsys):
    """Backwards compat: bare `burn --cache-key X` still works (no --kind)."""
    from arborist.cli import build_parser

    db = tmp_path / "burn.db"
    # Seed a providence record reusing the existing test_burn helper logic.
    KEY = "ee" * 32
    conn = connect(db)
    try:
        with transaction(conn):
            event_hash = append_audit(
                conn,
                event_type="providence_query",
                subject_root=KEY,
                body={"cache_key": KEY},
            )
            conn.execute(
                "INSERT INTO providence_cache "
                "(cache_key, source_root, document_uri, question_hash, question_text, "
                " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
                " governance_policy_hash, schema_version, canonicalization_version, "
                " chunking_version, falsification_state, chain, audit_event_hash, "
                " created_at, hit_count, audit_mode, n_quotes, n_verified, "
                " unverified_quotes, verifier_method) "
                "VALUES (?, '00', 'u', 'qh', 'q', 'a', '{}', 'm', 'c', 'g', "
                "        'v9.8.0', 'norm-v1', 'tok-512-v1', 'live', 'private', "
                "        ?, ?, 0, 'STRICT', 1, 1, NULL, 'quote')",
                (KEY, event_hash, int(time.time())),
            )
    finally:
        conn.close()

    parser = build_parser()
    args = parser.parse_args([
        "--db", str(db),
        "burn",
        "--cache-key", KEY,
        "--reason", "back-compat",
    ])
    rc = args.func(args)
    assert rc == 0
    payload = json.loads(capsys.readouterr().out)
    assert payload["status"] == "burned"
    assert payload["cache_key"] == KEY


def test_burn_cli_document_returns_non_zero_on_refused(tmp_path, capsys):
    from arborist.cli import build_parser

    db = tmp_path / "burn.db"
    root = _ingest_one(db, "html://cli-refuse")
    _seed_incoming_edge(db, dst_root=root)

    parser = build_parser()
    args = parser.parse_args([
        "--db", str(db),
        "burn",
        "--kind", "document",
        "--root", root,
        "--reason", "wont land",
    ])
    rc = args.func(args)
    assert rc == 1
    payload = json.loads(capsys.readouterr().out)
    assert payload["status"] == "refused_has_children"
    assert payload["incoming_edges"] == 1


def test_burn_unknown_root_returns_not_found(tmp_path):
    db = tmp_path / "burn.db"
    result = _burn_document_root(
        "00" * 32, reason="", by_actor="alice", shards_dir=None, db_path=db,
    )
    assert result["status"] == "not_found"
    assert result["kind"] == "surface"