181 lines
6.7 KiB
Python
181 lines
6.7 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
verify_claims.py -- independent re-verification of the agentic-memory dossier.
|
|
|
|
This script does NOT import build_dossier.py. It re-derives every claim from
|
|
the pinned corpus bytes and the recorded pins, so it is a genuine second
|
|
implementation of the grounding check.
|
|
|
|
For every claim in claims.json it checks:
|
|
1. the corpus file's sha256 equals the pin recorded at build time;
|
|
2. the claim's `quote` is a literal substring of that corpus file;
|
|
3. the claim's headline `value` appears inside that quote;
|
|
4. the headline `value` appears in DOSSIER.md.
|
|
|
|
It then runs three mutation batteries. Each battery corrupts the artifact in
|
|
a way a real drift would look like, and requires the corruption to be
|
|
DETECTED. A mutation that passes undetected falsifies this verifier, and the
|
|
script exits nonzero.
|
|
|
|
battery A: corrupt one quote at a time (all claims)
|
|
battery B: corrupt one corpus byte (sha256 pin must catch it)
|
|
battery C: corrupt one headline value everywhere in DOSSIER.md (all values)
|
|
|
|
Exit 0 = the dossier is fully grounded and the verifier is shown to have
|
|
teeth. Exit 1 = an ungrounded claim, or a vacuous check.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent
|
|
CORPUS = ROOT / "corpus"
|
|
EXPECTED_CLAIMS = 44
|
|
EXPECTED_PAPERS = 5
|
|
|
|
|
|
def sha256_bytes(b: bytes) -> str:
|
|
return hashlib.sha256(b).hexdigest()
|
|
|
|
|
|
def verify(claims_doc: dict, dossier_text: str, corpus_dir: Path,
|
|
cache: dict | None = None) -> list[str]:
|
|
"""Return a list of failure strings. Empty list == pass."""
|
|
cache = {} if cache is None else cache
|
|
fails: list[str] = []
|
|
|
|
papers = {p["arxiv_id"]: p for p in claims_doc.get("papers", [])}
|
|
if len(papers) != EXPECTED_PAPERS:
|
|
fails.append(f"papers: expected {EXPECTED_PAPERS}, found {len(papers)}")
|
|
if len(claims_doc.get("claims", [])) != EXPECTED_CLAIMS:
|
|
fails.append(f"claims: expected {EXPECTED_CLAIMS}, found "
|
|
f"{len(claims_doc.get('claims', []))}")
|
|
|
|
seen_ids: set[str] = set()
|
|
counts: dict[str, int] = {}
|
|
for c in claims_doc.get("claims", []):
|
|
cid, pid = c.get("id", "?"), c.get("paper", "?")
|
|
if cid in seen_ids:
|
|
fails.append(f"{cid}: duplicate claim id")
|
|
seen_ids.add(cid)
|
|
counts[pid] = counts.get(pid, 0) + 1
|
|
|
|
if pid not in papers:
|
|
fails.append(f"{cid}: unknown paper {pid!r}")
|
|
continue
|
|
rec = papers[pid]
|
|
if rec.get("sha256") != c.get("corpus_sha256"):
|
|
fails.append(f"{cid}: pin in papers[] disagrees with claim pin")
|
|
|
|
f = corpus_dir / f"{pid}.txt"
|
|
if not f.exists():
|
|
fails.append(f"{cid}: corpus file missing: {f.name}")
|
|
continue
|
|
if pid not in cache:
|
|
cache[pid] = f.read_bytes()
|
|
data = cache[pid]
|
|
dig = sha256_bytes(data)
|
|
if dig != c.get("corpus_sha256"):
|
|
fails.append(f"{cid}: corpus sha256 drift for {pid} "
|
|
f"(pin {str(c.get('corpus_sha256'))[:12]}, got {dig[:12]})")
|
|
continue
|
|
text = data.decode("utf-8", errors="replace")
|
|
if c["quote"] not in text:
|
|
fails.append(f"{cid}: quote not found verbatim in {pid}.txt")
|
|
continue
|
|
if c["value"] not in c["quote"]:
|
|
fails.append(f"{cid}: value {c['value']!r} not inside its own quote")
|
|
if c["value"] not in dossier_text:
|
|
fails.append(f"{cid}: value {c['value']!r} missing from DOSSIER.md")
|
|
|
|
for pid, rec in papers.items():
|
|
f = corpus_dir / f"{pid}.txt"
|
|
if f.exists():
|
|
if pid not in cache:
|
|
cache[pid] = f.read_bytes()
|
|
if sha256_bytes(cache[pid]) != rec.get("sha256"):
|
|
fails.append(f"{pid}: papers[] sha256 pin does not match corpus")
|
|
if counts.get(pid, 0) != rec.get("claims"):
|
|
fails.append(f"{pid}: claims[] count {rec.get('claims')} != "
|
|
f"actual {counts.get(pid, 0)}")
|
|
return fails
|
|
|
|
|
|
def corrupt_char(s: str) -> str:
|
|
if not s:
|
|
return "X"
|
|
return s[:-1] + ("X" if s[-1] != "X" else "Y")
|
|
|
|
|
|
def main() -> int:
|
|
claims_doc = json.loads((ROOT / "claims.json").read_text(encoding="utf-8"))
|
|
dossier_text = (ROOT / "DOSSIER.md").read_text(encoding="utf-8")
|
|
cache: dict[str, bytes] = {}
|
|
|
|
fails = verify(claims_doc, dossier_text, CORPUS, cache)
|
|
if fails:
|
|
print("FAIL: dossier is not fully grounded")
|
|
for f in fails:
|
|
print(" -", f)
|
|
return 1
|
|
n = len(claims_doc["claims"])
|
|
print(f"grounding: {n} claims x 4 checks -- all pass")
|
|
|
|
# ---- battery A: corrupt one quote at a time; every corruption must be caught
|
|
undetected = []
|
|
for i, c in enumerate(claims_doc["claims"]):
|
|
mutant = copy.deepcopy(claims_doc)
|
|
q = mutant["claims"][i]["quote"]
|
|
mid = len(q) // 2
|
|
mutant["claims"][i]["quote"] = q[:mid] + corrupt_char(q[mid]) + q[mid + 1:]
|
|
if not verify(mutant, dossier_text, CORPUS, cache):
|
|
undetected.append(c["id"])
|
|
if undetected:
|
|
print(f"FAIL: battery A vacuous -- quote corruption undetected for {undetected}")
|
|
return 1
|
|
print(f"battery A: {n} quote corruptions, all detected")
|
|
|
|
# ---- battery B: flip one corpus byte; the sha256 pin must catch it
|
|
with tempfile.TemporaryDirectory() as td:
|
|
tdir = Path(td)
|
|
victim = claims_doc["claims"][0]["paper"]
|
|
for pid in {c["paper"] for c in claims_doc["claims"]}:
|
|
(tdir / f"{pid}.txt").write_bytes(cache[pid])
|
|
p = tdir / f"{victim}.txt"
|
|
b = bytearray(p.read_bytes())
|
|
b[len(b) // 2] ^= 0x01
|
|
p.write_bytes(bytes(b))
|
|
detected = verify(claims_doc, dossier_text, tdir, {})
|
|
if detected:
|
|
print(f"battery B: corpus byte flip in {victim} detected via sha256 pin")
|
|
else:
|
|
print(f"FAIL: battery B vacuous -- corpus byte flip in {victim} undetected")
|
|
return 1
|
|
|
|
# ---- battery C: corrupt one headline value everywhere in the dossier
|
|
undetected_c = []
|
|
for c in claims_doc["claims"]:
|
|
v = c["value"]
|
|
if v not in dossier_text:
|
|
continue
|
|
bad = dossier_text.replace(v, corrupt_char(v))
|
|
if not verify(claims_doc, bad, CORPUS, cache):
|
|
undetected_c.append(c["id"])
|
|
if undetected_c:
|
|
print(f"FAIL: battery C vacuous -- dossier value corruption undetected for "
|
|
f"{undetected_c}")
|
|
return 1
|
|
print(f"battery C: {n} dossier value corruptions, all detected")
|
|
|
|
print("VERDICT: PASS -- dossier grounded, verifier falsified-checked")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|