commit 856b3116d71584927dee040976c5d67a50813dff Author: russell@unturf.com Date: Mon Apr 27 07:53:18 2026 -0400 phase 0 explore: aborist core + sources + distill + evict A content-addressed, Merkle-committed document store implementing the runtime spec from Merkle Providence Reverse RAG (April 2026 whitepaper) and Merkle-AGI v9.8 admissibility ledger. Ports proxy.unturf.com Go merkle conventions to Python: non-commutative HashCombine with 0x03 prefix, explicit IsLeft per sibling, self-duplicate odd elements. What's in: - merkle.py — proof generation/verification, JSON serialization - store.py — v9.8 SQLite schema: 8-dim providence_cache key, falsification_state, append-only audit chain, surface/core kind, hot/warm/cold tier, derivations, edges - ingest.py — Source -> normalize -> chunk -> merkle -> upsert, idempotent on document_root collision - search/ — SearchBackend ABC with explicit AuditMode (STRICT/HYBRID/ VISUAL), FTS5 backend returning VISUAL hits - sources/ — wikipedia.py (streaming bz2/MySQL extended-INSERT parser for 2003-era cur dumps); html_page.py (selectolax + httpx, robots.txt honored automatically) - distill/ — Distiller ABC + first-sentence-v1 stub. Runner generates per-contributing-chunk Merkle proofs binding cores back to source document_root. - evict.py — hot->cold demote (NULLs content, drops FTS row, retains leaf_hash). rehydrate() refetches via source pipeline; matching root restores content, mismatching root marks providence stale and writes rehydrate_drift event. Cores never evict. - cli.py — ingest / search / verify / stats / distill / evict / rehydrate - 31 tests covering merkle round-trip, ingest+audit, chunker version binding, html parse, distillation proof verification, evict+ rehydrate including drift detection. Smoke: 503 Wikipedia 2003-05-16 + 3 fox-owned HTML pages ingested, 478 cores produced (24 surface->core merkle dedups), 7 chunks evicted to cold and round-tripped via rehydrate, 987 audit events chained 0 breaks. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..37d21cb --- /dev/null +++ b/.gitignore @@ -0,0 +1,14 @@ +__pycache__/ +*.py[cod] +*.egg-info/ +.venv/ +.pytest_cache/ +.coverage +*.swp + +# data + caches stay out of git +data/ +*.db +*.db-journal +*.db-wal +*.db-shm diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..445c514 --- /dev/null +++ b/LICENSE @@ -0,0 +1,13 @@ +aborist — an arborist for trees and forests of cross-linked information +Copyright (C) 2026 Russell Ballestrini and contributors + +This program is free software: you can redistribute it and/or modify it under +the terms of the GNU Affero General Public License as published by the Free +Software Foundation, version 3 of the License (only). + +This program is distributed in the hope that it will be useful, but WITHOUT ANY +WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A +PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. + +You should have received a copy of the GNU Affero General Public License along +with this program. If not, see . diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..938be27 --- /dev/null +++ b/Makefile @@ -0,0 +1,70 @@ +# aborist — Makefile entry points +# Every workflow lives behind a `make` target. Bare python commands are not +# the user interface. + +# Tools and config +PYTHON ?= python3 +VENV ?= .venv +PIP := $(VENV)/bin/pip +PY := $(VENV)/bin/python +ABORIST := $(VENV)/bin/aborist + +# Data + DB +DATA_DIR ?= data +WP_DUMP_URL ?= https://dumps.wikimedia.org/archive/2003/2003-05-16/en/20030516_cur_tablesql.bz2 +WP_DUMP := $(DATA_DIR)/20030516_cur_tablesql.bz2 +DB ?= $(HOME)/.aborist/aborist.db + +# Smoke-test caps so make all stays fast +INGEST_LIMIT ?= 500 +VERIFY_N ?= 10 +SEARCH_Q ?= computer + +.PHONY: all bootstrap fetch ingest verify search stats test clean clean-db clean-data help + +all: bootstrap fetch ingest verify stats ## bootstrap → fetch → ingest → verify → stats + +help: ## show this help + @awk 'BEGIN{FS=":.*##"} /^[a-zA-Z0-9_-]+:.*##/{printf " %-14s %s\n", $$1, $$2}' $(MAKEFILE_LIST) + +$(VENV)/bin/activate: pyproject.toml + $(PYTHON) -m venv $(VENV) + $(PIP) install --upgrade pip wheel + $(PIP) install -e '.[dev]' + @touch $(VENV)/bin/activate + +bootstrap: $(VENV)/bin/activate ## create venv and install editable package + +$(WP_DUMP): | $(DATA_DIR) + @echo ">> fetching $(WP_DUMP_URL)" + curl -fL --retry 3 -o $@ "$(WP_DUMP_URL)" + +$(DATA_DIR): + mkdir -p $(DATA_DIR) + +fetch: $(WP_DUMP) ## download Wikipedia 2003-05-16 cur dump (cached) + +ingest: bootstrap fetch ## ingest INGEST_LIMIT articles into $(DB) + $(ABORIST) --db $(DB) ingest --source wikipedia_cur --path $(WP_DUMP) --limit $(INGEST_LIMIT) + +verify: bootstrap ## round-trip Merkle proofs for VERIFY_N random documents + $(ABORIST) --db $(DB) verify -n $(VERIFY_N) + +search: bootstrap ## keyword search; override SEARCH_Q (or pass Q=...) + $(ABORIST) --db $(DB) search '$(if $(Q),$(Q),$(SEARCH_Q))' + +stats: bootstrap ## counts: documents, chunks, edges, audit chain + $(ABORIST) --db $(DB) stats + +test: bootstrap ## run pytest suite + $(VENV)/bin/pytest -q + +clean: ## remove venv + caches (keeps fetched data and db) + rm -rf $(VENV) .pytest_cache **/__pycache__ aborist.egg-info + find . -type d -name __pycache__ -prune -exec rm -rf {} + + +clean-db: ## drop the aborist db (keeps fetched data and venv) + rm -f $(DB) $(DB)-journal $(DB)-wal $(DB)-shm + +clean-data: ## remove fetched dumps + rm -rf $(DATA_DIR) diff --git a/README.md b/README.md new file mode 100644 index 0000000..95060bf --- /dev/null +++ b/README.md @@ -0,0 +1,33 @@ +# aborist + +An arborist for trees and forests of cross-linked information. + +Aborist ingests documents into a content-addressed, Merkle-committed store. Every +document carries a URI for backtrack and cross-link. Search hooks return hits with +declared audit modes (`STRICT` / `HYBRID` / `VISUAL`) so callers never overclaim +what a result actually proves. + +## Design + +Aborist implements the runtime spec from Merkle-AGI v9.8 / Merkle Providence +Reverse RAG. The 8-dim admissibility key (`source_root`, `question_hash`, +`model_profile_hash`, `conversation_hash`, `governance_policy_hash`, +`schema_version`, `canonicalization_version`, `chunking_version`) plus +falsification state ensures cached records are never reused under drift. + +Two layered document kinds: +- **surface** — diverse ingested content (full chunks, FTS-indexed) +- **core** — distilled records (haiku/equation/snippet) Merkle-signed back to + source surface roots via `derivations` + +Reversible eviction via chunk `tier` ∈ {`hot`, `warm`, `cold`}. Cold = leaf hash ++ URI only; rehydratable from URI, identity verifiable via leaf hash. + +## Quick start + + make all # bootstrap + fetch + ingest + verify + stats + make search Q='…' + +## License + +AGPL-3.0-only. diff --git a/aborist/__init__.py b/aborist/__init__.py new file mode 100644 index 0000000..1fda08b --- /dev/null +++ b/aborist/__init__.py @@ -0,0 +1,9 @@ +"""aborist — an arborist for trees and forests of cross-linked information.""" + +__version__ = "0.0.1" + +# Schema/canonicalization/chunking versions are part of the v9.8 admissibility +# key. Bumping any of these auto-stales every prior cache record. +SCHEMA_VERSION = "v9.8.0" +CANONICALIZATION_VERSION = "norm-v1" +CHUNKING_VERSION = "tok-512-v1" diff --git a/aborist/cli.py b/aborist/cli.py new file mode 100644 index 0000000..198ec13 --- /dev/null +++ b/aborist/cli.py @@ -0,0 +1,315 @@ +"""Aborist CLI: ingest / search / verify / stats.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from aborist import __version__ +from aborist.ingest import ingest_source, verify_random_sample +from aborist.search import FTS5Backend +from aborist.sources import WikipediaCurDump +from aborist.store import DEFAULT_DB_PATH, connect, stats + + +def _cmd_ingest(args: argparse.Namespace) -> int: + if args.source == "wikipedia_cur": + if not args.path: + print("--path is required for wikipedia_cur", file=sys.stderr) + return 2 + src = WikipediaCurDump(path=args.path) + elif args.source == "html": + try: + from aborist.sources import HtmlPageSource + except ImportError: + print( + "html source requires extras: pip install 'aborist[html]'", + file=sys.stderr, + ) + return 2 + urls: list[str] = list(args.url or []) + if args.urls_from: + urls.extend( + line.strip() + for line in Path(args.urls_from).read_text(encoding="utf-8").splitlines() + if line.strip() and not line.lstrip().startswith("#") + ) + if not urls: + print("html source needs --url or --urls-from", file=sys.stderr) + return 2 + src = HtmlPageSource(urls, respect_robots=not args.no_robots) + else: + print(f"unknown source: {args.source}", file=sys.stderr) + return 2 + + conn = connect(args.db) + try: + result = ingest_source(conn, src, chunker_name=args.chunker, limit=args.limit) + finally: + conn.close() + print(json.dumps(result.__dict__, indent=2)) + return 0 + + +def _cmd_search(args: argparse.Namespace) -> int: + conn = connect(args.db) + try: + backend = FTS5Backend(conn) + hits = backend.search(args.query, limit=args.limit) + finally: + conn.close() + if args.json: + print( + json.dumps( + [ + { + "document_root": h.document_root, + "document_uri": h.document_uri, + "chunk_idx": h.chunk_idx, + "snippet": h.snippet, + "score": h.score, + "audit_mode": h.audit_mode.value, + "title": h.title, + } + for h in hits + ], + indent=2, + ) + ) + else: + for h in hits: + print(f"[{h.audit_mode.value}] {h.score:7.3f} {h.title or h.document_uri}") + print(f" chunk {h.chunk_idx} root={h.document_root[:16]}…") + print(f" {h.snippet}") + print() + return 0 + + +def _cmd_verify(args: argparse.Namespace) -> int: + conn = connect(args.db) + try: + result = verify_random_sample(conn, n=args.n) + finally: + conn.close() + print(json.dumps(result, indent=2)) + return 0 if result["failed"] == 0 else 1 + + +def _cmd_distill(args: argparse.Namespace) -> int: + from aborist.distill import get_distiller + from aborist.distill.runner import distill_existing + + try: + distiller = get_distiller(args.process) + except ValueError as e: + print(str(e), file=sys.stderr) + return 2 + conn = connect(args.db) + try: + result = distill_existing( + conn, + distiller, + source_type=args.source_type, + limit=args.limit, + chunker_name=args.chunker, + ) + finally: + conn.close() + print(json.dumps(result, indent=2)) + return 0 + + +def _cmd_evict(args: argparse.Namespace) -> int: + from aborist.evict import evict_to_cold + + conn = connect(args.db) + try: + result = evict_to_cold( + conn, + source_type=args.source_type, + older_than_days=args.older_than_days, + document_roots=args.document_root or None, + ) + finally: + conn.close() + print(json.dumps(result, indent=2)) + return 0 + + +def _cmd_rehydrate(args: argparse.Namespace) -> int: + from aborist.evict import rehydrate + + conn = connect(args.db) + try: + if args.all_cold: + roots = [ + r["document_root"] + for r in conn.execute( + "SELECT DISTINCT document_root FROM chunks WHERE tier = 'cold'" + ).fetchall() + ] + else: + roots = list(args.document_root or []) + if not roots: + print( + "rehydrate needs --document-root R or --all-cold", + file=sys.stderr, + ) + return 2 + results = [] + for r in roots: + res = rehydrate(conn, r) + res["document_root"] = r + results.append(res) + finally: + conn.close() + print(json.dumps(results, indent=2)) + drift = sum(1 for r in results if r.get("status") == "drift_detected") + return 1 if drift else 0 + + +def _cmd_stats(args: argparse.Namespace) -> int: + conn = connect(args.db) + try: + result = stats(conn) + finally: + conn.close() + print(json.dumps(result, indent=2)) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="aborist", + description="An arborist for trees and forests of cross-linked information.", + ) + p.add_argument("--version", action="version", version=f"aborist {__version__}") + p.add_argument( + "--db", + type=Path, + default=DEFAULT_DB_PATH, + help=f"path to aborist SQLite db (default: {DEFAULT_DB_PATH})", + ) + sub = p.add_subparsers(dest="cmd", required=True) + + ingest = sub.add_parser("ingest", help="ingest documents from a source") + ingest.add_argument( + "--source", + required=True, + choices=["wikipedia_cur", "html"], + help="source type", + ) + ingest.add_argument("--path", help="path to dump file (file-backed sources)") + ingest.add_argument( + "--url", action="append", help="URL to ingest (html source; repeatable)" + ) + ingest.add_argument( + "--urls-from", + dest="urls_from", + help="file with one URL per line (html source)", + ) + ingest.add_argument( + "--no-robots", + dest="no_robots", + action="store_true", + help="do not consult robots.txt (use only for explicitly authorized sites)", + ) + ingest.add_argument( + "--chunker", default=None, help="chunker name (default: tok-512-v1)" + ) + ingest.add_argument( + "--limit", type=int, default=None, help="cap number of documents" + ) + ingest.set_defaults(func=_cmd_ingest) + + search = sub.add_parser("search", help="keyword search (VISUAL audit mode)") + search.add_argument("query", help="query string") + search.add_argument("--limit", type=int, default=20) + search.add_argument("--json", action="store_true", help="output JSON") + search.set_defaults(func=_cmd_search) + + verify = sub.add_parser( + "verify", help="round-trip Merkle proofs for N random documents" + ) + verify.add_argument("-n", type=int, default=10) + verify.set_defaults(func=_cmd_verify) + + distill = sub.add_parser( + "distill", + help="compress surface docs into Merkle-signed core docs", + ) + distill.add_argument( + "--process", default="first-sentence-v1", help="distiller name" + ) + distill.add_argument( + "--source-type", + dest="source_type", + default=None, + help="restrict to one source_type", + ) + distill.add_argument( + "--chunker", default=None, help="chunker for the core doc" + ) + distill.add_argument( + "--limit", type=int, default=None, help="cap number of surface docs" + ) + distill.set_defaults(func=_cmd_distill) + + evict_cmd = sub.add_parser( + "evict", + help="demote surface chunks hot→cold (NULL content, retain leaf_hash)", + ) + evict_cmd.add_argument( + "--source-type", + dest="source_type", + default=None, + help="restrict to one source_type", + ) + evict_cmd.add_argument( + "--older-than-days", + dest="older_than_days", + type=int, + default=None, + help="only evict docs older than N days", + ) + evict_cmd.add_argument( + "--document-root", + action="append", + default=None, + help="explicit document_root(s) to evict; repeatable", + ) + evict_cmd.set_defaults(func=_cmd_evict) + + rehydrate_cmd = sub.add_parser( + "rehydrate", + help="refetch URI, verify leaves, restore cold content if root matches", + ) + rehydrate_cmd.add_argument( + "--document-root", + action="append", + default=None, + help="explicit document_root(s) to rehydrate; repeatable", + ) + rehydrate_cmd.add_argument( + "--all-cold", + dest="all_cold", + action="store_true", + help="rehydrate every document with cold chunks", + ) + rehydrate_cmd.set_defaults(func=_cmd_rehydrate) + + stats_cmd = sub.add_parser("stats", help="counts: docs, chunks, edges, audit") + stats_cmd.set_defaults(func=_cmd_stats) + + return p + + +def main(argv: list[str] | None = None) -> int: + args = build_parser().parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/aborist/distill/__init__.py b/aborist/distill/__init__.py new file mode 100644 index 0000000..4b034dc --- /dev/null +++ b/aborist/distill/__init__.py @@ -0,0 +1,12 @@ +"""Distillation: surface docs -> core docs, Merkle-signed back.""" + +from aborist.distill.base import DistillationResult, Distiller +from aborist.distill.first_sentence import FirstSentenceDistiller + +__all__ = ["DistillationResult", "Distiller", "FirstSentenceDistiller"] + + +def get_distiller(name: str) -> Distiller: + if name == FirstSentenceDistiller.name: + return FirstSentenceDistiller() + raise ValueError(f"unknown distiller: {name}") diff --git a/aborist/distill/base.py b/aborist/distill/base.py new file mode 100644 index 0000000..80be4a8 --- /dev/null +++ b/aborist/distill/base.py @@ -0,0 +1,42 @@ +"""Distiller ABC. + +A Distiller compresses a surface Document into a core Document. The runner +generates Merkle proofs for every contributing source chunk so the resulting +derivation row cryptographically binds the core back to its source. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass + +from aborist.document import Document + + +@dataclass +class DistillationResult: + """What a Distiller returns for one source. + + contributing_chunk_indices: 0-based indices into source_chunks that fed + the core's content. The runner Merkle-proves each one against the + source's document_root and stores those proofs in derivations.proof_blob. + """ + + core: Document + contributing_chunk_indices: list[int] + + +class Distiller(ABC): + """Pure function: surface Document + its chunks -> core DistillationResult. + + Distillers must be deterministic — same source bytes produce the same core + bytes. Bumping a distiller's algorithm requires bumping its `name`. + """ + + name: str + + @abstractmethod + def distill( + self, source: Document, source_chunks: list[str] + ) -> DistillationResult: + ... diff --git a/aborist/distill/first_sentence.py b/aborist/distill/first_sentence.py new file mode 100644 index 0000000..15f0a76 --- /dev/null +++ b/aborist/distill/first_sentence.py @@ -0,0 +1,63 @@ +"""First-sentence-per-chunk distiller (deterministic, no ML). + +Compresses a source by taking the first non-trivial sentence of each chunk +and concatenating. Useful as a stub to exercise the core/derivation schema +without external models. +""" + +from __future__ import annotations + +import re + +from aborist.distill.base import DistillationResult, Distiller +from aborist.document import Document + +_SENTENCE_BOUNDARY = re.compile(r"(?<=[.!?])\s+(?=[A-Z0-9])") +_MIN_SENTENCE_LEN = 10 + + +class FirstSentenceDistiller(Distiller): + name = "first-sentence-v1" + + def __init__(self, max_chars: int = 4096): + self.max_chars = max_chars + + def distill( + self, source: Document, source_chunks: list[str] + ) -> DistillationResult: + sentences: list[str] = [] + contributing: list[int] = [] + for chunk_idx, chunk in enumerate(source_chunks): + first = self._first_sentence(chunk) + if first: + sentences.append(first) + contributing.append(chunk_idx) + + core_text = "\n".join(sentences) + if len(core_text) > self.max_chars: + core_text = core_text[: self.max_chars].rstrip() + + core = Document( + uri=f"{source.uri}#core/{self.name}", + content=core_text, + source_type=f"core:{self.name}", + title=(source.title or "") + " [CORE]", + ) + return DistillationResult(core=core, contributing_chunk_indices=contributing) + + @staticmethod + def _first_sentence(text: str) -> str | None: + text = text.strip() + if not text: + return None + # Try paragraph-aware: split on blank lines first, then take first paragraph. + for para in text.split("\n\n"): + para = para.strip() + if not para: + continue + # First sentence within the paragraph. + parts = _SENTENCE_BOUNDARY.split(para, maxsplit=1) + first = parts[0].strip() + if len(first) >= _MIN_SENTENCE_LEN: + return first + return None diff --git a/aborist/distill/runner.py b/aborist/distill/runner.py new file mode 100644 index 0000000..372e222 --- /dev/null +++ b/aborist/distill/runner.py @@ -0,0 +1,216 @@ +"""Distill existing surface docs into core docs. + +For each surface row in `documents`: +1. Load its chunks (must be hot tier — content present). +2. Run the Distiller. +3. Compute the core's own Merkle tree (it's a normal Document). +4. Generate a Merkle proof for every contributing source chunk against the + source document_root. Pack into proof_blob (JSON). +5. Insert the core Document, its chunks, FTS rows, interior merkle nodes, + the derivation row, a `derived_from` edge, and an audit event. +6. Idempotent: re-running with the same distiller skips existing core_root. +""" + +from __future__ import annotations + +import json +import sqlite3 +import time + +from aborist import ( + CANONICALIZATION_VERSION, + SCHEMA_VERSION, +) +from aborist.distill.base import Distiller +from aborist.document import Document, canonicalize, get_chunker +from aborist.merkle import MerkleTree, hash_leaf, proof_to_dict +from aborist.store import append_audit, transaction + + +def distill_existing( + conn: sqlite3.Connection, + distiller: Distiller, + *, + source_type: str | None = None, + limit: int | None = None, + chunker_name: str | None = None, +) -> dict: + """Distill surface documents already in the store. Returns counts.""" + chunker = get_chunker(chunker_name) + + where_clauses = ["kind = 'surface'"] + params: list = [] + if source_type: + where_clauses.append("source_type = ?") + params.append(source_type) + sql = ( + "SELECT document_root, document_uri, title, source_type FROM documents " + "WHERE " + " AND ".join(where_clauses) + " ORDER BY ingest_ts ASC" + ) + if limit: + sql += f" LIMIT {int(limit)}" + surface_rows = conn.execute(sql, params).fetchall() + + distilled = 0 + skipped_cold = 0 + skipped_existing = 0 + skipped_empty = 0 + + for s in surface_rows: + src_root = s["document_root"] + chunk_rows = conn.execute( + "SELECT idx, leaf_hash, content FROM chunks " + "WHERE document_root = ? ORDER BY idx ASC", + (src_root,), + ).fetchall() + if any(r["content"] is None for r in chunk_rows): + skipped_cold += 1 + continue + + chunk_strs = [r["content"] for r in chunk_rows] + src_doc = Document( + uri=s["document_uri"], + content="", # not used by current distillers + source_type=s["source_type"], + title=s["title"], + ) + result = distiller.distill(src_doc, chunk_strs) + + core_text = canonicalize(result.core.content) + core_chunk_strs = chunker.split(core_text) + if not core_chunk_strs: + skipped_empty += 1 + continue + + core_leaves = [hash_leaf(c.encode("utf-8")) for c in core_chunk_strs] + core_tree = MerkleTree.build(core_leaves) + core_root = core_tree.root.hex() + + # Build source tree from the stored leaves to generate per-chunk proofs. + src_leaves = [bytes.fromhex(r["leaf_hash"]) for r in chunk_rows] + src_tree = MerkleTree.build(src_leaves) + contributing: list[dict] = [] + for idx in result.contributing_chunk_indices: + if 0 <= idx < len(src_leaves): + p = src_tree.proof(idx) + contributing.append( + {"src_chunk_idx": idx, "proof": proof_to_dict(p)} + ) + + proof_blob = json.dumps( + { + "process_id": distiller.name, + "core_root": core_root, + "src_root": src_root, + "contributing": contributing, + }, + separators=(",", ":"), + sort_keys=True, + ) + + with transaction(conn): + existing = conn.execute( + "SELECT 1 FROM documents WHERE document_root = ?", (core_root,) + ).fetchone() + if existing: + # Ensure derivation row is present even if core was made earlier. + conn.execute( + "INSERT OR IGNORE INTO derivations " + "(core_root, src_root, proof_blob, process_id, distilled_at) " + "VALUES (?, ?, ?, ?, ?)", + (core_root, src_root, proof_blob, distiller.name, int(time.time())), + ) + conn.execute( + "INSERT OR IGNORE INTO edges " + "(src_root, dst_root, dst_uri, edge_type, anchor) " + "VALUES (?, ?, ?, 'derived_from', '')", + (core_root, src_root, s["document_uri"]), + ) + skipped_existing += 1 + continue + + now = int(time.time()) + 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', 1, ?, ?, ?, ?, ?)", + ( + core_root, + result.core.uri, + result.core.source_type, + result.core.title, + chunker.name, + CANONICALIZATION_VERSION, + SCHEMA_VERSION, + now, + ), + ) + conn.executemany( + "INSERT INTO chunks (document_root, idx, leaf_hash, content) " + "VALUES (?, ?, ?, ?)", + [ + (core_root, i, core_leaves[i].hex(), core_chunk_strs[i]) + for i in range(len(core_chunk_strs)) + ], + ) + conn.executemany( + "INSERT INTO chunks_fts (document_root, idx, content) VALUES (?, ?, ?)", + [ + (core_root, i, core_chunk_strs[i]) + for i in range(len(core_chunk_strs)) + ], + ) + interior: list[tuple] = [] + for layer_idx in range(1, len(core_tree.layers)): + for node_idx, h in enumerate(core_tree.layers[layer_idx]): + interior.append((core_root, layer_idx, node_idx, h.hex())) + if interior: + conn.executemany( + "INSERT INTO merkle_nodes (document_root, layer, idx, hash) " + "VALUES (?, ?, ?, ?)", + interior, + ) + conn.execute( + "INSERT INTO derivations " + "(core_root, src_root, proof_blob, process_id, distilled_at) " + "VALUES (?, ?, ?, ?, ?)", + (core_root, src_root, proof_blob, distiller.name, now), + ) + conn.execute( + "INSERT OR IGNORE INTO edges " + "(src_root, dst_root, dst_uri, edge_type, anchor) " + "VALUES (?, ?, ?, 'derived_from', '')", + (core_root, src_root, s["document_uri"]), + ) + + src_chars = sum(len(c) for c in chunk_strs) + core_chars = sum(len(c) for c in core_chunk_strs) + append_audit( + conn, + event_type="derive", + subject_root=core_root, + body={ + "src_root": src_root, + "process_id": distiller.name, + "core_chunks": len(core_chunk_strs), + "src_chunks_used": len(contributing), + "src_chars": src_chars, + "core_chars": core_chars, + "compression_ratio": ( + round(core_chars / src_chars, 4) if src_chars else 0.0 + ), + }, + ts=now, + ) + + distilled += 1 + + return { + "scanned": len(surface_rows), + "distilled": distilled, + "skipped_existing": skipped_existing, + "skipped_cold": skipped_cold, + "skipped_empty": skipped_empty, + } diff --git a/aborist/document.py b/aborist/document.py new file mode 100644 index 0000000..73e9448 --- /dev/null +++ b/aborist/document.py @@ -0,0 +1,101 @@ +"""Document and Chunk dataclasses + canonical chunkers. + +Every Document carries a URI (identification, backtrack, cross-link) and content. +Chunkers split content into byte-determined chunks before Merkle hashing. The +chunker's name is committed in chunking_version — changing chunker invalidates +all prior cache records under v9.8 admissibility. +""" + +from __future__ import annotations + +import re +import unicodedata +from dataclasses import dataclass, field +from typing import Protocol + + +@dataclass(frozen=True) +class Edge: + """A cross-link from this document to another.""" + + edge_type: str # wikilink | citation | derived_from | ... + dst_uri: str # always present + dst_root: str | None = None # filled in later if/when target is ingested + anchor: str | None = None # optional fragment / chunk index + + +@dataclass +class Document: + """An ingestable document: URI + content + outbound edges.""" + + uri: str + content: str # normalized text + source_type: str # wikipedia_xml | html | git | ... + title: str | None = None + edges: list[Edge] = field(default_factory=list) + extra: dict = field(default_factory=dict) # source-specific metadata + + +def canonicalize(text: str) -> str: + """Stable text normalization. Bumping this requires CANONICALIZATION_VERSION bump.""" + # NFC unicode, normalize whitespace runs to single spaces, strip ends. + text = unicodedata.normalize("NFC", text) + text = re.sub(r"[\r\n\t\f\v]+", "\n", text) + text = re.sub(r"[ ]{2,}", " ", text) + return text.strip() + + +class Chunker(Protocol): + """A chunker splits canonicalized text into ordered chunks.""" + + name: str + + def split(self, text: str) -> list[str]: ... + + +class TokenChunker: + """512-token chunker (whitespace-tokenized, byte-deterministic). + + "Token" here means whitespace-separated unit, NOT a model BPE token. This + avoids tokenizer-version drift in the chunking_version. + """ + + name = "tok-512-v1" + + def __init__(self, tokens_per_chunk: int = 512): + self.tokens_per_chunk = tokens_per_chunk + + def split(self, text: str) -> list[str]: + if not text: + return [] + tokens = text.split() + if not tokens: + return [] + chunks: list[str] = [] + for start in range(0, len(tokens), self.tokens_per_chunk): + chunks.append(" ".join(tokens[start : start + self.tokens_per_chunk])) + return chunks + + +class SentenceChunker: + """Sentence-aligned chunker (better for short docs like 2003 Wikipedia).""" + + name = "sent-v1" + + _split_re = re.compile(r"(?<=[.!?])\s+(?=[A-Z0-9])") + + def split(self, text: str) -> list[str]: + if not text: + return [] + # Naive but deterministic. + sentences = [s.strip() for s in self._split_re.split(text) if s.strip()] + return sentences or ([text] if text else []) + + +def get_chunker(name: str | None = None) -> Chunker: + """Lookup chunker by name. Default = TokenChunker.""" + if name is None or name == TokenChunker.name: + return TokenChunker() + if name == SentenceChunker.name: + return SentenceChunker() + raise ValueError(f"unknown chunker: {name}") diff --git a/aborist/evict.py b/aborist/evict.py new file mode 100644 index 0000000..7641d2c --- /dev/null +++ b/aborist/evict.py @@ -0,0 +1,216 @@ +"""Reversible eviction + rehydrate. + +Implements the systematic-forgetting mechanic from the design philosophy: + +- evict_to_cold: surface chunks demote from `hot` to `cold`; content set to + NULL, FTS5 row deleted. leaf_hash retained — identity preserved. +- rehydrate: refetch URI through the same source pipeline, re-chunk with the + original chunking_version, compare leaves and root. Match -> content + restored, tier hot. Mismatch -> drift event in audit chain, providence + records flipped to falsification_state='stale'. No content restored. + +Cores never evict. +""" + +from __future__ import annotations + +import sqlite3 +import time +from typing import Callable, Iterable + +from aborist.document import canonicalize, get_chunker +from aborist.merkle import MerkleTree, hash_leaf +from aborist.store import append_audit, transaction + + +# Re-fetcher signature: takes a URI, returns parsed/canonicalized text or None. +Fetcher = Callable[[str], str | None] + + +def _default_html_fetcher(uri: str) -> str | None: + """Reuse HtmlPageSource so rehydrate runs the exact same pipeline as ingest.""" + try: + from aborist.sources.html_page import HtmlPageSource + except ImportError: + return None + src = HtmlPageSource([uri]) + for doc in src.iter_documents(): + return doc.content + return None + + +_FETCHERS: dict[str, Fetcher] = { + "html": _default_html_fetcher, +} + + +def evict_to_cold( + conn: sqlite3.Connection, + *, + source_type: str | None = None, + older_than_days: int | None = None, + document_roots: Iterable[str] | None = None, +) -> dict: + """Demote matching surface chunks from hot to cold. + + Cores are never evicted. Content is NULLed; FTS row removed. + """ + where = ["d.kind = 'surface'", "c.tier = 'hot'"] + params: list = [] + if source_type: + where.append("d.source_type = ?") + params.append(source_type) + if older_than_days is not None: + cutoff = int(time.time()) - older_than_days * 86400 + where.append("d.ingest_ts < ?") + params.append(cutoff) + if document_roots is not None: + roots = list(document_roots) + if not roots: + return {"evicted_chunks": 0, "documents_affected": 0} + placeholders = ",".join("?" for _ in roots) + where.append(f"c.document_root IN ({placeholders})") + params.extend(roots) + + sql = ( + "SELECT c.document_root, c.idx FROM chunks c " + "JOIN documents d ON d.document_root = c.document_root " + "WHERE " + " AND ".join(where) + ) + candidates = conn.execute(sql, params).fetchall() + if not candidates: + return {"evicted_chunks": 0, "documents_affected": 0} + + per_doc: dict[str, int] = {} + with transaction(conn): + for r in candidates: + conn.execute( + "UPDATE chunks SET content=NULL, tier='cold' " + "WHERE document_root=? AND idx=?", + (r["document_root"], r["idx"]), + ) + conn.execute( + "DELETE FROM chunks_fts WHERE document_root=? AND idx=?", + (r["document_root"], r["idx"]), + ) + per_doc[r["document_root"]] = per_doc.get(r["document_root"], 0) + 1 + + for doc_root, n in per_doc.items(): + append_audit( + conn, + event_type="evict_cold", + subject_root=doc_root, + body={"chunks_evicted": n}, + ) + + return { + "evicted_chunks": len(candidates), + "documents_affected": len(per_doc), + } + + +def rehydrate( + conn: sqlite3.Connection, + document_root: str, + *, + fetcher: Fetcher | None = None, +) -> dict: + """Refetch URI, verify leaves, restore content if and only if root matches. + + Returns a dict with `status` ∈ { + unknown_document, nothing_to_do, source_not_rehydratable, + fetch_failed, drift_detected, rehydrated + }. + """ + doc_row = conn.execute( + "SELECT document_uri, source_type, chunking_version " + "FROM documents WHERE document_root = ?", + (document_root,), + ).fetchone() + if doc_row is None: + return {"status": "unknown_document"} + + cold_chunks = conn.execute( + "SELECT idx, leaf_hash FROM chunks " + "WHERE document_root = ? AND tier = 'cold' ORDER BY idx", + (document_root,), + ).fetchall() + if not cold_chunks: + return {"status": "nothing_to_do", "cold_chunks": 0} + + # Pick fetcher by source_type unless caller supplies one. + use_fetcher = fetcher or _FETCHERS.get(doc_row["source_type"]) + if use_fetcher is None: + return { + "status": "source_not_rehydratable", + "source_type": doc_row["source_type"], + } + + try: + text = use_fetcher(doc_row["document_uri"]) + except Exception as e: # noqa: BLE001 — surface any error in status + return {"status": "fetch_failed", "error": repr(e)} + if text is None: + return {"status": "fetch_failed", "error": "fetcher returned None"} + + chunker = get_chunker(doc_row["chunking_version"]) + new_text = canonicalize(text) + new_chunk_strs = chunker.split(new_text) + new_leaves = [hash_leaf(c.encode("utf-8")) for c in new_chunk_strs] + new_root = MerkleTree.build(new_leaves).root.hex() + + if new_root != document_root: + with transaction(conn): + append_audit( + conn, + event_type="rehydrate_drift", + subject_root=document_root, + body={ + "expected_root": document_root, + "actual_root": new_root, + "uri": doc_row["document_uri"], + }, + ) + # v9.8 falsification: any cached providence record from this + # source is now stale. + conn.execute( + "UPDATE providence_cache SET falsification_state = 'stale' " + "WHERE source_root = ? AND falsification_state = 'live'", + (document_root,), + ) + return { + "status": "drift_detected", + "expected_root": document_root, + "actual_root": new_root, + } + + # Roots match. Restore content for every cold chunk. + restored = 0 + with transaction(conn): + for c in cold_chunks: + i = c["idx"] + if i >= len(new_chunk_strs): + continue + recomputed = hash_leaf(new_chunk_strs[i].encode("utf-8")).hex() + if recomputed != c["leaf_hash"]: + # Defensive: shouldn't happen if roots match, but bail safely. + continue + conn.execute( + "UPDATE chunks SET content = ?, tier = 'hot' " + "WHERE document_root = ? AND idx = ?", + (new_chunk_strs[i], document_root, i), + ) + conn.execute( + "INSERT INTO chunks_fts (document_root, idx, content) " + "VALUES (?, ?, ?)", + (document_root, i, new_chunk_strs[i]), + ) + restored += 1 + append_audit( + conn, + event_type="rehydrate_success", + subject_root=document_root, + body={"chunks_restored": restored, "uri": doc_row["document_uri"]}, + ) + + return {"status": "rehydrated", "chunks_restored": restored} diff --git a/aborist/ingest.py b/aborist/ingest.py new file mode 100644 index 0000000..e1cb3e0 --- /dev/null +++ b/aborist/ingest.py @@ -0,0 +1,220 @@ +"""Ingest pipeline: Source -> normalize -> chunk -> merkle -> upsert. + +Idempotent: re-ingesting the same Document is a no-op (document_root collision +is the upsert key). Edges with unresolved dst_root are stored with NULL and +backfilled later when the target document is ingested. +""" + +from __future__ import annotations + +import sqlite3 +import time +from dataclasses import dataclass + +from aborist import ( + CANONICALIZATION_VERSION, + CHUNKING_VERSION, + SCHEMA_VERSION, +) +from aborist.document import Document, canonicalize, get_chunker +from aborist.merkle import MerkleTree, hash_leaf +from aborist.source import Source +from aborist.store import append_audit, transaction + + +@dataclass +class IngestStats: + seen: int = 0 + inserted: int = 0 + skipped_duplicate: int = 0 + chunks_total: int = 0 + edges_total: int = 0 + + +def ingest_source( + conn: sqlite3.Connection, + source: Source, + chunker_name: str | None = None, + limit: int | None = None, +) -> IngestStats: + """Ingest every document the source yields. Returns counts.""" + chunker = get_chunker(chunker_name) + stats = IngestStats() + + for doc in source.iter_documents(): + stats.seen += 1 + if limit is not None and stats.seen > limit: + break + if _ingest_document(conn, doc, chunker): + stats.inserted += 1 + else: + stats.skipped_duplicate += 1 + + # Recompute chunk/edge totals once at end. + stats.chunks_total = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] + stats.edges_total = conn.execute("SELECT COUNT(*) FROM edges").fetchone()[0] + return stats + + +def _ingest_document( + conn: sqlite3.Connection, + doc: Document, + chunker, +) -> bool: + """Returns True if newly inserted, False if document_root already exists.""" + text = canonicalize(doc.content) + chunk_strs = chunker.split(text) + if not chunk_strs: + return False + + leaves = [hash_leaf(c.encode("utf-8")) for c in chunk_strs] + tree = MerkleTree.build(leaves) + document_root = tree.root.hex() + + with transaction(conn): + existing = conn.execute( + "SELECT 1 FROM documents WHERE document_root = ?", (document_root,) + ).fetchone() + if existing: + # Same content → same root → idempotent skip. Still backfill edges + # in case URI is new, but don't re-hash. + _upsert_edges(conn, document_root, doc) + return False + + ingest_ts = int(time.time()) + conn.execute( + "INSERT INTO documents " + "(document_root, document_uri, source_type, kind, compression_depth, " + " title, chunking_version, canonicalization_version, schema_version, " + " ingest_ts) " + "VALUES (?, ?, ?, 'surface', 0, ?, ?, ?, ?, ?)", + ( + document_root, + doc.uri, + doc.source_type, + doc.title, + chunker.name, + CANONICALIZATION_VERSION, + SCHEMA_VERSION, + ingest_ts, + ), + ) + + # Insert chunks (tier='hot' default). + conn.executemany( + "INSERT INTO chunks (document_root, idx, leaf_hash, content) " + "VALUES (?, ?, ?, ?)", + [ + (document_root, i, leaves[i].hex(), chunk_strs[i]) + for i in range(len(chunk_strs)) + ], + ) + + # Insert FTS rows. + conn.executemany( + "INSERT INTO chunks_fts (document_root, idx, content) VALUES (?, ?, ?)", + [(document_root, i, chunk_strs[i]) for i in range(len(chunk_strs))], + ) + + # Insert interior Merkle layers (skip layer 0 — it's chunks.leaf_hash). + rows: list[tuple] = [] + for layer_idx in range(1, len(tree.layers)): + for node_idx, h in enumerate(tree.layers[layer_idx]): + rows.append((document_root, layer_idx, node_idx, h.hex())) + if rows: + conn.executemany( + "INSERT INTO merkle_nodes (document_root, layer, idx, hash) " + "VALUES (?, ?, ?, ?)", + rows, + ) + + _upsert_edges(conn, document_root, doc) + + append_audit( + conn, + event_type="ingest", + subject_root=document_root, + body={ + "document_uri": doc.uri, + "source_type": doc.source_type, + "chunks": len(chunk_strs), + "chunking_version": chunker.name, + "canonicalization_version": CANONICALIZATION_VERSION, + "schema_version": SCHEMA_VERSION, + }, + ts=ingest_ts, + ) + + return True + + +def _upsert_edges(conn: sqlite3.Connection, src_root: str, doc: Document) -> None: + """Insert edges; backfill dst_root from documents table if URI matches. + + Unresolved edges store dst_root=''; resolved edges store the target's root. + """ + for e in doc.edges: + dst_root = e.dst_root or "" + if not dst_root: + row = conn.execute( + "SELECT document_root FROM documents WHERE document_uri = ? " + "ORDER BY ingest_ts ASC LIMIT 1", + (e.dst_uri,), + ).fetchone() + if row: + dst_root = row["document_root"] + conn.execute( + "INSERT OR IGNORE INTO edges " + "(src_root, dst_root, dst_uri, edge_type, anchor) VALUES (?, ?, ?, ?, ?)", + (src_root, dst_root, e.dst_uri or "", e.edge_type, e.anchor or ""), + ) + + +def verify_random_sample(conn: sqlite3.Connection, n: int = 10) -> dict: + """Sample N documents, regenerate Merkle proof for chunk 0, verify.""" + from aborist.merkle import MerkleProof, ProofNode, hash_leaf, verify_proof + + rows = conn.execute( + "SELECT document_root FROM documents ORDER BY RANDOM() LIMIT ?", (n,) + ).fetchall() + if not rows: + return {"sampled": 0, "passed": 0, "failed": 0} + + passed = 0 + failed = 0 + for row in rows: + document_root = row["document_root"] + chunk_rows = conn.execute( + "SELECT idx, leaf_hash, content FROM chunks " + "WHERE document_root = ? ORDER BY idx ASC", + (document_root,), + ).fetchall() + if not chunk_rows: + failed += 1 + continue + + # Rebuild the tree from leaves and verify chunk 0's proof against root. + leaves = [bytes.fromhex(r["leaf_hash"]) for r in chunk_rows] + # First content -> recompute leaf_hash to confirm content matches stored hash. + c0 = chunk_rows[0] + if c0["content"] is None: + # Cold tier: skip content-roundtrip but verify structural proof. + recomputed_leaf = leaves[0] + else: + recomputed_leaf = hash_leaf(c0["content"].encode("utf-8")) + if recomputed_leaf != leaves[0]: + failed += 1 + continue + + from aborist.merkle import MerkleTree + tree = MerkleTree.build(leaves) + if tree.root.hex() != document_root: + failed += 1 + continue + proof = tree.proof(0) + if verify_proof(proof) and proof.root.hex() == document_root: + passed += 1 + else: + failed += 1 + + return {"sampled": len(rows), "passed": passed, "failed": failed} diff --git a/aborist/merkle.py b/aborist/merkle.py new file mode 100644 index 0000000..0243f03 --- /dev/null +++ b/aborist/merkle.py @@ -0,0 +1,160 @@ +"""Merkle tree with non-commutative HashCombine. + +Python port of ~/git/proxy.unturf.com/pkg/verified/merkle.go conventions: + +- Domain separation via single-byte prefixes (leaf=0x00, node=0x03). +- HashCombine is non-commutative; sibling order matters always. +- Odd layers self-duplicate the trailing element (NOT zero-pad). +- Proof carries explicit IsLeft flag per sibling (NOT lexical sort). +- Empty tree root is ZeroHash (32 zero bytes). +""" + +from __future__ import annotations + +import hashlib +from dataclasses import dataclass, field +from typing import Iterable + +LEAF_PREFIX = b"\x00" +NODE_PREFIX = b"\x03" +ZERO_HASH = b"\x00" * 32 +HASH_LEN = 32 + + +def _sha256(*parts: bytes) -> bytes: + h = hashlib.sha256() + for p in parts: + h.update(p) + return h.digest() + + +def hash_leaf(content: bytes) -> bytes: + """Hash a leaf with domain prefix 0x00.""" + return _sha256(LEAF_PREFIX, content) + + +def hash_combine(left: bytes, right: bytes) -> bytes: + """Non-commutative interior combine with domain prefix 0x03.""" + if len(left) != HASH_LEN or len(right) != HASH_LEN: + raise ValueError("hash inputs must be 32 bytes") + return _sha256(NODE_PREFIX, left, right) + + +@dataclass(frozen=True) +class ProofNode: + """One sibling step in a Merkle inclusion proof. + + is_left=True means the sibling sits to the LEFT of the running hash, + so verification order is: HashCombine(sibling, current). + """ + + hash: bytes + is_left: bool + + +@dataclass(frozen=True) +class MerkleProof: + leaf: bytes + leaf_index: int + siblings: tuple[ProofNode, ...] + root: bytes + + +@dataclass +class MerkleTree: + """Layered tree. layers[0] = leaves, layers[-1] = [root].""" + + layers: list[list[bytes]] = field(default_factory=list) + + @property + def root(self) -> bytes: + if not self.layers or not self.layers[-1]: + return ZERO_HASH + return self.layers[-1][0] + + @property + def leaves(self) -> list[bytes]: + return self.layers[0] if self.layers else [] + + @classmethod + def build(cls, leaves: Iterable[bytes]) -> MerkleTree: + leaves = list(leaves) + if not leaves: + return cls(layers=[[]]) + layers: list[list[bytes]] = [list(leaves)] + current = list(leaves) + while len(current) > 1: + nxt: list[bytes] = [] + i = 0 + while i < len(current): + left = current[i] + right = current[i + 1] if i + 1 < len(current) else current[i] + nxt.append(hash_combine(left, right)) + i += 2 + layers.append(nxt) + current = nxt + return cls(layers=layers) + + def proof(self, leaf_index: int) -> MerkleProof: + if not self.layers or not self.layers[0]: + raise IndexError("empty tree has no proofs") + if leaf_index < 0 or leaf_index >= len(self.layers[0]): + raise IndexError(f"leaf_index {leaf_index} out of range") + + siblings: list[ProofNode] = [] + idx = leaf_index + # Walk up every layer except the root layer. + for layer in self.layers[:-1]: + if idx % 2 == 0: + sibling_idx = idx + 1 + is_left = False # sibling is to our right + else: + sibling_idx = idx - 1 + is_left = True # sibling is to our left + if sibling_idx >= len(layer): + # odd-element rule: self-duplicate + sibling_idx = idx + siblings.append(ProofNode(hash=layer[sibling_idx], is_left=is_left)) + idx //= 2 + + return MerkleProof( + leaf=self.layers[0][leaf_index], + leaf_index=leaf_index, + siblings=tuple(siblings), + root=self.root, + ) + + +def verify_proof(proof: MerkleProof) -> bool: + """Recompute root from leaf + sibling path. Returns True iff matches.""" + current = proof.leaf + for node in proof.siblings: + if node.is_left: + current = hash_combine(node.hash, current) + else: + current = hash_combine(current, node.hash) + return current == proof.root + + +def proof_to_dict(proof: MerkleProof) -> dict: + """JSON-serializable form for storage in providence_cache.merkle_proof.""" + return { + "leaf": proof.leaf.hex(), + "leaf_index": proof.leaf_index, + "siblings": [ + {"hash": s.hash.hex(), "is_left": s.is_left} for s in proof.siblings + ], + "root": proof.root.hex(), + } + + +def proof_from_dict(d: dict) -> MerkleProof: + return MerkleProof( + leaf=bytes.fromhex(d["leaf"]), + leaf_index=int(d["leaf_index"]), + siblings=tuple( + ProofNode(hash=bytes.fromhex(s["hash"]), is_left=bool(s["is_left"])) + for s in d["siblings"] + ), + root=bytes.fromhex(d["root"]), + ) diff --git a/aborist/search/__init__.py b/aborist/search/__init__.py new file mode 100644 index 0000000..383ee59 --- /dev/null +++ b/aborist/search/__init__.py @@ -0,0 +1,6 @@ +"""Search backends.""" + +from aborist.search.base import AuditMode, Hit, SearchBackend +from aborist.search.fts5 import FTS5Backend + +__all__ = ["AuditMode", "Hit", "SearchBackend", "FTS5Backend"] diff --git a/aborist/search/base.py b/aborist/search/base.py new file mode 100644 index 0000000..a9238bd --- /dev/null +++ b/aborist/search/base.py @@ -0,0 +1,47 @@ +"""Search backend ABC + Hit dataclass with explicit audit mode. + +Every search hit carries an `audit_mode` so callers never overclaim. Per +Merkle-AGI v7 §: +- STRICT — Merkle-verified, deterministic, full local reveal. Supports formal + claims (e.g., providence_cache hit with verified proof). +- HYBRID — random-challenge spot check (RCA(K)). Non-normative. +- VISUAL — exploration / debug only. No proof claim. Keyword search lives here. +""" + +from __future__ import annotations + +import enum +import sqlite3 +from abc import ABC, abstractmethod +from dataclasses import dataclass + + +class AuditMode(str, enum.Enum): + STRICT = "STRICT" + HYBRID = "HYBRID" + VISUAL = "VISUAL" + + +@dataclass(frozen=True) +class Hit: + document_root: str + document_uri: str + chunk_idx: int + snippet: str + score: float + audit_mode: AuditMode + title: str | None = None + + +class SearchBackend(ABC): + """A search hook over the chunk store.""" + + name: str + audit_mode: AuditMode # default mode this backend reports + + def __init__(self, conn: sqlite3.Connection): + self.conn = conn + + @abstractmethod + def search(self, query: str, limit: int = 20) -> list[Hit]: + ... diff --git a/aborist/search/fts5.py b/aborist/search/fts5.py new file mode 100644 index 0000000..5331d61 --- /dev/null +++ b/aborist/search/fts5.py @@ -0,0 +1,62 @@ +"""SQLite FTS5 keyword search. Returns VISUAL-mode hits (no proof claim).""" + +from __future__ import annotations + +from aborist.search.base import AuditMode, Hit, SearchBackend + + +def _escape_fts5(query: str) -> str: + """Wrap each token in double quotes and escape internal quotes. + + Prevents user-supplied FTS5 operators from breaking syntax. We accept + space-separated terms and AND them implicitly. + """ + tokens = [t for t in query.split() if t] + if not tokens: + return '""' + quoted = [] + for t in tokens: + # FTS5 doubles internal quotes to escape them. + quoted.append('"' + t.replace('"', '""') + '"') + return " ".join(quoted) + + +class FTS5Backend(SearchBackend): + name = "fts5" + audit_mode = AuditMode.VISUAL + + def search(self, query: str, limit: int = 20) -> list[Hit]: + if not query.strip(): + return [] + fts_query = _escape_fts5(query) + rows = self.conn.execute( + """ + SELECT + f.document_root, + f.idx, + snippet(chunks_fts, 2, '[', ']', '…', 16) AS snip, + bm25(chunks_fts) AS rank, + d.document_uri, + d.title + FROM chunks_fts AS f + JOIN documents AS d ON d.document_root = f.document_root + WHERE chunks_fts MATCH ? + ORDER BY rank ASC + LIMIT ? + """, + (fts_query, limit), + ).fetchall() + + return [ + Hit( + document_root=r["document_root"], + document_uri=r["document_uri"], + chunk_idx=r["idx"], + snippet=r["snip"] or "", + # bm25 returns negative numbers (lower = better); flip sign for sanity. + score=-float(r["rank"]) if r["rank"] is not None else 0.0, + audit_mode=self.audit_mode, + title=r["title"], + ) + for r in rows + ] diff --git a/aborist/source.py b/aborist/source.py new file mode 100644 index 0000000..f862ee2 --- /dev/null +++ b/aborist/source.py @@ -0,0 +1,24 @@ +"""Source ABC. + +Adding a new corpus to aborist = one new Source subclass. The Source contract +is intentionally minimal: yield Document objects, one at a time. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Iterator + +from aborist.document import Document + + +class Source(ABC): + """A corpus that yields documents into the ingest pipeline.""" + + #: source_type tag stored on every Document this source produces. + source_type: str + + @abstractmethod + def iter_documents(self) -> Iterator[Document]: + """Yield Document objects. Must be deterministic & idempotent.""" + ... diff --git a/aborist/sources/__init__.py b/aborist/sources/__init__.py new file mode 100644 index 0000000..0fa884e --- /dev/null +++ b/aborist/sources/__init__.py @@ -0,0 +1,13 @@ +"""Source implementations. Add a new corpus = add a new module here.""" + +from aborist.sources.wikipedia import WikipediaCurDump + +__all__ = ["WikipediaCurDump"] + +# HtmlPageSource has optional deps (httpx + selectolax). Surface import errors +# only when callers actually request it. +try: + from aborist.sources.html_page import HtmlPageSource # noqa: F401 + __all__.append("HtmlPageSource") +except ImportError: # pragma: no cover + pass diff --git a/aborist/sources/html_page.py b/aborist/sources/html_page.py new file mode 100644 index 0000000..f5b15f8 --- /dev/null +++ b/aborist/sources/html_page.py @@ -0,0 +1,146 @@ +"""HTML page source. + +Fetches URLs, honors robots.txt automatically, strips noise (script/style/nav/ +footer/header), extracts main body text + outbound `` links as edges. + +Optional dependency. Install with `pip install aborist[html]`. +""" + +from __future__ import annotations + +import re +import urllib.parse +import urllib.robotparser +from pathlib import Path +from typing import Iterable, Iterator + +try: + import httpx + from selectolax.parser import HTMLParser +except ImportError as e: # pragma: no cover + raise ImportError( + "HTML source requires extras: pip install 'aborist[html]'" + ) from e + +from aborist.document import Document, Edge +from aborist.source import Source + + +USER_AGENT = "aborist/0.0.1 (+https://unturf.com)" +NOISE_SELECTORS = ("script", "style", "noscript", "nav", "header", "footer", "aside") + + +def _normalize_text(text: str) -> str: + text = re.sub(r"[ \t]+", " ", text) + text = re.sub(r"\n{3,}", "\n\n", text) + return text.strip() + + +def parse_html(url: str, html: str, source_type: str = "html") -> Document | None: + """Pure parse function. Separated so tests can run without network.""" + tree = HTMLParser(html) + for sel in NOISE_SELECTORS: + for node in tree.css(sel): + node.decompose() + + body = tree.css_first("body") or tree.root + if body is None: + return None + text = _normalize_text(body.text(separator="\n", strip=True)) + if not text: + return None + + title_node = tree.css_first("title") + title = title_node.text(strip=True) if title_node is not None else None + + edges: list[Edge] = [] + seen: set[tuple[str, str]] = set() + for a in tree.css("a[href]"): + href = (a.attributes.get("href") or "").strip() + if not href or href.startswith(("javascript:", "mailto:", "tel:", "#")): + continue + absolute = urllib.parse.urljoin(url, href) + split = urllib.parse.urlsplit(absolute) + if split.scheme not in ("http", "https"): + continue + anchor = split.fragment or "" + dst_uri = urllib.parse.urlunsplit(split._replace(fragment="")) + key = (dst_uri, anchor) + if key in seen: + continue + seen.add(key) + edges.append(Edge(edge_type="hyperlink", dst_uri=dst_uri, anchor=anchor or None)) + + return Document( + uri=url, + content=text, + source_type=source_type, + title=title, + edges=edges, + ) + + +class HtmlPageSource(Source): + """Iterates a list of URLs, fetching and parsing each as HTML.""" + + source_type = "html" + + def __init__( + self, + urls: Iterable[str], + *, + respect_robots: bool = True, + timeout: float = 30.0, + ): + self.urls = list(urls) + self.respect_robots = respect_robots + self.timeout = timeout + self._robots_cache: dict[str, urllib.robotparser.RobotFileParser] = {} + + @classmethod + def from_file(cls, path: str | Path, **kwargs) -> HtmlPageSource: + urls = [ + line.strip() + for line in Path(path).read_text(encoding="utf-8").splitlines() + if line.strip() and not line.lstrip().startswith("#") + ] + return cls(urls, **kwargs) + + def iter_documents(self) -> Iterator[Document]: + with httpx.Client( + headers={"User-Agent": USER_AGENT}, + timeout=self.timeout, + follow_redirects=True, + ) as client: + for url in self.urls: + if self.respect_robots and not self._allowed(client, url): + continue + try: + resp = client.get(url) + resp.raise_for_status() + except httpx.HTTPError: + continue + ctype = resp.headers.get("content-type", "").lower() + if "html" not in ctype and "xml" not in ctype: + continue + doc = parse_html(str(resp.url), resp.text, self.source_type) + if doc is not None: + yield doc + + def _allowed(self, client: "httpx.Client", url: str) -> bool: + parsed = urllib.parse.urlparse(url) + origin = f"{parsed.scheme}://{parsed.netloc}" + rp = self._robots_cache.get(origin) + if rp is None: + rp = urllib.robotparser.RobotFileParser() + try: + resp = client.get(f"{origin}/robots.txt") + except httpx.HTTPError: + resp = None + if resp is not None and resp.status_code == 200: + rp.parse(resp.text.splitlines()) + else: + # Missing robots.txt = no rules per RFC 9309. + rp.allow_all = True + self._robots_cache[origin] = rp + return rp.can_fetch(USER_AGENT, url) diff --git a/aborist/sources/wikipedia.py b/aborist/sources/wikipedia.py new file mode 100644 index 0000000..9aee18b --- /dev/null +++ b/aborist/sources/wikipedia.py @@ -0,0 +1,223 @@ +"""MediaWiki 'cur' table SQL dump source. + +Handles 2003-era SQL dumps in bz2 format (e.g. 20030516_cur_tablesql.bz2). +Yields one Document per non-redirect main-namespace article, with +[[wikilinks]] extracted as outbound edges. + +Implements a stream parser for MySQL extended INSERT syntax. The cur table +schema for that era starts: cur_id, cur_namespace, cur_title, cur_text, ... +We rely on positional access for the first four columns. +""" + +from __future__ import annotations + +import bz2 +import re +from pathlib import Path +from typing import IO, Iterator + +from aborist.document import Document, Edge +from aborist.source import Source + + +# Match [[Target]], [[Target|display]], [[Target#anchor]] forms. +WIKILINK_RE = re.compile(r"\[\[([^\]\|#\n\r]+)(?:#[^\]\|\n\r]*)?(?:\|[^\]\n\r]*)?\]\]") +NAMESPACE_MAIN = 0 + + +def _decode_mysql_string(s: str) -> str: + """Decode MySQL-escaped string content (outer quotes already stripped).""" + out: list[str] = [] + i = 0 + n = len(s) + while i < n: + c = s[i] + if c == "\\" and i + 1 < n: + nxt = s[i + 1] + mapped = { + "n": "\n", + "r": "\r", + "t": "\t", + "0": "\0", + "\\": "\\", + "'": "'", + '"': '"', + "Z": "\x1a", + }.get(nxt, nxt) + out.append(mapped) + i += 2 + else: + out.append(c) + i += 1 + return "".join(out) + + +def _split_values_tuples(payload: str) -> list[list[str | None]]: + """Parse `(v1,v2,...),(...),...` into tuples of decoded strings or None.""" + rows: list[list[str | None]] = [] + i = 0 + n = len(payload) + while i < n: + while i < n and payload[i] in " \t\n\r,": + i += 1 + if i >= n or payload[i] != "(": + break + i += 1 + values: list[str | None] = [] + while True: + while i < n and payload[i] in " \t": + i += 1 + if i >= n: + break + if payload[i] == "'": + # Quoted string + i += 1 + buf: list[str] = [] + while i < n: + c = payload[i] + if c == "\\" and i + 1 < n: + # Preserve escape pair; decode later in one pass. + buf.append(c) + buf.append(payload[i + 1]) + i += 2 + elif c == "'": + i += 1 + break + else: + buf.append(c) + i += 1 + values.append(_decode_mysql_string("".join(buf))) + else: + # NULL or number + start = i + while i < n and payload[i] not in ",)": + i += 1 + v = payload[start:i].strip() + values.append(None if v.upper() == "NULL" else v) + while i < n and payload[i] in " \t": + i += 1 + if i < n and payload[i] == ",": + i += 1 + continue + if i < n and payload[i] == ")": + i += 1 + break + break + rows.append(values) + return rows + + +def _iter_insert_statements(file_obj: IO[str]) -> Iterator[str]: + """Yield complete SQL statements (text up to ; outside a quoted string).""" + buf: list[str] = [] + in_string = False + escape = False + while True: + chunk = file_obj.read(1 << 17) + if not chunk: + break + for c in chunk: + buf.append(c) + if escape: + escape = False + continue + if in_string: + if c == "\\": + escape = True + elif c == "'": + in_string = False + continue + if c == "'": + in_string = True + elif c == ";": + yield "".join(buf) + buf = [] + if buf: + tail = "".join(buf).strip() + if tail: + yield tail + + +def _extract_wikilinks(text: str, base_uri: str) -> list[Edge]: + seen: set[str] = set() + edges: list[Edge] = [] + for m in WIKILINK_RE.finditer(text): + target = m.group(1).strip() + if not target or target.startswith(":"): + continue + # Skip image / file / category interlinks (they often start "Image:" etc.) + if ":" in target: + continue + uri = base_uri + target.replace(" ", "_") + if uri in seen: + continue + seen.add(uri) + edges.append(Edge(edge_type="wikilink", dst_uri=uri)) + return edges + + +class WikipediaCurDump(Source): + """Iterates a MediaWiki 'cur' table SQL dump (bz2-compressed).""" + + source_type = "wikipedia_cur" + + def __init__( + self, + path: str | Path, + namespace: int = NAMESPACE_MAIN, + base_uri: str = "https://en.wikipedia.org/wiki/", + ): + self.path = Path(path) + self.namespace = namespace + self.base_uri = base_uri + + def iter_documents(self) -> Iterator[Document]: + opener = bz2.open if str(self.path).endswith(".bz2") else open + with opener(self.path, "rt", encoding="utf-8", errors="replace") as f: + for stmt in _iter_insert_statements(f): + head = stmt.lstrip() + if not head.upper().startswith("INSERT INTO"): + continue + up = head.upper() + vidx = up.find("VALUES") + if vidx < 0: + continue + table_clause = head[:vidx].lower() + if "cur" not in table_clause: + continue + payload = head[vidx + len("VALUES"):] + payload = payload.rstrip().rstrip(";").rstrip() + for row in _split_values_tuples(payload): + yield from self._row_to_doc(row) + + def _row_to_doc(self, row: list[str | None]) -> Iterator[Document]: + if len(row) < 4: + return + try: + ns = int(row[1]) if row[1] is not None else None + except (ValueError, TypeError): + return + if ns != self.namespace: + return + title = row[2] or "" + text = row[3] or "" + if not title or not text: + return + # is_redirect lives at position 10 in the standard 2003 schema; be defensive. + is_redirect = False + if len(row) > 10 and row[10] is not None: + try: + is_redirect = bool(int(row[10])) + except (ValueError, TypeError): + is_redirect = False + if is_redirect: + return + edges = _extract_wikilinks(text, self.base_uri) + uri = self.base_uri + title.replace(" ", "_") + yield Document( + uri=uri, + content=text, + source_type=self.source_type, + title=title, + edges=edges, + ) diff --git a/aborist/store.py b/aborist/store.py new file mode 100644 index 0000000..b6cb568 --- /dev/null +++ b/aborist/store.py @@ -0,0 +1,243 @@ +"""SQLite-backed v9.8 store. + +Schema implements the Merkle-AGI v9.8 admissibility ledger: +- 8-dim providence_cache key (source_root, question_hash, model_profile_hash, + conversation_hash, governance_policy_hash, schema_version, + canonicalization_version, chunking_version) +- falsification_state ∈ {live, failed, stale, quarantined} +- audit_events append-only chain (event_hash chains via prev_event_hash) +- documents.kind ∈ {surface, core} for layered compression +- chunks.tier ∈ {hot, warm, cold} for reversible eviction +- derivations table binds core docs back to source surface roots + +The providence_cache layer is schema-only in Phase 0 — no Q&A inference yet. +""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from contextlib import contextmanager +from pathlib import Path +from typing import Iterator + + +DEFAULT_DB_PATH = Path.home() / ".aborist" / "aborist.db" + + +SCHEMA_SQL = """ +PRAGMA journal_mode = WAL; +PRAGMA foreign_keys = ON; + +CREATE TABLE IF NOT EXISTS schema_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL +); + +-- Documents: surface (raw ingest) or core (distilled, Merkle-signed back). +CREATE TABLE IF NOT EXISTS documents ( + document_root TEXT PRIMARY KEY, -- hex sha256 of merkle root + document_uri TEXT NOT NULL, + source_type TEXT NOT NULL, + kind TEXT NOT NULL DEFAULT 'surface' + CHECK (kind IN ('surface','core')), + compression_depth INTEGER NOT NULL DEFAULT 0, + title TEXT, + chunking_version TEXT NOT NULL, + canonicalization_version TEXT NOT NULL, + schema_version TEXT NOT NULL, + ingest_ts INTEGER NOT NULL, + hit_count INTEGER NOT NULL DEFAULT 0, + last_hit_at INTEGER +); +CREATE INDEX IF NOT EXISTS idx_documents_uri ON documents(document_uri); +CREATE INDEX IF NOT EXISTS idx_documents_kind ON documents(kind); + +-- Chunks with tier-based reversible eviction. +-- content nullable: cold tier evicts content but retains leaf_hash + URI for +-- rehydration. Identity verified on rehydrate by recomputing leaf_hash. +CREATE TABLE IF NOT EXISTS chunks ( + document_root TEXT NOT NULL, + idx INTEGER NOT NULL, + leaf_hash TEXT NOT NULL, + content TEXT, + tier TEXT NOT NULL DEFAULT 'hot' + CHECK (tier IN ('hot','warm','cold')), + PRIMARY KEY (document_root, idx), + FOREIGN KEY (document_root) REFERENCES documents(document_root) ON DELETE CASCADE +); +CREATE INDEX IF NOT EXISTS idx_chunks_leaf ON chunks(leaf_hash); + +-- Interior Merkle nodes (layer >= 1). Layer 0 lives in chunks.leaf_hash. +CREATE TABLE IF NOT EXISTS merkle_nodes ( + document_root TEXT NOT NULL, + layer INTEGER NOT NULL, + idx INTEGER NOT NULL, + hash TEXT NOT NULL, + PRIMARY KEY (document_root, layer, idx), + FOREIGN KEY (document_root) REFERENCES documents(document_root) ON DELETE CASCADE +); + +-- Cross-links between documents (the forest). +-- Unresolved forward links (dst not yet ingested) carry dst_root='' and the +-- ingest pass backfills dst_root when the target appears. +CREATE TABLE IF NOT EXISTS edges ( + src_root TEXT NOT NULL, + dst_root TEXT NOT NULL DEFAULT '', -- '' = unresolved, backfilled later + dst_uri TEXT NOT NULL DEFAULT '', -- always present so we can resolve later + edge_type TEXT NOT NULL, -- wikilink, citation, derived_from, ... + anchor TEXT NOT NULL DEFAULT '', -- chunk index or fragment, '' if N/A + PRIMARY KEY (src_root, edge_type, dst_root, dst_uri, anchor) +); +CREATE INDEX IF NOT EXISTS idx_edges_dst_root ON edges(dst_root) WHERE dst_root <> ''; +CREATE INDEX IF NOT EXISTS idx_edges_dst_uri ON edges(dst_uri) WHERE dst_uri <> ''; + +-- Distillation: core_root <- src_root with Merkle-signed proof binding. +CREATE TABLE IF NOT EXISTS derivations ( + core_root TEXT NOT NULL, + src_root TEXT NOT NULL, + proof_blob TEXT NOT NULL, -- JSON merkle proof + process_id TEXT NOT NULL, -- distillation process identifier + distilled_at INTEGER NOT NULL, + PRIMARY KEY (core_root, src_root, process_id), + FOREIGN KEY (core_root) REFERENCES documents(document_root) ON DELETE CASCADE, + FOREIGN KEY (src_root) REFERENCES documents(document_root) ON DELETE CASCADE +); + +-- v9.8 providence cache: 8-dim admissibility key + falsification state. +-- Schema-only in Phase 0 (no Q&A runs yet); ready for Phase 1. +CREATE TABLE IF NOT EXISTS providence_cache ( + cache_key TEXT PRIMARY KEY, + source_root TEXT NOT NULL, + document_uri TEXT NOT NULL, + question_hash TEXT NOT NULL, + question_text TEXT NOT NULL, + answer_text TEXT NOT NULL, + merkle_proof TEXT NOT NULL, -- JSON + model_profile_hash TEXT NOT NULL, -- model_id + revision + quantization + conversation_hash TEXT NOT NULL, + governance_policy_hash TEXT NOT NULL, + schema_version TEXT NOT NULL, + canonicalization_version TEXT NOT NULL, + chunking_version TEXT NOT NULL, + falsification_state TEXT NOT NULL DEFAULT 'live' + CHECK (falsification_state IN ('live','failed','stale','quarantined')), + chain TEXT NOT NULL DEFAULT 'private' + CHECK (chain IN ('private','public')), + audit_event_hash TEXT, -- latest audit event for this record + created_at INTEGER NOT NULL, + last_hit_at INTEGER, + hit_count INTEGER NOT NULL DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_providence_root ON providence_cache(source_root); +CREATE INDEX IF NOT EXISTS idx_providence_state ON providence_cache(falsification_state); + +-- Append-only audit chain. event_hash = sha256(prev_event_hash || canonical(body)). +CREATE TABLE IF NOT EXISTS audit_events ( + seq INTEGER PRIMARY KEY AUTOINCREMENT, + event_hash TEXT NOT NULL UNIQUE, + prev_event_hash TEXT, -- NULL for genesis + event_type TEXT NOT NULL, -- ingest|falsify|evict_warm|evict_cold|derive|rehydrate|... + subject_root TEXT, -- document_root or cache_key + body TEXT NOT NULL, -- canonical JSON + ts INTEGER NOT NULL +); +CREATE INDEX IF NOT EXISTS idx_audit_subject ON audit_events(subject_root); + +-- Falsification log: which records were marked failed/stale/quarantined and why. +CREATE TABLE IF NOT EXISTS falsifications ( + cache_key TEXT NOT NULL, + state TEXT NOT NULL, + reason TEXT, + by_actor TEXT, + at INTEGER NOT NULL, + audit_event_hash TEXT NOT NULL, + PRIMARY KEY (cache_key, at) +); + +-- FTS5 over chunk content for VISUAL-mode keyword search. +CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5( + document_root UNINDEXED, + idx UNINDEXED, + content, + tokenize = 'porter unicode61' +); +""" + + +def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: + """Open a connection, creating the parent dir and applying schema if needed.""" + p = Path(db_path) + p.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(p, isolation_level=None) # autocommit; we'll BEGIN manually + conn.row_factory = sqlite3.Row + conn.executescript(SCHEMA_SQL) + return conn + + +@contextmanager +def transaction(conn: sqlite3.Connection) -> Iterator[sqlite3.Connection]: + """BEGIN IMMEDIATE / COMMIT / ROLLBACK around a block.""" + conn.execute("BEGIN IMMEDIATE") + try: + yield conn + except Exception: + conn.execute("ROLLBACK") + raise + else: + conn.execute("COMMIT") + + +def _canonical_json(obj) -> str: + """Stable JSON for audit hashing: sorted keys, no whitespace.""" + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def append_audit( + conn: sqlite3.Connection, + event_type: str, + body: dict, + subject_root: str | None = None, + ts: int | None = None, +) -> str: + """Append one event to the audit chain. Returns the new event_hash (hex).""" + import hashlib + + if ts is None: + ts = int(time.time()) + row = conn.execute( + "SELECT event_hash FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + prev = row["event_hash"] if row else None + body_json = _canonical_json(body) + h = hashlib.sha256() + if prev is not None: + h.update(bytes.fromhex(prev)) + h.update(body_json.encode("utf-8")) + event_hash = h.hexdigest() + conn.execute( + "INSERT INTO audit_events (event_hash, prev_event_hash, event_type, subject_root, body, ts) " + "VALUES (?, ?, ?, ?, ?, ?)", + (event_hash, prev, event_type, subject_root, body_json, ts), + ) + return event_hash + + +def stats(conn: sqlite3.Connection) -> dict: + """Quick landscape report.""" + def one(sql: str, *args) -> int: + return conn.execute(sql, args).fetchone()[0] + + return { + "documents_total": one("SELECT COUNT(*) FROM documents"), + "documents_surface": one("SELECT COUNT(*) FROM documents WHERE kind='surface'"), + "documents_core": one("SELECT COUNT(*) FROM documents WHERE kind='core'"), + "chunks_total": one("SELECT COUNT(*) FROM chunks"), + "chunks_hot": one("SELECT COUNT(*) FROM chunks WHERE tier='hot'"), + "chunks_warm": one("SELECT COUNT(*) FROM chunks WHERE tier='warm'"), + "chunks_cold": one("SELECT COUNT(*) FROM chunks WHERE tier='cold'"), + "edges_total": one("SELECT COUNT(*) FROM edges"), + "providence_total": one("SELECT COUNT(*) FROM providence_cache"), + "audit_events_total": one("SELECT COUNT(*) FROM audit_events"), + } diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e874e39 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,30 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "aborist" +version = "0.0.1" +description = "An arborist for trees and forests of cross-linked information" +readme = "README.md" +license = { text = "AGPL-3.0-only" } +requires-python = ">=3.10" +authors = [{ name = "Russell Ballestrini", email = "russell@unturf.com" }] +dependencies = [] + +[project.optional-dependencies] +html = [ + "httpx>=0.27", + "selectolax>=0.3", +] +dev = [ + "pytest>=8", + "aborist[html]", +] + +[project.scripts] +aborist = "aborist.cli:main" + +[tool.setuptools.packages.find] +where = ["."] +include = ["aborist*"] diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_distill.py b/tests/test_distill.py new file mode 100644 index 0000000..2c5e04d --- /dev/null +++ b/tests/test_distill.py @@ -0,0 +1,132 @@ +"""Distillation: cores must Merkle-bind back to their source surface roots.""" + +from __future__ import annotations + +import json +from typing import Iterator + +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.merkle import proof_from_dict, verify_proof +from aborist.source import Source +from aborist.store import connect + + +class FakeSource(Source): + source_type = "test" + + 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) -> Document: + return Document(uri=uri, content=content, source_type="test", title=uri) + + +# A long, multi-paragraph source so the chunker produces multiple chunks. +LONG_TEXT = ( + "The quick brown fox jumps over the lazy dog. " * 30 + + "\n\n" + + "Merkle providence proves answer derives from a specific source. " * 30 + + "\n\n" + + "The eight forms of capital include living, social, and intellectual. " * 30 +) + + +def test_distillation_produces_core_with_verifiable_proofs(tmp_path): + db = tmp_path / "distill.db" + src_doc = _doc("test://long", LONG_TEXT) + conn = connect(db) + try: + ingest_source(conn, FakeSource([src_doc])) + result = distill_existing(conn, FirstSentenceDistiller()) + + assert result["distilled"] == 1 + assert result["skipped_existing"] == 0 + + # Core document exists and is marked correctly. + cores = conn.execute( + "SELECT * FROM documents WHERE kind = 'core'" + ).fetchall() + assert len(cores) == 1 + core = cores[0] + assert core["compression_depth"] == 1 + assert core["source_type"] == "core:first-sentence-v1" + + # Derivation row binds core to source. + derivs = conn.execute("SELECT * FROM derivations").fetchall() + assert len(derivs) == 1 + d = derivs[0] + assert d["core_root"] == core["document_root"] + assert d["process_id"] == "first-sentence-v1" + + # CRITICAL: every contributing-chunk Merkle proof in proof_blob must + # reconstruct the source's document_root. This is the cryptographic + # binding of compressed core back to surface. + proof_data = json.loads(d["proof_blob"]) + assert proof_data["core_root"] == core["document_root"] + src_root = proof_data["src_root"] + assert len(proof_data["contributing"]) >= 1 + for entry in proof_data["contributing"]: + proof = proof_from_dict(entry["proof"]) + assert verify_proof(proof), f"proof invalid for chunk {entry['src_chunk_idx']}" + assert proof.root.hex() == src_root, ( + "contributing-chunk proof must reconstruct source root" + ) + + # derived_from edge in the forest. + edges = conn.execute( + "SELECT * FROM edges WHERE edge_type = 'derived_from'" + ).fetchall() + assert len(edges) == 1 + assert edges[0]["src_root"] == core["document_root"] + assert edges[0]["dst_root"] == src_root + + # Audit chain has both ingest and derive events. + events = conn.execute( + "SELECT event_type FROM audit_events ORDER BY seq ASC" + ).fetchall() + types = [r["event_type"] for r in events] + assert "ingest" in types and "derive" in types + finally: + conn.close() + + +def test_distill_idempotent(tmp_path): + db = tmp_path / "idempotent.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("test://x", LONG_TEXT)])) + d = FirstSentenceDistiller() + first = distill_existing(conn, d) + second = distill_existing(conn, d) + assert first["distilled"] == 1 + assert second["distilled"] == 0 + assert second["skipped_existing"] == 1 + # Still only one core document. + n_cores = conn.execute( + "SELECT COUNT(*) FROM documents WHERE kind='core'" + ).fetchone()[0] + assert n_cores == 1 + finally: + conn.close() + + +def test_distill_skips_cold_chunks(tmp_path): + """If any source chunk has been evicted (content NULL), we must skip.""" + db = tmp_path / "cold.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("test://cold", LONG_TEXT)])) + # Manually evict one chunk to cold tier. + conn.execute("UPDATE chunks SET content=NULL, tier='cold' WHERE idx=0") + result = distill_existing(conn, FirstSentenceDistiller()) + assert result["distilled"] == 0 + assert result["skipped_cold"] == 1 + finally: + conn.close() diff --git a/tests/test_evict.py b/tests/test_evict.py new file mode 100644 index 0000000..7724c95 --- /dev/null +++ b/tests/test_evict.py @@ -0,0 +1,254 @@ +"""Eviction & rehydrate: lossless reversible forgetting.""" + +from __future__ import annotations + +import json +from typing import Iterator + +from aborist.distill import FirstSentenceDistiller +from aborist.distill.runner import distill_existing +from aborist.document import Document +from aborist.evict import evict_to_cold, rehydrate +from aborist.ingest import ingest_source +from aborist.source import Source +from aborist.store import connect + + +class FakeSource(Source): + source_type = "html" # so rehydrate path treats these as html-like + + 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) -> Document: + return Document(uri=uri, content=content, source_type="html", title=uri) + + +LONG = ( + "The eight forms of capital include living, social, and intellectual. " * 30 + + "\n\n" + + "Merkle providence proves answer derives from a specific source. " * 30 +) + + +def test_evict_marks_chunks_cold_and_clears_fts(tmp_path): + db = tmp_path / "evict.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("html://a", LONG)])) + # Before: hot chunks, FTS rows present + before = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE tier='hot'" + ).fetchone()[0] + before_fts = conn.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] + assert before > 0 + assert before_fts == before + + result = evict_to_cold(conn) + assert result["evicted_chunks"] == before + assert result["documents_affected"] == 1 + + cold = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE tier='cold'" + ).fetchone()[0] + assert cold == before + # Content NULLed + nulls = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE content IS NULL" + ).fetchone()[0] + assert nulls == before + # leaf_hash retained + no_hash = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE leaf_hash IS NULL OR leaf_hash = ''" + ).fetchone()[0] + assert no_hash == 0 + # FTS rows removed + fts_after = conn.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] + assert fts_after == 0 + finally: + conn.close() + + +def test_evict_skips_cores(tmp_path): + db = tmp_path / "cores.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("html://x", LONG)])) + distill_existing(conn, FirstSentenceDistiller()) + # Verify there's a core + n_cores = conn.execute( + "SELECT COUNT(*) FROM documents WHERE kind='core'" + ).fetchone()[0] + assert n_cores == 1 + + evict_to_cold(conn) + + # All surface chunks are cold; core chunks still hot. + surface_tiers = conn.execute( + "SELECT DISTINCT c.tier FROM chunks c " + "JOIN documents d ON d.document_root=c.document_root " + "WHERE d.kind='surface'" + ).fetchall() + assert {r["tier"] for r in surface_tiers} == {"cold"} + + core_tiers = conn.execute( + "SELECT DISTINCT c.tier FROM chunks c " + "JOIN documents d ON d.document_root=c.document_root " + "WHERE d.kind='core'" + ).fetchall() + assert {r["tier"] for r in core_tiers} == {"hot"} + finally: + conn.close() + + +def test_rehydrate_restores_content_when_uri_matches(tmp_path): + """Mock fetcher returns the same canonicalized text — root matches → restore.""" + db = tmp_path / "rh.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("html://stable", LONG)])) + evict_to_cold(conn) + nulls_before = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE content IS NULL" + ).fetchone()[0] + assert nulls_before > 0 + + # Inject a fetcher that returns the original text verbatim. + def fake_fetch(uri: str) -> str: + return LONG + + root = conn.execute( + "SELECT document_root FROM documents WHERE document_uri='html://stable'" + ).fetchone()["document_root"] + + result = rehydrate(conn, root, fetcher=fake_fetch) + assert result["status"] == "rehydrated" + assert result["chunks_restored"] > 0 + + nulls_after = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE content IS NULL" + ).fetchone()[0] + assert nulls_after == 0 + # All restored chunks tier=hot, FTS repopulated. + cold = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE tier='cold'" + ).fetchone()[0] + assert cold == 0 + finally: + conn.close() + + +def test_rehydrate_drift_marks_providence_stale(tmp_path): + """If URI's content has changed, leaves don't match → drift event + + every providence_cache row for this source flips to 'stale'.""" + import time + + db = tmp_path / "drift.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("html://drifty", LONG)])) + root = conn.execute( + "SELECT document_root FROM documents WHERE document_uri='html://drifty'" + ).fetchone()["document_root"] + + # Insert a fake live providence record bound to this source. + 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, created_at) " + "VALUES (?, ?, 'html://drifty', 'q1', 'qtxt', 'atxt', '{}', 'm1', 'c1', " + "'g1', 'v9.8.0', 'norm-v1', 'tok-512-v1', 'live', ?)", + (root + ":q1", root, int(time.time())), + ) + + evict_to_cold(conn) + + # Fetcher returns DIFFERENT text — drift. + def drifted_fetch(uri: str) -> str: + return LONG + "\n\nNEW PARAGRAPH ADDED AFTER INGEST." + + result = rehydrate(conn, root, fetcher=drifted_fetch) + assert result["status"] == "drift_detected" + assert result["expected_root"] == root + assert result["actual_root"] != root + + # No chunk content was restored. + cold = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE tier='cold'" + ).fetchone()[0] + assert cold > 0 + + # Providence record flipped to stale. + state = conn.execute( + "SELECT falsification_state FROM providence_cache WHERE source_root=?", + (root,), + ).fetchone()["falsification_state"] + assert state == "stale" + + # Drift event recorded in audit chain. + last = conn.execute( + "SELECT event_type FROM audit_events ORDER BY seq DESC LIMIT 1" + ).fetchone() + assert last["event_type"] == "rehydrate_drift" + finally: + conn.close() + + +def test_rehydrate_unknown_document(tmp_path): + db = tmp_path / "u.db" + conn = connect(db) + try: + result = rehydrate(conn, "00" * 32) + assert result["status"] == "unknown_document" + finally: + conn.close() + + +def test_rehydrate_nothing_to_do_when_all_hot(tmp_path): + db = tmp_path / "n.db" + conn = connect(db) + try: + ingest_source(conn, FakeSource([_doc("html://hot", LONG)])) + root = conn.execute( + "SELECT document_root FROM documents WHERE document_uri='html://hot'" + ).fetchone()["document_root"] + result = rehydrate(conn, root, fetcher=lambda u: LONG) + assert result["status"] == "nothing_to_do" + finally: + conn.close() + + +def test_rehydrate_non_rehydratable_source(tmp_path): + """A source_type without a registered fetcher (and no override) is honest about it.""" + from aborist.source import Source + + class WikiSource(Source): + source_type = "wikipedia_cur" + + def iter_documents(self): + yield Document( + uri="https://en.wikipedia.org/wiki/Foo", + content=LONG, + source_type="wikipedia_cur", + title="Foo", + ) + + db = tmp_path / "ns.db" + conn = connect(db) + try: + ingest_source(conn, WikiSource()) + root = conn.execute( + "SELECT document_root FROM documents LIMIT 1" + ).fetchone()["document_root"] + evict_to_cold(conn) + result = rehydrate(conn, root) # no fetcher override + assert result["status"] == "source_not_rehydratable" + assert result["source_type"] == "wikipedia_cur" + finally: + conn.close() diff --git a/tests/test_html_source.py b/tests/test_html_source.py new file mode 100644 index 0000000..b31e459 --- /dev/null +++ b/tests/test_html_source.py @@ -0,0 +1,83 @@ +"""Pure-parse tests for HtmlPageSource. No network.""" + +from __future__ import annotations + +import pytest + +# Skip if optional extras are not installed. +selectolax = pytest.importorskip("selectolax") + +from aborist.sources.html_page import parse_html + + +SAMPLE_HTML = """ + + + Eight Forms of Capital + + +

Site nav we want stripped

+ +
+

Eight Forms of Capital

+

Living, Material, Financial, Intellectual.

+

Experiential, Social, Cultural, Spiritual.

+

See also this page and + an external link.

+ js link + mail link +
+
strip me too
+ + + +""" + + +def test_parse_extracts_title_and_body(): + doc = parse_html("https://example.com/page", SAMPLE_HTML) + assert doc is not None + assert doc.title == "Eight Forms of Capital" + assert "Living" in doc.content + assert "Spiritual" in doc.content + + +def test_parse_strips_noise(): + doc = parse_html("https://example.com/page", SAMPLE_HTML) + assert doc is not None + assert "console.log" not in doc.content + assert "color: red" not in doc.content + assert "Site nav we want stripped" not in doc.content + assert "strip me too" not in doc.content + assert "nav links also stripped" not in doc.content + + +def test_parse_extracts_edges_and_resolves_relative(): + doc = parse_html("https://example.com/page", SAMPLE_HTML) + assert doc is not None + uris = {e.dst_uri for e in doc.edges} + # Relative href resolved against the page URL. + assert "https://example.com/eight-forms-of-capital/" in uris + # External link kept; fragment moved to anchor. + assert "https://example.org/external" in uris + # javascript: / mailto: / tel: / hash-only anchors must be skipped. + assert all("javascript" not in u for u in uris) + assert all("mailto" not in u for u in uris) + + +def test_parse_anchor_split(): + doc = parse_html("https://example.com/page", SAMPLE_HTML) + assert doc is not None + external = next(e for e in doc.edges if e.dst_uri == "https://example.org/external") + assert external.anchor == "section" + + +def test_parse_empty_body_returns_none(): + doc = parse_html("https://example.com/empty", "") + assert doc is None + + +def test_parse_no_html_returns_none_or_empty(): + # Selectolax tolerates non-HTML; we want no Document for empty content. + doc = parse_html("https://example.com/x", "") + assert doc is None diff --git a/tests/test_ingest.py b/tests/test_ingest.py new file mode 100644 index 0000000..7d36549 --- /dev/null +++ b/tests/test_ingest.py @@ -0,0 +1,119 @@ +"""End-to-end ingest test using a hand-rolled in-memory Source.""" + +from __future__ import annotations + +from typing import Iterator + +from aborist.document import Document, Edge +from aborist.ingest import ingest_source, verify_random_sample +from aborist.search import FTS5Backend +from aborist.search.base import AuditMode +from aborist.source import Source +from aborist.store import connect, stats + + +class FakeSource(Source): + source_type = "fake" + + 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: list[Edge] | None = None) -> Document: + return Document( + uri=uri, + content=content, + source_type="fake", + title=uri.rsplit("/", 1)[-1], + edges=edges or [], + ) + + +def test_ingest_basic_round_trip(tmp_path): + db_path = tmp_path / "test.db" + src = FakeSource([ + _doc("test://a", "alpha bravo charlie delta echo foxtrot golf hotel"), + _doc("test://b", "the quick brown fox jumps over the lazy dog"), + _doc( + "test://c", + "merkle providence reverse rag verifies provenance", + edges=[Edge(edge_type="wikilink", dst_uri="test://a")], + ), + ]) + conn = connect(db_path) + try: + result = ingest_source(conn, src) + assert result.seen == 3 + assert result.inserted == 3 + assert result.skipped_duplicate == 0 + + # Verify Merkle round-trip. + v = verify_random_sample(conn, n=3) + assert v["sampled"] == 3 + assert v["passed"] == 3 + assert v["failed"] == 0 + + # Idempotent re-ingest. + result2 = ingest_source(conn, src) + assert result2.inserted == 0 + assert result2.skipped_duplicate == 3 + + # FTS5 search returns VISUAL hits. + backend = FTS5Backend(conn) + hits = backend.search("merkle") + assert len(hits) >= 1 + assert hits[0].audit_mode == AuditMode.VISUAL + assert "merkle" in hits[0].snippet.lower() + + # Edge resolution: c -> a should be backfilled (a was ingested first). + row = conn.execute( + "SELECT dst_root FROM edges WHERE dst_uri = ?", ("test://a",) + ).fetchone() + assert row is not None + assert row["dst_root"] != "" # backfilled (was '' before resolution) + + # Stats reflect ingest. + s = stats(conn) + assert s["documents_total"] == 3 + assert s["documents_surface"] == 3 + assert s["documents_core"] == 0 + assert s["chunks_total"] >= 3 + assert s["audit_events_total"] == 3 # one ingest event per doc + finally: + conn.close() + + +def test_audit_chain_links_correctly(tmp_path): + """Each audit event chains to the previous via prev_event_hash.""" + db_path = tmp_path / "audit.db" + src = FakeSource([_doc(f"test://{i}", f"document number {i} content") for i in range(5)]) + conn = connect(db_path) + try: + ingest_source(conn, src) + rows = conn.execute( + "SELECT seq, event_hash, prev_event_hash FROM audit_events ORDER BY seq" + ).fetchall() + assert len(rows) == 5 + assert rows[0]["prev_event_hash"] is None # genesis + for i in range(1, len(rows)): + assert rows[i]["prev_event_hash"] == rows[i - 1]["event_hash"] + finally: + conn.close() + + +def test_chunker_version_persisted(tmp_path): + db_path = tmp_path / "chunker.db" + conn = connect(db_path) + try: + ingest_source(conn, FakeSource([_doc("test://x", "alpha beta gamma")])) + row = conn.execute( + "SELECT chunking_version, canonicalization_version, schema_version FROM documents" + ).fetchone() + assert row["chunking_version"] == "tok-512-v1" + assert row["canonicalization_version"] == "norm-v1" + assert row["schema_version"] == "v9.8.0" + finally: + conn.close() diff --git a/tests/test_merkle.py b/tests/test_merkle.py new file mode 100644 index 0000000..63b18c3 --- /dev/null +++ b/tests/test_merkle.py @@ -0,0 +1,155 @@ +"""Merkle round-trip and tamper-detection tests. + +These exercise the proxy.unturf.com Go-merkle conventions ported into Python: +- non-commutative HashCombine (0x03 prefix) +- self-duplicate odd elements +- explicit IsLeft per sibling (no lexical ordering) +""" + +from __future__ import annotations + +import pytest + +from aborist.merkle import ( + HASH_LEN, + MerkleTree, + ZERO_HASH, + hash_combine, + hash_leaf, + proof_from_dict, + proof_to_dict, + verify_proof, +) + + +def _leaf(s: str) -> bytes: + return hash_leaf(s.encode("utf-8")) + + +def test_empty_tree_root_is_zero_hash(): + tree = MerkleTree.build([]) + assert tree.root == ZERO_HASH + + +def test_single_leaf_root_equals_leaf(): + leaves = [_leaf("only-chunk")] + tree = MerkleTree.build(leaves) + # No interior layer needed; root is the single leaf itself. + assert tree.root == leaves[0] + + +def test_combine_is_non_commutative(): + a = _leaf("a") + b = _leaf("b") + assert hash_combine(a, b) != hash_combine(b, a) + + +def test_two_leaves_round_trip(): + leaves = [_leaf("alpha"), _leaf("beta")] + tree = MerkleTree.build(leaves) + expected_root = hash_combine(leaves[0], leaves[1]) + assert tree.root == expected_root + + for i in range(len(leaves)): + proof = tree.proof(i) + assert verify_proof(proof) + assert proof.root == tree.root + + +def test_three_leaves_self_duplicate_odd(): + leaves = [_leaf("a"), _leaf("b"), _leaf("c")] + tree = MerkleTree.build(leaves) + # Layer 1: combine(a,b), combine(c,c) + n01 = hash_combine(leaves[0], leaves[1]) + n22 = hash_combine(leaves[2], leaves[2]) + expected_root = hash_combine(n01, n22) + assert tree.root == expected_root + + for i in range(3): + proof = tree.proof(i) + assert verify_proof(proof), f"proof for index {i} should verify" + + +def test_four_leaves_full_round_trip(): + leaves = [_leaf(f"chunk-{i}") for i in range(4)] + tree = MerkleTree.build(leaves) + for i in range(4): + proof = tree.proof(i) + assert verify_proof(proof) + assert len(proof.siblings) == 2 # log2(4) = 2 levels + + +def test_seven_leaves_full_round_trip(): + """Odd intermediate layers self-duplicate; every proof must still verify.""" + leaves = [_leaf(f"chunk-{i}") for i in range(7)] + tree = MerkleTree.build(leaves) + for i in range(7): + proof = tree.proof(i) + assert verify_proof(proof), f"proof for index {i} should verify" + + +def test_proof_serialization_round_trip(): + leaves = [_leaf(f"x-{i}") for i in range(5)] + tree = MerkleTree.build(leaves) + proof = tree.proof(2) + d = proof_to_dict(proof) + restored = proof_from_dict(d) + assert restored == proof + assert verify_proof(restored) + + +def test_tampered_leaf_fails_verification(): + leaves = [_leaf(f"chunk-{i}") for i in range(4)] + tree = MerkleTree.build(leaves) + proof = tree.proof(1) + # Tamper: replace the leaf bytes. + bad_leaf = _leaf("not-the-real-chunk") + bad_proof = type(proof)( + leaf=bad_leaf, + leaf_index=proof.leaf_index, + siblings=proof.siblings, + root=proof.root, + ) + assert not verify_proof(bad_proof) + + +def test_tampered_sibling_fails_verification(): + leaves = [_leaf(f"chunk-{i}") for i in range(4)] + tree = MerkleTree.build(leaves) + proof = tree.proof(0) + from aborist.merkle import ProofNode + + bad_siblings = list(proof.siblings) + s0 = bad_siblings[0] + bad_siblings[0] = ProofNode(hash=b"\xff" * HASH_LEN, is_left=s0.is_left) + bad_proof = type(proof)( + leaf=proof.leaf, + leaf_index=proof.leaf_index, + siblings=tuple(bad_siblings), + root=proof.root, + ) + assert not verify_proof(bad_proof) + + +def test_swapped_is_left_flag_fails(): + """Order must be preserved — flipping the IsLeft flag must reject.""" + leaves = [_leaf(f"chunk-{i}") for i in range(4)] + tree = MerkleTree.build(leaves) + proof = tree.proof(1) + from aborist.merkle import MerkleProof, ProofNode + + flipped = MerkleProof( + leaf=proof.leaf, + leaf_index=proof.leaf_index, + siblings=tuple( + ProofNode(hash=s.hash, is_left=not s.is_left) for s in proof.siblings + ), + root=proof.root, + ) + assert not verify_proof(flipped) + + +def test_out_of_range_leaf_index(): + tree = MerkleTree.build([_leaf("x"), _leaf("y")]) + with pytest.raises(IndexError): + tree.proof(5)