engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

qa: per-run Merkle-DAG provenance — run_dag_root on every providence record

Browse source 
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).
This commit is contained in:
russell@unturf.com 2026-04-29 18:44:17 -04:00
parent 3cdebb7e2a
commit 28feee3efb
No known key found for this signature in database
6 changed files with 362 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

128
aborist/qa/dag.py Normal file
View file

@ -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": <hex>, "nodes": [<stage>, <hash>], ...}``.
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")

30
aborist/qa/query.py
View file

@ -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,

31
aborist/qa/runner.py
View file

@ -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,

25
aborist/store.py
View file

@ -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")

117
tests/test_dag.py Normal file
View file

@ -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

38
tests/test_query.py
View file

@ -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"