cli: extend 'aborist burn' to documents and cores

burn --kind {providence,document,core}; default 'providence' preserves
the prior --cache-key surface (no breaking change).

Children gates for document/core leaves:
  derivations.src_root        downstream cores derived from this doc
  edges.dst_root              other docs link to this doc
  providence_cache.source_root Q&A grounded in this doc

derivations has ON DELETE CASCADE on BOTH core_root and src_root, so a
naive DELETE FROM documents would silently cascade-prune downstream
derivations and orphan deeper cores. Gate before delete; --force to
override with the count recorded in the audit body.

Burn manually clears chunks_fts (FTS5 has no FK) and outbound edges
(no FK either) before deleting the document. chunks + merkle_nodes +
inbound derivations cascade via FK.

Audit event types are document_burn / core_burn — distinct from
providence_burn so chain consumers can tell leaf kinds apart.

CLAUDE.md "Cores never evict" remains intact: burn is operator-driven
removal, evict is the automated cold-tier compression that touches
only kind='surface'.

Makefile: KIND=providence|document|core selects the leaf type;
KEY for providence, ROOT for document/core. Single 'burn' target keeps
the surface DRY.

Tests (tests/test_burn_doc.py, 13 new):
- happy path: chunks + merkle_nodes vanish via cascade
- refused on derivation downstream / providence ref / incoming edge
- --force overrides and records counts
- core burn happy path (no children) preserves parent surface
- core burn refused with downstream derivation
- audit chain integrity preserved (no dangling prev_event_hash)
- CLI invocations for document, core, default-providence, refused

Full suite: 247 passed, 1 skipped.

Makefile

@@ -123,9 +123,17 @@ falsify: bootstrap ## mark a cached answer wrong: make falsify KEY=hex REASON='w
 	@if [ -z "$(KEY)" ]; then echo "usage: make falsify KEY=<cache_key> REASON='why'" >&2; exit 2; fi
 	$(ABORIST) --shards-dir $(SHARDS_DIR) providence --falsify $(KEY) --reason "$(REASON)"
 
-burn: bootstrap ## delete a leaf with no children: make burn KEY=hex REASON='why' [FORCE=1]
-	@if [ -z "$(KEY)" ]; then echo "usage: make burn KEY=<cache_key> REASON='why' [FORCE=1]" >&2; exit 2; fi
-	$(ABORIST) --shards-dir $(SHARDS_DIR) burn --cache-key $(KEY) --reason "$(REASON)" $(if $(FORCE),--force,)
+burn: bootstrap ## delete a leaf with no children. providence: KEY=<cache_key>; document/core: KIND=document|core ROOT=<hex>. REASON='why' [FORCE=1]
+	@kind="$${KIND:-providence}"; \
+	if [ "$$kind" = "providence" ]; then \
+	  if [ -z "$(KEY)" ]; then echo "usage: make burn KEY=<cache_key> REASON='why' [FORCE=1]" >&2; exit 2; fi; \
+	  $(ABORIST) --shards-dir $(SHARDS_DIR) burn --kind providence --cache-key $(KEY) --reason "$(REASON)" $(if $(FORCE),--force,); \
+	elif [ "$$kind" = "document" ] || [ "$$kind" = "core" ]; then \
+	  if [ -z "$(ROOT)" ]; then echo "usage: make burn KIND=$$kind ROOT=<document_root> REASON='why' [FORCE=1]" >&2; exit 2; fi; \
+	  $(ABORIST) --shards-dir $(SHARDS_DIR) burn --kind $$kind --root $(ROOT) --reason "$(REASON)" $(if $(FORCE),--force,); \
+	else \
+	  echo "unknown KIND: $$kind (expected: providence|document|core)" >&2; exit 2; \
+	fi
 
 # Multi-source RAG query against the shard cluster.
 # Usage: make query Q="What is anarcho-capitalism?"

aborist/cli.py

@@ -533,16 +533,330 @@ def _burn_cache_key(
     return {"status": "not_found", "cache_key": cache_key_value}
 
 
-def _cmd_burn(args: argparse.Namespace) -> int:
-    """CLI: burn a providence_cache leaf with no children."""
-    result = _burn_cache_key(
-        args.cache_key,
-        reason=args.reason or "",
-        by_actor=args.by_actor or os.environ.get("USER", "unknown"),
-        shards_dir=Path(args.global_shards_dir) if args.global_shards_dir else None,
-        db_path=Path(args.db) if args.db else None,
-        force=bool(args.force),
+def _count_document_children(c, document_root: str) -> dict:
+    """Count outbound child references that 'burn' must protect.
+
+    For a document/core leaf, "children" = anything downstream that built on
+    this row. Specifically:
+      - derivations rows where ``src_root = root`` (a core was distilled
+        from this — burning would orphan or silently cascade-truncate the
+        derivation, leaving the descendant core dangling).
+      - edges rows where ``dst_root = root`` (other documents link to this
+        one; burning leaves dangling references).
+      - providence_cache rows where ``source_root = root`` (Q&A grounded
+        in this document).
+
+    NOTE on schema: derivations has ON DELETE CASCADE on BOTH ``core_root``
+    AND ``src_root``. Without this gate, a bare DELETE FROM documents would
+    silently cascade-prune derivations and orphan downstream cores.
+    """
+    derivations_downstream = c.execute(
+        "SELECT COUNT(*) FROM derivations WHERE src_root = ?",
+        (document_root,),
+    ).fetchone()[0]
+    incoming_edges = c.execute(
+        "SELECT COUNT(*) FROM edges WHERE dst_root = ?",
+        (document_root,),
+    ).fetchone()[0]
+    providence_refs = c.execute(
+        "SELECT COUNT(*) FROM providence_cache WHERE source_root = ?",
+        (document_root,),
+    ).fetchone()[0]
+    return {
+        "derivations_downstream": int(derivations_downstream),
+        "incoming_edges": int(incoming_edges),
+        "providence_refs": int(providence_refs),
+    }
+
+
+def _burn_document_root(
+    document_root_value: str,
+    *,
+    reason: str,
+    by_actor: str,
+    shards_dir: Path | None,
+    db_path: Path | None,
+    force: bool = False,
+) -> dict:
+    """Delete a surface document leaf, but only if it has no children.
+
+    Children:
+      - 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)
+
+    On burn:
+      - DELETE FROM chunks_fts (FTS5 has no FK; clear before chunks vanish).
+      - DELETE FROM documents — cascades to chunks + merkle_nodes via FK.
+      - Append a 'document_burn' audit event recording counts + forced flag.
+
+    Refuses with status='refused_has_children' (and skips the audit event)
+    when any child count > 0 and ``--force`` is not set, so callers can fix
+    state and retry idempotently.
+    """
+    import time as _time
+
+    from aborist.store import append_audit, discover_shards, transaction
+
+    paths: list[Path] = (
+        discover_shards(shards_dir) if shards_dir else [Path(db_path)]
     )
+
+    for sp in paths:
+        c = connect(sp)
+        try:
+            row = c.execute(
+                "SELECT document_root, document_uri, kind, title, source_type, "
+                "       chunking_version, canonicalization_version, schema_version "
+                "FROM documents WHERE document_root = ? AND kind = 'surface'",
+                (document_root_value,),
+            ).fetchone()
+            if row is None:
+                continue
+            counts = _count_document_children(c, document_root_value)
+            total_children = sum(counts.values())
+            if total_children > 0 and not force:
+                return {
+                    "status": "refused_has_children",
+                    "document_root": document_root_value,
+                    "kind": "surface",
+                    "shard": sp.name,
+                    **counts,
+                    "hint": "use --force to burn anyway (orphans descendants); "
+                            "prefer evict for cold-tier compression",
+                }
+            chunk_count = c.execute(
+                "SELECT COUNT(*) FROM chunks WHERE document_root = ?",
+                (document_root_value,),
+            ).fetchone()[0]
+            now = int(_time.time())
+            with transaction(c):
+                # FTS5 has no FK to chunks; clear by chunk_id before the
+                # CASCADE on documents wipes the rows that resolve them.
+                for cr in c.execute(
+                    "SELECT chunk_id FROM chunks WHERE document_root = ?",
+                    (document_root_value,),
+                ).fetchall():
+                    c.execute(
+                        "DELETE FROM chunks_fts WHERE rowid = ?",
+                        (cr["chunk_id"],),
+                    )
+                # Outbound edges (src_root = this) carry no FK; clean them
+                # explicitly so we don't leave half-edges pointing from a
+                # ghost. Incoming edges (dst_root = this) are already gated
+                # above by the children check.
+                c.execute(
+                    "DELETE FROM edges WHERE src_root = ?",
+                    (document_root_value,),
+                )
+                # documents -> chunks/merkle_nodes/derivations cascade via FK.
+                c.execute(
+                    "DELETE FROM documents WHERE document_root = ?",
+                    (document_root_value,),
+                )
+            event_hash = append_audit(
+                c,
+                event_type="document_burn",
+                subject_root=document_root_value,
+                body={
+                    "document_root": document_root_value,
+                    "document_uri": row["document_uri"],
+                    "kind": "surface",
+                    "title": row["title"],
+                    "source_type": row["source_type"],
+                    "burned_chunk_count": int(chunk_count),
+                    "child_counts_at_burn": counts,
+                    "reason": reason,
+                    "by_actor": by_actor,
+                    "forced": bool(total_children > 0 and force),
+                },
+                ts=now,
+            )
+            return {
+                "status": "burned",
+                "document_root": document_root_value,
+                "kind": "surface",
+                "shard": sp.name,
+                "burned_chunk_count": int(chunk_count),
+                "child_counts_at_burn": counts,
+                "reason": reason,
+                "by_actor": by_actor,
+                "audit_event_hash": event_hash,
+                "ts": now,
+            }
+        finally:
+            c.close()
+
+    return {"status": "not_found", "document_root": document_root_value, "kind": "surface"}
+
+
+def _burn_core_root(
+    document_root_value: str,
+    *,
+    reason: str,
+    by_actor: str,
+    shards_dir: Path | None,
+    db_path: Path | None,
+    force: bool = False,
+) -> dict:
+    """Delete a core document leaf, but only if it has no children.
+
+    Same children gates as ``_burn_document_root`` (derivations.src_root,
+    edges.dst_root, providence_cache.source_root). The "PLUS no further
+    derivations build cores from this core" rule from the spec is
+    structurally identical to derivations.src_root > 0 — a core acts as a
+    src_root only when something deeper distilled from it.
+
+    CLAUDE.md says "Cores never evict" — that's the eviction subsystem,
+    which only touches kind='surface'. Burn is operator-driven removal:
+    cores CAN be burned, but the children gate is enforced.
+
+    Audit event type is 'core_burn' so chain consumers can distinguish
+    surface vs core leaf removals at a glance.
+    """
+    import time as _time
+
+    from aborist.store import append_audit, discover_shards, transaction
+
+    paths: list[Path] = (
+        discover_shards(shards_dir) if shards_dir else [Path(db_path)]
+    )
+
+    for sp in paths:
+        c = connect(sp)
+        try:
+            row = c.execute(
+                "SELECT document_root, document_uri, kind, title, source_type, "
+                "       compression_depth, chunking_version, "
+                "       canonicalization_version, schema_version "
+                "FROM documents WHERE document_root = ? AND kind = 'core'",
+                (document_root_value,),
+            ).fetchone()
+            if row is None:
+                continue
+            counts = _count_document_children(c, document_root_value)
+            total_children = sum(counts.values())
+            if total_children > 0 and not force:
+                return {
+                    "status": "refused_has_children",
+                    "document_root": document_root_value,
+                    "kind": "core",
+                    "shard": sp.name,
+                    **counts,
+                    "hint": "use --force to burn anyway; cores carry "
+                            "downstream derivations that will be orphaned",
+                }
+            chunk_count = c.execute(
+                "SELECT COUNT(*) FROM chunks WHERE document_root = ?",
+                (document_root_value,),
+            ).fetchone()[0]
+            # Inbound derivations (where this core is core_root, i.e. its
+            # binding back to source surfaces). These are NOT children —
+            # they're the core's own provenance and cascade-delete with it.
+            inbound_derivations = c.execute(
+                "SELECT COUNT(*) FROM derivations WHERE core_root = ?",
+                (document_root_value,),
+            ).fetchone()[0]
+            now = int(_time.time())
+            with transaction(c):
+                for cr in c.execute(
+                    "SELECT chunk_id FROM chunks WHERE document_root = ?",
+                    (document_root_value,),
+                ).fetchall():
+                    c.execute(
+                        "DELETE FROM chunks_fts WHERE rowid = ?",
+                        (cr["chunk_id"],),
+                    )
+                c.execute(
+                    "DELETE FROM edges WHERE src_root = ?",
+                    (document_root_value,),
+                )
+                c.execute(
+                    "DELETE FROM documents WHERE document_root = ?",
+                    (document_root_value,),
+                )
+            event_hash = append_audit(
+                c,
+                event_type="core_burn",
+                subject_root=document_root_value,
+                body={
+                    "document_root": document_root_value,
+                    "document_uri": row["document_uri"],
+                    "kind": "core",
+                    "title": row["title"],
+                    "source_type": row["source_type"],
+                    "compression_depth": int(row["compression_depth"]),
+                    "burned_chunk_count": int(chunk_count),
+                    "burned_inbound_derivations": int(inbound_derivations),
+                    "child_counts_at_burn": counts,
+                    "reason": reason,
+                    "by_actor": by_actor,
+                    "forced": bool(total_children > 0 and force),
+                },
+                ts=now,
+            )
+            return {
+                "status": "burned",
+                "document_root": document_root_value,
+                "kind": "core",
+                "shard": sp.name,
+                "burned_chunk_count": int(chunk_count),
+                "burned_inbound_derivations": int(inbound_derivations),
+                "child_counts_at_burn": counts,
+                "reason": reason,
+                "by_actor": by_actor,
+                "audit_event_hash": event_hash,
+                "ts": now,
+            }
+        finally:
+            c.close()
+
+    return {"status": "not_found", "document_root": document_root_value, "kind": "core"}
+
+
+def _cmd_burn(args: argparse.Namespace) -> int:
+    """CLI: burn a leaf with no children.
+
+    Dispatches on ``--kind`` to the matching helper. Default 'providence'
+    preserves the original surface (`--cache-key` only) so existing scripts
+    keep working. Document/core kinds use ``--root``.
+    """
+    kind = getattr(args, "kind", "providence") or "providence"
+    shards = Path(args.global_shards_dir) if args.global_shards_dir else None
+    db = Path(args.db) if args.db else None
+    actor = args.by_actor or os.environ.get("USER", "unknown")
+    reason = args.reason or ""
+    force = bool(args.force)
+
+    if kind == "providence":
+        if not getattr(args, "cache_key", None):
+            print("burn --kind providence requires --cache-key", file=sys.stderr)
+            return 2
+        result = _burn_cache_key(
+            args.cache_key,
+            reason=reason,
+            by_actor=actor,
+            shards_dir=shards,
+            db_path=db,
+            force=force,
+        )
+    elif kind in ("document", "core"):
+        if not getattr(args, "root", None):
+            print(f"burn --kind {kind} requires --root", file=sys.stderr)
+            return 2
+        helper = _burn_document_root if kind == "document" else _burn_core_root
+        result = helper(
+            args.root,
+            reason=reason,
+            by_actor=actor,
+            shards_dir=shards,
+            db_path=db,
+            force=force,
+        )
+    else:
+        print(f"unknown burn kind: {kind}", file=sys.stderr)
+        return 2
+
     print(json.dumps(result, indent=2))
     return 0 if result.get("status") == "burned" else 1
 
@@ -1961,20 +2275,32 @@ def build_parser() -> argparse.ArgumentParser:
     burn_cmd = sub.add_parser(
         "burn",
         help=(
-            "delete a providence_cache leaf with no children "
-            "(kindergarten use; falsify is the audit-preserving alternative)"
+            "delete a leaf with no children — providence_cache, document, "
+            "or core (kindergarten use; falsify/evict are audit-preserving)"
         ),
     )
+    burn_cmd.add_argument(
+        "--kind",
+        choices=("providence", "document", "core"),
+        default="providence",
+        help="leaf kind to burn (default: providence — backwards-compatible)",
+    )
     burn_cmd.add_argument(
         "--cache-key",
         dest="cache_key",
-        required=True,
-        help="cache_key (hex) of the providence record to burn",
+        default=None,
+        help="cache_key (hex) of the providence record to burn (kind=providence)",
+    )
+    burn_cmd.add_argument(
+        "--root",
+        dest="root",
+        default=None,
+        help="document_root (hex) of the document/core to burn (kind=document|core)",
     )
     burn_cmd.add_argument(
         "--reason",
         default=None,
-        help="reason text recorded in the providence_burn audit event",
+        help="reason text recorded in the burn audit event",
     )
     burn_cmd.add_argument(
         "--by-actor",
@@ -1985,7 +2311,7 @@ def build_parser() -> argparse.ArgumentParser:
     burn_cmd.add_argument(
         "--force",
         action="store_true",
-        help="burn even if children (falsifications) exist; not recommended",
+        help="burn even if children exist; not recommended",
     )
     burn_cmd.set_defaults(func=_cmd_burn)
 

tests/test_burn_doc.py

@@ -0,0 +1,582 @@
+"""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 aborist.cli import _burn_core_root, _burn_document_root
+from aborist.distill import FirstSentenceDistiller
+from aborist.distill.runner import distill_existing
+from aborist.document import Document
+from aborist.ingest import ingest_source
+from aborist.source import Source
+from aborist.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 = (
+    "Aborist 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 aborist.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 aborist.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 aborist.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 aborist.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"