From 28feee3efb9deb510fa421d983ebd5574c7e5a91 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Wed, 29 Apr 2026 18:44:17 -0400 Subject: [PATCH] =?UTF-8?q?qa:=20per-run=20Merkle-DAG=20provenance=20?= =?UTF-8?q?=E2=80=94=20run=5Fdag=5Froot=20on=20every=20providence=20record?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each query/ask call now emits a 7-stage Merkle-DAG fingerprint stored alongside the providence record. F from the toy-Hermes design pass. Stages, in order: question hash of question_hash (the 8-dim cache_key dim) retrieval hash of sources summary (document_roots + roles + scores + chunk_idx) — captures which docs ranked context context_root (the source-Merkle for the assembly) prompt conversation_hash answer sha256(answer_text) verify hash of verdict (audit_mode, verifier_method, n_quotes, n_verified, claim_statuses) final_label hash of (audit_mode, verifier_method, lookup_path) run_dag_root = MerkleTree over those stage hashes (aborist conventions: non-commutative HashCombine 0x03, leaf prefix 0x00, self-dup odd rule). run_dag_blob = canonical JSON of {root, nodes} so an auditor can recompute & verify (`verify_run_dag(blob)` returns True/False). The DAG is NOT in cache_key. cache_key inputs determine the answer; the answer determines the DAG — folding it back would create a cycle. Instead it rides alongside as a per-record computation fingerprint. Distinct from the linear `audit_events` chain (which tracks DB-wide state changes); this is per-run computation provenance. Schema: ALTER TABLE providence_cache ADD COLUMN run_dag_root TEXT; ADD COLUMN run_dag_blob TEXT; Idempotent migration in `_migrate_audit_mode`. Both rebuild templates (VISUAL→UNGROUNDED dance, paraphrase verifier_method dance) updated to include the new columns. Legacy records pre-2026-04-30 carry NULL. Result dict gains `run_dag_root` so callers can verify without a DB round-trip. Tests: - test_dag.py (9 tests): determinism, reactivity to each stage's input, fixed stage order, round-trip verify, tamper-detection, JSON-string acceptance. - test_query.py: persistence on record + result, verify_run_dag round- trip on the persisted blob. 473 tests pass (DAG +9, query +1, integration unchanged). --- aborist/qa/dag.py | 128 +++++++++++++++++++++++++++++++++++++++++++ aborist/qa/query.py | 30 +++++++++- aborist/qa/runner.py | 31 ++++++++++- aborist/store.py | 25 ++++++++- tests/test_dag.py | 117 +++++++++++++++++++++++++++++++++++++++ tests/test_query.py | 38 +++++++++++++ 6 files changed, 362 insertions(+), 7 deletions(-) create mode 100644 aborist/qa/dag.py create mode 100644 tests/test_dag.py diff --git a/aborist/qa/dag.py b/aborist/qa/dag.py new file mode 100644 index 0000000..ba4c20c --- /dev/null +++ b/aborist/qa/dag.py @@ -0,0 +1,128 @@ +"""Per-run Merkle-DAG provenance for providence records. + +Each query/ask call passes through several stages: + + question → retrieval → context → prompt → answer → verify → final_label + +Each stage emits a hash; the run's identity is the Merkle root over the +ordered sequence of stage hashes. Stored on the providence record as +``run_dag_root`` (alongside ``cache_key``). The DAG is verifiable: given +the persisted node list & the same Merkle conventions aborist uses +elsewhere (non-commutative HashCombine, prefix 0x03, leaf prefix 0x00, +self-duplicate odd rule), an auditor can recompute the root from the +nodes & confirm the run was constructed as recorded. + +Distinct from the linear ``audit_events`` chain — that chain tracks +state-changing operations across the DB. This DAG tracks the +computation provenance of one specific answer. Both coexist; the +record's ``audit_event_hash`` links to the chain, ``run_dag_root`` & +``run_dag_blob`` carry the per-run computation graph. + +Stages chosen to mirror the toy-Hermes design (fox 2026-04-30): + + question hash of question_hash (8-dim cache_key dim) + retrieval hash of sources summary (document_roots + roles + + scores) — captures which docs ranked & how + context context_root (Merkle root over sorted source roots, + the "source" dim of the cache_key) + prompt conversation_hash (the assembled messages) + answer sha256(answer_text) + verify hash of verdict summary (audit_mode, verifier_method, + n_quotes, n_verified, claim_statuses) + final_label hash of (audit_mode, verifier_method, lookup_path) + +The DAG is NOT part of cache_key. cache_key inputs (the 8 dims) +determine the answer; the answer determines the DAG. Folding the DAG +back into cache_key would create a circular dependency. +""" + +from __future__ import annotations + +import hashlib +import json + +from aborist.merkle import MerkleTree + + +def _sha256_hex(s: str) -> str: + return hashlib.sha256(s.encode("utf-8")).hexdigest() + + +def _canonical_json(obj) -> str: + return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) + + +def build_run_dag( + *, + question_hash: str, + sources: list[dict], + context_root: str, + conversation_hash: str, + answer_text: str, + audit_mode: str, + verifier_method: str, + n_quotes: int, + n_verified: int, + claim_statuses: list[dict] | None = None, + lookup_path: str | None = None, +) -> dict: + """Return ``{"root": , "nodes": [, ], ...}``. + + All inputs are already-computed hashes or text; no I/O. Idempotent & + deterministic — same inputs always produce the same root, byte-for- + byte across machines (as long as the Merkle conventions stay pinned; + they do, via ``aborist.merkle``). + """ + sources_summary = [ + { + "document_root": s.get("document_root"), + "source_role": s.get("source_role"), + "score": s.get("score"), + "chunk_idx": s.get("chunk_idx"), + } + for s in sources + ] + retrieval_hash = _sha256_hex(_canonical_json(sources_summary)) + answer_hash = _sha256_hex(answer_text) + verify_payload = { + "audit_mode": audit_mode, + "verifier_method": verifier_method, + "n_quotes": n_quotes, + "n_verified": n_verified, + "claim_statuses": claim_statuses or [], + } + verify_hash = _sha256_hex(_canonical_json(verify_payload)) + final_label_payload = { + "audit_mode": audit_mode, + "verifier_method": verifier_method, + "lookup_path": lookup_path, + } + final_label_hash = _sha256_hex(_canonical_json(final_label_payload)) + + nodes = [ + {"stage": "question", "hash": question_hash}, + {"stage": "retrieval", "hash": retrieval_hash}, + {"stage": "context", "hash": context_root}, + {"stage": "prompt", "hash": conversation_hash}, + {"stage": "answer", "hash": answer_hash}, + {"stage": "verify", "hash": verify_hash}, + {"stage": "final_label", "hash": final_label_hash}, + ] + leaves = [bytes.fromhex(n["hash"]) for n in nodes] + root_hex = MerkleTree.build(leaves).root.hex() + return {"root": root_hex, "nodes": nodes} + + +def verify_run_dag(blob: str | dict) -> bool: + """Recompute the Merkle root from ``blob`` and check it matches. + + Used by audit tooling. Accepts either a parsed dict or the JSON + string we persist in ``providence_cache.run_dag_blob``. + """ + if isinstance(blob, str): + blob = json.loads(blob) + nodes = blob.get("nodes") or [] + if not nodes: + return False + leaves = [bytes.fromhex(n["hash"]) for n in nodes] + return MerkleTree.build(leaves).root.hex() == blob.get("root") diff --git a/aborist/qa/query.py b/aborist/qa/query.py index 6ed063d..79aa74e 100644 --- a/aborist/qa/query.py +++ b/aborist/qa/query.py @@ -64,6 +64,7 @@ from aborist.qa.keys import ( model_profile_hash, question_hash, ) +from aborist.qa.dag import build_run_dag from aborist.qa.verify import verify_quotes try: @@ -1078,6 +1079,28 @@ def query( ], } proof_blob = json.dumps(proof_obj, separators=(",", ":")) + + # Per-run Merkle-DAG: question / retrieval / context / prompt / + # answer / verify / final_label nodes hashed in order. The root + # is the run's computation-provenance fingerprint, distinct from + # cache_key (which fingerprints the inputs that determined the + # answer; the DAG fingerprints the answer itself + the path + # that produced it). + run_dag = build_run_dag( + question_hash=qhash, + sources=proof_obj["sources"], + context_root=context_root, + conversation_hash=chash, + answer_text=answer_text, + audit_mode=verdict["audit_mode"], + verifier_method=verdict["verifier_method"], + n_quotes=verdict["n_quotes"], + n_verified=verdict["n_verified"], + claim_statuses=verdict.get("claim_statuses", []), + lookup_path="miss", + ) + run_dag_blob = json.dumps(run_dag, separators=(",", ":")) + now = int(time.time()) with transaction(qa_conn): event_hash = append_audit( @@ -1106,9 +1129,9 @@ def query( " governance_policy_hash, schema_version, canonicalization_version, " " chunking_version, falsification_state, chain, audit_event_hash, " " created_at, hit_count, audit_mode, n_quotes, n_verified, " - " unverified_quotes, verifier_method) " + " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, " - " ?, ?, ?, ?, ?)", + " ?, ?, ?, ?, ?, ?, ?)", ( ckey, context_root, @@ -1131,6 +1154,8 @@ def query( verdict["n_verified"], unverified_blob, verdict["verifier_method"], + run_dag["root"], + run_dag_blob, ), ) finally: @@ -1141,6 +1166,7 @@ def query( "status": "cache_miss_then_written", "audit_mode": verdict["audit_mode"], "cache_key": ckey, + "run_dag_root": run_dag["root"], "lookup_path": "miss", "burned_existing": burned_existing, "context_root": context_root, diff --git a/aborist/qa/runner.py b/aborist/qa/runner.py index e8b2a27..8313b21 100644 --- a/aborist/qa/runner.py +++ b/aborist/qa/runner.py @@ -34,6 +34,7 @@ from aborist.qa.keys import ( model_profile_hash, question_hash, ) +from aborist.qa.dag import build_run_dag from aborist.qa.verify import verify_quotes from aborist.store import append_audit, transaction @@ -302,6 +303,29 @@ def ask( } proof_blob = json.dumps(proof_obj, separators=(",", ":")) + # Per-run Merkle-DAG (see aborist/qa/dag.py). Single-doc shape: + # the only "source" is document_root; that's the retrieval/context + # node identity. + run_dag = build_run_dag( + question_hash=qhash, + sources=[{ + "document_root": document_root, + "source_role": "primary_answer_source", + "score": None, + "chunk_idx": None, + }], + context_root=document_root, + conversation_hash=chash, + answer_text=answer_text, + audit_mode=verdict["audit_mode"], + verifier_method=verdict["verifier_method"], + n_quotes=verdict["n_quotes"], + n_verified=verdict["n_verified"], + claim_statuses=verdict.get("claim_statuses", []), + lookup_path="miss", + ) + run_dag_blob = json.dumps(run_dag, separators=(",", ":")) + now = int(time.time()) with transaction(conn): event_hash = append_audit( @@ -329,9 +353,9 @@ def ask( " governance_policy_hash, schema_version, canonicalization_version, " " chunking_version, falsification_state, chain, audit_event_hash, " " created_at, hit_count, audit_mode, n_quotes, n_verified, " - " unverified_quotes, verifier_method) " + " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, " - " ?, ?, ?, ?, ?)", + " ?, ?, ?, ?, ?, ?, ?)", ( ckey, document_root, @@ -354,6 +378,8 @@ def ask( verdict["n_verified"], unverified_blob, verdict["verifier_method"], + run_dag["root"], + run_dag_blob, ), ) @@ -361,6 +387,7 @@ def ask( "status": "cache_miss_then_written", "audit_mode": verdict["audit_mode"], "cache_key": ckey, + "run_dag_root": run_dag["root"], "lookup_path": "miss", "source_root": document_root, "answer_text": answer_text, diff --git a/aborist/store.py b/aborist/store.py index f59cdde..890acfb 100644 --- a/aborist/store.py +++ b/aborist/store.py @@ -187,7 +187,14 @@ CREATE TABLE IF NOT EXISTS providence_cache ( -- spans matched but multi-word proper nouns did. 'none' = no -- evidence at all (truly emergent). verifier_method TEXT NOT NULL DEFAULT 'none' - CHECK (verifier_method IN ('quote','span','entity','paraphrase','none')) + CHECK (verifier_method IN ('quote','span','entity','paraphrase','none')), + -- Per-run Merkle-DAG. run_dag_root = MerkleTree over ordered stage + -- hashes (question / retrieval / context / prompt / answer / verify / + -- final_label). run_dag_blob carries the full {root, nodes} JSON so + -- an auditor can recompute & verify. Distinct from audit_event_hash + -- (linear DB-wide chain). NULL on legacy records pre-2026-04-30. + run_dag_root TEXT, + run_dag_blob TEXT ); 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); @@ -372,6 +379,14 @@ def _migrate_audit_mode(conn: sqlite3.Connection) -> None: "NOT NULL DEFAULT 'none' " "CHECK (verifier_method IN ('quote','span','entity','paraphrase','none'))" ) + if "run_dag_root" not in cols: + conn.execute( + "ALTER TABLE providence_cache ADD COLUMN run_dag_root TEXT" + ) + if "run_dag_blob" not in cols: + conn.execute( + "ALTER TABLE providence_cache ADD COLUMN run_dag_blob TEXT" + ) # VISUAL → UNGROUNDED rename. Detect legacy CHECK by inspecting DDL. ddl_row = conn.execute( @@ -485,7 +500,9 @@ def _rebuild_providence_cache_paraphrase(conn: sqlite3.Connection) -> None: n_verified INTEGER NOT NULL DEFAULT 0, unverified_quotes TEXT, verifier_method TEXT NOT NULL DEFAULT 'none' - CHECK (verifier_method IN ('quote','span','entity','paraphrase','none')) + CHECK (verifier_method IN ('quote','span','entity','paraphrase','none')), + run_dag_root TEXT, + run_dag_blob TEXT ) """ conn.execute("BEGIN IMMEDIATE") @@ -553,7 +570,9 @@ def _rebuild_providence_cache_ungrounded(conn: sqlite3.Connection) -> None: n_verified INTEGER NOT NULL DEFAULT 0, unverified_quotes TEXT, verifier_method TEXT NOT NULL DEFAULT 'none' - CHECK (verifier_method IN ('quote','span','entity','paraphrase','none')) + CHECK (verifier_method IN ('quote','span','entity','paraphrase','none')), + run_dag_root TEXT, + run_dag_blob TEXT ) """ conn.execute("BEGIN IMMEDIATE") diff --git a/tests/test_dag.py b/tests/test_dag.py new file mode 100644 index 0000000..76ce4ef --- /dev/null +++ b/tests/test_dag.py @@ -0,0 +1,117 @@ +"""Per-run Merkle-DAG provenance. + +`build_run_dag` produces a deterministic Merkle root over the seven +stages of a query/ask call. `verify_run_dag` recomputes the root from +the persisted node list and confirms it matches. + +The DAG is not part of cache_key (cache inputs determine the answer; +the answer determines the DAG — folding it back would create a cycle). +It rides alongside the providence record as `run_dag_root` / +`run_dag_blob` so an auditor can verify the run was constructed +exactly as recorded. +""" + +from __future__ import annotations + +from aborist.qa.dag import build_run_dag, verify_run_dag + + +def _kw(**overrides): + base = dict( + question_hash="a" * 64, + sources=[ + { + "document_root": "b" * 64, + "source_role": "primary_answer_source", + "score": 1.0, + "chunk_idx": 0, + } + ], + context_root="b" * 64, + conversation_hash="c" * 64, + answer_text="The cat is on the mat.", + audit_mode="STRICT", + verifier_method="quote", + n_quotes=1, + n_verified=1, + claim_statuses=[ + {"text": "the cat", "status": "VERIFIED_QUOTE", "method": "quote"} + ], + lookup_path="miss", + ) + base.update(overrides) + return base + + +def test_dag_root_is_deterministic(): + """Same inputs → same root, byte-for-byte.""" + a = build_run_dag(**_kw()) + b = build_run_dag(**_kw()) + assert a["root"] == b["root"] + assert len(a["nodes"]) == 7 + + +def test_dag_root_changes_when_answer_changes(): + a = build_run_dag(**_kw()) + b = build_run_dag(**_kw(answer_text="Different answer.")) + assert a["root"] != b["root"] + + +def test_dag_root_changes_when_audit_mode_changes(): + a = build_run_dag(**_kw()) + b = build_run_dag(**_kw(audit_mode="UNGROUNDED")) + assert a["root"] != b["root"] + + +def test_dag_root_changes_when_verifier_method_changes(): + a = build_run_dag(**_kw()) + b = build_run_dag(**_kw(verifier_method="paraphrase")) + assert a["root"] != b["root"] + + +def test_dag_root_changes_when_question_hash_changes(): + a = build_run_dag(**_kw()) + b = build_run_dag(**_kw(question_hash="d" * 64)) + assert a["root"] != b["root"] + + +def test_dag_root_changes_when_sources_change(): + a = build_run_dag(**_kw()) + new_sources = [{"document_root": "e" * 64, "source_role": "primary_answer_source", + "score": 1.0, "chunk_idx": 0}] + b = build_run_dag(**_kw(sources=new_sources)) + assert a["root"] != b["root"] + + +def test_dag_seven_stages_in_order(): + """The seven stages are emitted in fixed order so the root is + canonical: question / retrieval / context / prompt / answer / + verify / final_label.""" + out = build_run_dag(**_kw()) + stages = [n["stage"] for n in out["nodes"]] + assert stages == [ + "question", "retrieval", "context", "prompt", + "answer", "verify", "final_label", + ] + + +def test_verify_dag_round_trip(): + """Recomputing the root from the persisted nodes must match.""" + out = build_run_dag(**_kw()) + assert verify_run_dag(out) is True + + +def test_verify_dag_detects_node_tampering(): + """Mutating a stage hash makes verify fail.""" + out = build_run_dag(**_kw()) + out["nodes"][3]["hash"] = "f" * 64 # tamper with prompt node + # Root no longer matches the (mutated) leaves. + assert verify_run_dag(out) is False + + +def test_verify_dag_accepts_json_string(): + """Sidecar/audit tools persist the DAG as JSON; verify accepts both.""" + import json + out = build_run_dag(**_kw()) + blob = json.dumps(out, separators=(",", ":")) + assert verify_run_dag(blob) is True diff --git a/tests/test_query.py b/tests/test_query.py index 33eac33..ea84e19 100644 --- a/tests/test_query.py +++ b/tests/test_query.py @@ -374,6 +374,44 @@ def test_query_role_weighted_budget_persists_role_on_sources(tmp_path): assert all("source_role" in s for s in result["sources"]) +def test_query_persists_run_dag_root_on_record_and_result(tmp_path): + """Per-run Merkle-DAG fingerprint surfaces on both the result dict + & the persisted providence_cache row. Recomputing the root from + the persisted blob matches what was stored.""" + from aborist.qa.dag import verify_run_dag + + main_db = tmp_path / "corpus.db" + qa_db = tmp_path / "qa.db" + conn = connect(main_db) + try: + ingest_source(conn, FakeSource(DOCS)) + finally: + conn.close() + + result = query( + question="What is anarchism?", + qa_db=qa_db, + chat_client=StubClient(answer='The source: "a political philosophy that opposes the state"'), + model_id="m", + single_db=main_db, + ) + assert "run_dag_root" in result + assert isinstance(result["run_dag_root"], str) + assert len(result["run_dag_root"]) == 64 # sha256 hex + + qa_conn = connect(qa_db) + try: + row = qa_conn.execute( + "SELECT run_dag_root, run_dag_blob FROM providence_cache " + "WHERE cache_key = ?", + (result["cache_key"],), + ).fetchone() + finally: + qa_conn.close() + assert row["run_dag_root"] == result["run_dag_root"] + assert verify_run_dag(row["run_dag_blob"]) is True + + def test_query_no_sources_when_empty_corpus(tmp_path): main_db = tmp_path / "empty.db" qa_db = tmp_path / "qa.db"