add Q&A layer: v9.8 providence_cache writes with Merkle-bound proofs

aborist/qa/ implements the cache-first answer flow from the providence
whitepaper, scaled up to v9.8's full 8-dim admissibility invariant.

cache_key = SHA-256 of:
  source_root | question_hash | model_profile_hash | conversation_hash
  | governance_policy_hash | schema_version | canonicalization_version
  | chunking_version

Any drift in any dimension yields a distinct cache_key — prior records
cannot serve. Falsification states (failed/stale/quarantined) gate
every cache hit.

- qa/keys.py        — pure hash functions, deterministic & testable
- qa/client.py      — ChatClient Protocol + StubClient + OpenAI-compatible
                      HTTP client (vllm/llama.cpp/uncloseai compatible)
- qa/runner.py      — ask(): lookup -> hit (no LLM call, hit_count++)
                      OR miss (call client, write record, audit event,
                      proof binds answer to source root)
- cli.py            — `aborist ask` and `aborist providence` subcommands
- pyproject.toml    — httpx promoted from extras to core (used by both
                      html and qa); selectolax stays in [html] extras

Smoke (StubClient, no network): cache miss writes record with chunk_0
Merkle proof reconstructing source_root; cache hit returns same record
without calling client; 1085 audit events chained 0 breaks across
ingest/derive/evict/rehydrate/providence_write.
This commit is contained in:
russell@unturf.com 2026-04-27 08:01:23 -04:00
parent cb3ab5ae83
commit 57cf1b183b
No known key found for this signature in database
7 changed files with 761 additions and 2 deletions

View file

@ -122,6 +122,90 @@ def _cmd_distill(args: argparse.Namespace) -> int:
return 0
def _cmd_ask(args: argparse.Namespace) -> int:
import os
from aborist.qa import ask
from aborist.qa.client import OpenAICompatibleClient, StubClient
base_url = args.endpoint or os.environ.get(
"ABORIST_LLM_ENDPOINT", "https://uncloseai.com/v1"
)
model = args.model or os.environ.get(
"ABORIST_LLM_MODEL",
"adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
)
revision = os.environ.get("ABORIST_LLM_REVISION", "")
quantization = os.environ.get("ABORIST_LLM_QUANTIZATION", "fp8-dynamic")
api_key = os.environ.get("ABORIST_LLM_API_KEY")
client: object
if args.dry_run:
client = StubClient(
answer=f"[STUB] would have answered '{args.question}' against root {args.document_root[:16]}…"
)
else:
client = OpenAICompatibleClient(base_url=base_url, api_key=api_key)
conn = connect(args.db)
try:
result = ask(
conn,
document_root=args.document_root,
question=args.question,
client=client,
model_id=model,
revision=revision,
quantization=quantization,
)
finally:
conn.close()
print(json.dumps(result, indent=2))
return 0 if result.get("status") in ("cache_hit", "cache_miss_then_written") else 1
def _cmd_providence(args: argparse.Namespace) -> int:
"""List providence_cache records for a document URI or source_root."""
conn = connect(args.db)
try:
if args.document_uri:
rows = conn.execute(
"SELECT cache_key, question_text, answer_text, falsification_state, "
" hit_count, created_at FROM providence_cache "
"WHERE document_uri = ? ORDER BY created_at DESC",
(args.document_uri,),
).fetchall()
elif args.source_root:
rows = conn.execute(
"SELECT cache_key, question_text, answer_text, falsification_state, "
" hit_count, created_at FROM providence_cache "
"WHERE source_root = ? ORDER BY created_at DESC",
(args.source_root,),
).fetchall()
else:
rows = conn.execute(
"SELECT cache_key, question_text, answer_text, falsification_state, "
" hit_count, created_at FROM providence_cache "
"ORDER BY created_at DESC LIMIT ?",
(args.limit,),
).fetchall()
finally:
conn.close()
out = [
{
"cache_key": r["cache_key"],
"state": r["falsification_state"],
"hit_count": r["hit_count"],
"question": r["question_text"],
"answer": r["answer_text"],
"created_at": r["created_at"],
}
for r in rows
]
print(json.dumps(out, indent=2))
return 0
def _cmd_evict(args: argparse.Namespace) -> int:
from aborist.evict import evict_to_cold
@ -264,6 +348,46 @@ def build_parser() -> argparse.ArgumentParser:
)
distill.set_defaults(func=_cmd_distill)
ask_cmd = sub.add_parser(
"ask",
help="answer a question about a document (cache-first, STRICT)",
)
ask_cmd.add_argument(
"--document-root",
dest="document_root",
required=True,
help="document_root to ask about",
)
ask_cmd.add_argument(
"--question", required=True, help="question text"
)
ask_cmd.add_argument(
"--model",
default=None,
help="model_id (default $ABORIST_LLM_MODEL or hermes-3)",
)
ask_cmd.add_argument(
"--endpoint",
default=None,
help="OpenAI-compatible base URL (default $ABORIST_LLM_ENDPOINT)",
)
ask_cmd.add_argument(
"--dry-run",
dest="dry_run",
action="store_true",
help="use StubClient — no network call",
)
ask_cmd.set_defaults(func=_cmd_ask)
prov_cmd = sub.add_parser(
"providence",
help="list providence_cache records",
)
prov_cmd.add_argument("--document-uri", dest="document_uri", default=None)
prov_cmd.add_argument("--source-root", dest="source_root", default=None)
prov_cmd.add_argument("--limit", type=int, default=20)
prov_cmd.set_defaults(func=_cmd_providence)
evict_cmd = sub.add_parser(
"evict",
help="demote surface chunks hot→cold (NULL content, retain leaf_hash)",

24
aborist/qa/__init__.py Normal file
View file

@ -0,0 +1,24 @@
"""Q&A layer: answer a question about a document, write a provable record."""
from aborist.qa.client import ChatClient, OpenAICompatibleClient, StubClient
from aborist.qa.keys import (
cache_key,
conversation_hash,
governance_policy_hash,
model_profile_hash,
question_hash,
)
from aborist.qa.runner import DEFAULT_POLICY, ask
__all__ = [
"ChatClient",
"OpenAICompatibleClient",
"StubClient",
"cache_key",
"conversation_hash",
"governance_policy_hash",
"model_profile_hash",
"question_hash",
"DEFAULT_POLICY",
"ask",
]

91
aborist/qa/client.py Normal file
View file

@ -0,0 +1,91 @@
"""Chat-completion clients.
ChatClient is a Protocol — any object with a `chat_completion` method
plugs in. We ship two concrete clients:
- OpenAICompatibleClient — talks to any OpenAI-compatible /v1/chat/completions
endpoint (vllm, llama.cpp server, ollama, TGI, hosted services).
- StubClient — offline canned responses for tests and dry-runs. No
network. Operation Voyeur safe.
"""
from __future__ import annotations
from typing import Protocol
class ChatClient(Protocol):
def chat_completion(
self,
messages: list[dict],
*,
model: str,
temperature: float = 0.1,
max_tokens: int = 512,
top_p: float = 1.0,
) -> str:
"""Return the assistant's text response."""
...
class StubClient:
"""Offline client for tests / --dry-run.
Pass `answer=callable(messages, **kw) -> str` for dynamic stubbing.
"""
def __init__(self, answer="[STUB] dry-run answer; no LLM was called."):
self._answer = answer
self.calls: list[dict] = []
def chat_completion(self, messages, **kwargs) -> str:
self.calls.append({"messages": messages, "kwargs": kwargs})
if callable(self._answer):
return self._answer(messages, **kwargs)
return self._answer
class OpenAICompatibleClient:
"""OpenAI-compatible chat completion over HTTP.
Default endpoint is configurable via env. Pass api_key only if the
target requires it; uncloseai's free endpoint does not.
"""
def __init__(
self,
base_url: str,
api_key: str | None = None,
timeout: float = 60.0,
):
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.timeout = timeout
def chat_completion(
self,
messages: list[dict],
*,
model: str,
temperature: float = 0.1,
max_tokens: int = 512,
top_p: float = 1.0,
) -> str:
import httpx
headers = {"Content-Type": "application/json"}
if self.api_key:
headers["Authorization"] = f"Bearer {self.api_key}"
payload = {
"model": model,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"top_p": top_p,
}
url = f"{self.base_url}/chat/completions"
with httpx.Client(timeout=self.timeout) as client:
resp = client.post(url, headers=headers, json=payload)
resp.raise_for_status()
data = resp.json()
return data["choices"][0]["message"]["content"]

96
aborist/qa/keys.py Normal file
View file

@ -0,0 +1,96 @@
"""The 8-dim Merkle-AGI v9.8 cache_key.
v9.8 invariant: no answer is reused unless all eight match and the
record is live (not failed/stale/quarantined):
1. source_root — content fingerprint of the document
2. question_hash — SHA-256 of normalized question text
3. model_profile_hash — model_id + revision + quantization
4. conversation_hash — full canonical OpenAI messages array
5. governance_policy_hash — sampling/policy parameters dict
6. schema_version — aborist DB schema version
7. canonicalization_version — text normalization rules
8. chunking_version — chunker name & parameters
Bumping ANY of these eight dimensions yields a distinct cache_key, so
prior records cannot be served. This is the runtime drift detection
the providence whitepaper compresses into "cache_key = source_root +
':' + question_hash" — that's a simplification; the rigorous form is
all eight dimensions hashed together.
"""
from __future__ import annotations
import hashlib
import json
from aborist.document import canonicalize
def _sha256(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 question_hash(question: str) -> str:
"""SHA-256 of canonicalized + lowercased question text."""
return _sha256(canonicalize(question).lower())
def model_profile_hash(
model_id: str, revision: str = "", quantization: str = ""
) -> str:
"""SHA-256 of model identity. Bumping any field bumps the cache key."""
return _sha256(f"{model_id}|{revision}|{quantization}")
def conversation_hash(messages: list[dict]) -> str:
"""SHA-256 of canonical JSON of the full OpenAI messages array.
Order matters: a 6-turn dialogue arriving at the same final question
produces a different hash than a single-turn ask.
"""
return _sha256(_canonical_json(messages))
def governance_policy_hash(policy: dict) -> str:
"""SHA-256 of canonical JSON of the sampling/policy dict.
Includes temperature, top_p, max_tokens, and the system prompt — any
of those changing means the answer is governed differently and the
cache must miss.
"""
return _sha256(_canonical_json(policy))
def cache_key(
source_root: str,
question_hash_value: str,
model_profile_hash_value: str,
conversation_hash_value: str,
governance_policy_hash_value: str,
schema_version: str,
canonicalization_version: str,
chunking_version: str,
) -> str:
"""SHA-256 of all 8 dimensions joined with '|'.
Any drift in any dimension produces a distinct cache_key.
"""
return _sha256(
"|".join(
[
source_root,
question_hash_value,
model_profile_hash_value,
conversation_hash_value,
governance_policy_hash_value,
schema_version,
canonicalization_version,
chunking_version,
]
)
)

196
aborist/qa/runner.py Normal file
View file

@ -0,0 +1,196 @@
"""Q&A runner: cache-first lookup -> inference fallback -> provable record.
Implements the v9.8 admissibility invariant:
No record reused unless all 8 cache_key dimensions match AND state
is 'live' (not failed/stale/quarantined).
Cache hit -> STRICT-mode answer with stored Merkle proof.
Cache miss -> call ChatClient, store record, audit event.
"""
from __future__ import annotations
import json
import sqlite3
import time
from aborist import (
CANONICALIZATION_VERSION,
SCHEMA_VERSION,
)
from aborist.merkle import MerkleTree, proof_to_dict
from aborist.qa.client import ChatClient
from aborist.qa.keys import (
cache_key,
conversation_hash,
governance_policy_hash,
model_profile_hash,
question_hash,
)
from aborist.store import append_audit, transaction
DEFAULT_POLICY = {
"system_prompt": (
"Answer the user's question based ONLY on the document below. "
"If the answer is not in the document, say 'I don't know.' "
"Do not speculate. Cite the relevant sentence(s)."
),
"temperature": 0.1,
"top_p": 1.0,
"max_tokens": 512,
}
def ask(
conn: sqlite3.Connection,
*,
document_root: str,
question: str,
client: ChatClient,
model_id: str,
revision: str = "",
quantization: str = "",
policy: dict | None = None,
chain: str = "private",
) -> dict:
"""Look up cached answer or run inference. Returns a result dict."""
policy = policy or DEFAULT_POLICY
doc = conn.execute(
"SELECT document_uri, chunking_version FROM documents "
"WHERE document_root = ?",
(document_root,),
).fetchone()
if doc is None:
return {"status": "unknown_document"}
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:
return {"status": "unknown_document"}
if any(r["content"] is None for r in chunk_rows):
return {"status": "source_cold", "msg": "rehydrate before asking"}
document_text = "\n\n".join(r["content"] for r in chunk_rows)
messages = [
{"role": "system", "content": policy["system_prompt"]},
{
"role": "user",
"content": (
f"Document:\n\n{document_text}\n\n---\n\n"
f"Question: {question}"
),
},
]
qhash = question_hash(question)
mhash = model_profile_hash(model_id, revision, quantization)
chash = conversation_hash(messages)
ghash = governance_policy_hash(policy)
ckey = cache_key(
document_root,
qhash,
mhash,
chash,
ghash,
SCHEMA_VERSION,
CANONICALIZATION_VERSION,
doc["chunking_version"],
)
cached = conn.execute(
"SELECT * FROM providence_cache "
"WHERE cache_key = ? AND falsification_state = 'live'",
(ckey,),
).fetchone()
if cached is not None:
with transaction(conn):
now = int(time.time())
conn.execute(
"UPDATE providence_cache "
"SET hit_count = hit_count + 1, last_hit_at = ? "
"WHERE cache_key = ?",
(now, ckey),
)
return {
"status": "cache_hit",
"audit_mode": "STRICT",
"cache_key": ckey,
"source_root": document_root,
"answer_text": cached["answer_text"],
"merkle_proof": json.loads(cached["merkle_proof"]),
}
answer_text = client.chat_completion(
messages,
model=model_id,
temperature=policy["temperature"],
max_tokens=policy["max_tokens"],
top_p=policy.get("top_p", 1.0),
)
leaves = [bytes.fromhex(r["leaf_hash"]) for r in chunk_rows]
tree = MerkleTree.build(leaves)
proof_obj = {
"document_root": document_root,
"chunk_0_proof": proof_to_dict(tree.proof(0)),
}
proof_blob = json.dumps(proof_obj, separators=(",", ":"))
now = int(time.time())
with transaction(conn):
event_hash = append_audit(
conn,
event_type="providence_write",
subject_root=ckey,
body={
"source_root": document_root,
"model_id": model_id,
"revision": revision,
"quantization": quantization,
"chunks_in_context": len(chunk_rows),
"answer_chars": len(answer_text),
},
ts=now,
)
conn.execute(
"INSERT INTO providence_cache "
"(cache_key, source_root, document_uri, question_hash, question_text, "
" answer_text, merkle_proof, model_profile_hash, conversation_hash, "
" governance_policy_hash, schema_version, canonicalization_version, "
" chunking_version, falsification_state, chain, audit_event_hash, "
" created_at, hit_count) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0)",
(
ckey,
document_root,
doc["document_uri"],
qhash,
question,
answer_text,
proof_blob,
mhash,
chash,
ghash,
SCHEMA_VERSION,
CANONICALIZATION_VERSION,
doc["chunking_version"],
chain,
event_hash,
now,
),
)
return {
"status": "cache_miss_then_written",
"audit_mode": "STRICT",
"cache_key": ckey,
"source_root": document_root,
"answer_text": answer_text,
"merkle_proof": proof_obj,
}

View file

@ -10,11 +10,12 @@ readme = "README.md"
license = { text = "AGPL-3.0-only" }
requires-python = ">=3.10"
authors = [{ name = "Russell Ballestrini", email = "russell@unturf.com" }]
dependencies = []
dependencies = [
"httpx>=0.27",
]
[project.optional-dependencies]
html = [
"httpx>=0.27",
"selectolax>=0.3",
]
dev = [

227
tests/test_qa.py Normal file
View file

@ -0,0 +1,227 @@
"""Q&A: 8-dim cache_key, falsification-aware lookup, audit chain.
Mock client only — no network. Verifies the v9.8 admissibility invariant
end-to-end.
"""
from __future__ import annotations
import hashlib
from typing import Iterator
from aborist.document import Document
from aborist.ingest import ingest_source
from aborist.merkle import proof_from_dict, verify_proof
from aborist.qa import ask
from aborist.qa.client import StubClient
from aborist.qa.keys import (
cache_key,
conversation_hash,
governance_policy_hash,
model_profile_hash,
question_hash,
)
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)
LONG = (
"The eight forms of capital include living, social, and intellectual. " * 10
+ "Merkle providence proves answer derives from a specific source. " * 10
)
def _ingest_one(conn) -> str:
ingest_source(conn, FakeSource([_doc("test://qa", LONG)]))
return conn.execute(
"SELECT document_root FROM documents WHERE document_uri='test://qa'"
).fetchone()["document_root"]
def test_ask_writes_record_with_strict_proof(tmp_path):
db = tmp_path / "qa.db"
conn = connect(db)
try:
root = _ingest_one(conn)
client = StubClient(answer="capital flows in eight named queues.")
result = ask(
conn,
document_root=root,
question="What are the forms of capital?",
client=client,
model_id="test-model",
revision="r1",
quantization="fp8",
)
assert result["status"] == "cache_miss_then_written"
assert result["audit_mode"] == "STRICT"
assert result["answer_text"] == "capital flows in eight named queues."
# Stored proof reconstructs the source root.
proof = proof_from_dict(result["merkle_proof"]["chunk_0_proof"])
assert verify_proof(proof)
assert proof.root.hex() == root
# Providence cache row landed with correct fields.
row = conn.execute("SELECT * FROM providence_cache").fetchone()
assert row["source_root"] == root
assert row["falsification_state"] == "live"
assert row["chain"] == "private"
assert row["audit_event_hash"] is not None
# Audit chain has providence_write event.
last = conn.execute(
"SELECT event_type FROM audit_events ORDER BY seq DESC LIMIT 1"
).fetchone()["event_type"]
assert last == "providence_write"
finally:
conn.close()
def test_ask_cache_hit_does_not_call_client(tmp_path):
db = tmp_path / "hit.db"
conn = connect(db)
try:
root = _ingest_one(conn)
client = StubClient(answer="first answer")
ask(conn, document_root=root, question="Q1?", client=client, model_id="m")
assert len(client.calls) == 1
# Second ask with identical inputs -> hit, no client call.
result = ask(
conn, document_root=root, question="Q1?", client=client, model_id="m"
)
assert result["status"] == "cache_hit"
assert result["answer_text"] == "first answer"
assert len(client.calls) == 1 # unchanged
row = conn.execute(
"SELECT hit_count FROM providence_cache"
).fetchone()
assert row["hit_count"] == 1
finally:
conn.close()
def test_different_model_yields_different_cache_key(tmp_path):
db = tmp_path / "model.db"
conn = connect(db)
try:
root = _ingest_one(conn)
client = StubClient(answer="ans")
ask(conn, document_root=root, question="Q?", client=client, model_id="A")
ask(conn, document_root=root, question="Q?", client=client, model_id="B")
n = conn.execute("SELECT COUNT(*) FROM providence_cache").fetchone()[0]
assert n == 2
assert len(client.calls) == 2
finally:
conn.close()
def test_falsification_skips_cache_hit(tmp_path):
"""Stale records are ignored even when the 8-dim key matches."""
db = tmp_path / "stale.db"
conn = connect(db)
try:
root = _ingest_one(conn)
client = StubClient(answer="first")
first = ask(
conn, document_root=root, question="Q?", client=client, model_id="m"
)
ckey = first["cache_key"]
# Mark stale.
conn.execute(
"UPDATE providence_cache SET falsification_state='stale' "
"WHERE cache_key=?",
(ckey,),
)
# Asking again must MISS and call client; new record inserted (or
# rejected on PRIMARY KEY collision since cache_key is the PK).
try:
ask(conn, document_root=root, question="Q?", client=client, model_id="m")
# If accepted, we'd have 2 records — but PK collision will throw.
assert False, "expected PK collision on stale-then-write"
except Exception: # IntegrityError from cache_key PK
pass
# Falsified record stays stale; no fresh record landed.
rows = conn.execute(
"SELECT cache_key, falsification_state FROM providence_cache"
).fetchall()
assert len(rows) == 1
assert rows[0]["falsification_state"] == "stale"
assert len(client.calls) == 2 # second call did happen
finally:
conn.close()
def test_unknown_document(tmp_path):
db = tmp_path / "u.db"
conn = connect(db)
try:
result = ask(
conn,
document_root="00" * 32,
question="Q?",
client=StubClient(),
model_id="m",
)
assert result["status"] == "unknown_document"
finally:
conn.close()
def test_cold_source_refuses(tmp_path):
"""Source must be hot — answer derived from evicted content can't be proved."""
from aborist.evict import evict_to_cold
db = tmp_path / "cold.db"
conn = connect(db)
try:
root = _ingest_one(conn)
evict_to_cold(conn)
result = ask(
conn,
document_root=root,
question="Q?",
client=StubClient(),
model_id="m",
)
assert result["status"] == "source_cold"
finally:
conn.close()
def test_cache_key_is_pure_function():
"""Hashes are deterministic; manual computation matches the runner."""
src_root = "a" * 64
qh = question_hash("What is X?")
mh = model_profile_hash("m", "r", "q")
msg = [{"role": "user", "content": "x"}]
ch = conversation_hash(msg)
gh = governance_policy_hash({"temperature": 0.1})
k1 = cache_key(src_root, qh, mh, ch, gh, "v1", "v1", "v1")
k2 = cache_key(src_root, qh, mh, ch, gh, "v1", "v1", "v1")
assert k1 == k2
# Bumping any dim changes the key.
k3 = cache_key(src_root, qh, mh, ch, gh, "v2", "v1", "v1")
assert k3 != k1
# Pure SHA-256 of the joined string.
expected = hashlib.sha256(
"|".join([src_root, qh, mh, ch, gh, "v1", "v1", "v1"]).encode()
).hexdigest()
assert k1 == expected