"""CAIN-A75-016 offline verifier (rules embedded verbatim). CAIN-A75-016 memory-consolidation rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. An agent consolidates memory offline ("dreaming"). Consolidation may merge or drop records, so the review checks that it did not launder authority, cross tenants or erase corrections: binding input and output text hashes cover their text; ids are unique isolation every output belongs to one tenant and every source of that output has the same tenant provenance every output names existing sources; a synthesised output is marked as synthesis, a merged output names at least two sources, and no output is invented without a source corrections a corrected input is never dropped and is carried into an output (verbatim) deletion removed ids exist and are not corrected conflicts every conflict names two existing memories and a resolution from the allowed set confidence an output's confidence does not exceed the strongest source plus the allowed gain claims every boolean the record asserts matches what this review computed Verdict: SOUND, else UNSOUND with every failing check named. Malformed input is UNSOUND. It reviews the consolidation record; it does not read the memory store or judge the merged text. """ from __future__ import annotations import hashlib import math import re _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") RESOLUTIONS = {"keep_both", "prefer_newer", "prefer_source", "quarantine", "human_review"} class Bad(ValueError): pass def text_hash(text): return hashlib.sha256(" ".join(text.split()).encode("utf-8")).hexdigest() def _num(v, what, lo=0.0, hi=1.0): if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v) or not (lo <= v <= hi): raise Bad(f"{what} must be a finite number in [{lo}, {hi}]") return float(v) def evaluate(record): checks, summary = [], {"outputs": {}, "sources": {}} def c(cid, ok, detail=""): checks.append({"id": cid, "ok": bool(ok), "detail": detail}) return ok try: if not isinstance(record, dict) or record.get("schema") != "cain42.a75.memory.consolidation.v1": raise Bad("not a cain42.a75.memory.consolidation.v1 record") inputs, outputs = record.get("inputs"), record.get("outputs") if not isinstance(inputs, list) or not inputs or len(inputs) > 5000: raise Bad("inputs must be a non-empty list") if not isinstance(outputs, list) or not outputs or len(outputs) > 5000: raise Bad("outputs must be a non-empty list") I = {} for m in inputs: if not isinstance(m, dict) or not isinstance(m.get("id"), str) or not _ID.match(m["id"]): raise Bad("an input memory needs an id") if m["id"] in I: raise Bad(f"duplicate input {m['id']}") if not isinstance(m.get("tenant"), str) or not m["tenant"]: raise Bad(f"{m['id']}: tenant required") if not isinstance(m.get("text"), str) or not m["text"].strip(): raise Bad(f"{m['id']}: text required") if m.get("sha256") != text_hash(m["text"]): raise Bad(f"{m['id']}: sha256 does not cover the text") _num(m.get("confidence", 0.5), f"{m['id']}.confidence") I[m["id"]] = m O = {} for out in outputs: if not isinstance(out, dict) or not isinstance(out.get("id"), str) or not _ID.match(out["id"]): raise Bad("an output memory needs an id") if out["id"] in O: raise Bad(f"duplicate output {out['id']}") if not isinstance(out.get("tenant"), str) or not out["tenant"]: raise Bad(f"{out['id']}: tenant required") if not isinstance(out.get("text"), str) or not out["text"].strip(): raise Bad(f"{out['id']}: text required") if out.get("sha256") != text_hash(out["text"]): raise Bad(f"{out['id']}: sha256 does not cover the text") O[out["id"]] = out gain = _num(record.get("max_confidence_gain", 0.0), "max_confidence_gain") corrected = {i for i, m in I.items() if m.get("corrected") is True} dropped, carried = set(), set() for oid in sorted(O): out = O[oid] srcs = out.get("derived_from") c(f"{oid}.sources_exist", isinstance(srcs, list) and bool(srcs) and all(s in I for s in srcs), f"derived from {srcs!r}") if not (isinstance(srcs, list) and srcs and all(s in I for s in srcs)): continue tenants = {I[s]["tenant"] for s in srcs} c(f"{oid}.tenant_isolation", tenants == {out["tenant"]}, f"output {out['tenant']!r}, sources {sorted(tenants)}") conf = _num(out.get("confidence"), f"{oid}.confidence") strongest = max(I[s].get("confidence", 0.5) for s in srcs) c(f"{oid}.confidence_bounded", conf <= strongest + gain + 1e-9, f"{conf} <= {strongest} + {gain}") merged = len(srcs) >= 2 if out.get("synthesis") is True: c(f"{oid}.synthesis_declared", True, "a synthesised memory names its sources") elif merged: c(f"{oid}.merge_declared", out.get("merged") is True, "a multi-source output is marked as merged") else: c(f"{oid}.single_source_verbatim", out.get("sha256") == I[srcs[0]]["sha256"], "a single-source output keeps the source text") for s in srcs: if I[s].get("corrected") is True and out.get("synthesis") is not True: c(f"{oid}.carries_correction", out.get("sha256") == I[s]["sha256"], "a corrected source must be carried verbatim unless the output is a declared synthesis") summary["outputs"][oid] = {"tenant": out["tenant"], "sources": srcs} dropped |= set(srcs) carried |= {s for s in srcs if out.get("sha256") == I[s]["sha256"]} for iid in sorted(corrected): c(f"{iid}.correction_preserved", iid in carried, "corrected input is carried verbatim by some output") removed = record.get("removed") if not isinstance(removed, list) or not all(isinstance(x, str) for x in removed): raise Bad("removed must be a list of ids") for rid in removed: c(f"removed.{rid}.exists", rid in I, "removed id exists in the inputs") c(f"removed.{rid}.not_corrected", rid not in corrected, "a corrected memory is never removed") overlap = sorted(set(removed) & dropped) c("removed.not_also_reused", not overlap, f"removed but reused: {overlap}" if overlap else "removed ids are not reused") conflicts = record.get("conflicts") if isinstance(record.get("conflicts"), list) else [] for k, cf in enumerate(conflicts): if not isinstance(cf, dict): raise Bad("a conflict must be an object") a, b = cf.get("a"), cf.get("b") c(f"conflict.{k}.ids_exist", a in I and b in I, f"a={a!r} b={b!r}") c(f"conflict.{k}.resolved", cf.get("resolution") in RESOLUTIONS, f"resolution {cf.get('resolution')!r}") unresolved = [k for k, cf in enumerate(conflicts) if isinstance(cf, dict) and cf.get("resolution") not in RESOLUTIONS] c("conflicts.none_unresolved", not unresolved, f"unresolved: {unresolved}" if unresolved else "all conflicts resolved") claims = record.get("claims") if isinstance(record.get("claims"), dict) else {} computed = { "tenant_isolated": all(set(o.get("derived_from") or []) and {I[s]["tenant"] for s in o.get("derived_from") or [] if s in I} == {o["tenant"]} for o in O.values()), "corrections_preserved": all(i in carried for i in corrected), "nothing_invented": all(o.get("derived_from") and all(s in I for s in o["derived_from"]) for o in O.values()), } for key in sorted(claims): c(f"claims.{key}", key in computed and claims[key] == computed[key], f"claimed {claims[key]!r}, computed {computed.get(key)!r}") summary["computed"] = computed except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "UNSOUND", checks, summary except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"[:200]}) return "UNSOUND", checks, summary return ("SOUND" if all(x["ok"] for x in checks) else "UNSOUND"), checks, summary # ---------------------------------------------------------------------------- offline verifier (no CAIN imports) # Every name below is prefixed _kit_verifier_ / _b64 / _hl / _json / _sys so it cannot shadow a name in the rules above. import base64 as _b64, hashlib as _hl, json as _json, sys as _sys _KIT_VERIFIER_GENESIS = _hl.sha256(b"cain42.a75.memoryconsolidation.genesis").hexdigest() def _kit_verifier_canon(o): return _json.dumps(o, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode() def _kit_verifier_sha(o): return _hl.sha256(_kit_verifier_canon(o)).hexdigest() def _kit_verifier_main(argv): from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey if len(argv) < 2: print("usage: verify_offline.py run.json --pinned-key BASE64"); return 2 run = _json.load(open(argv[1])) pin = argv[argv.index("--pinned-key") + 1] if "--pinned-key" in argv else None problems, prev, recs = [], _KIT_VERIFIER_GENESIS, run.get("records") or [] for i, r in enumerate(recs): body = {k: v for k, v in r.items() if k not in ("record_hash", "signature_b64")} if _kit_verifier_sha(body) != r.get("record_hash"): problems.append(f"record {i}: record_hash does not match") if body.get("subject_sha256") != _kit_verifier_sha(body.get("subject")): problems.append(f"record {i}: subject_sha256 does not match") if pin and r.get("public_key_b64") != pin: problems.append(f"record {i}: different key than the pinned one") try: Ed25519PublicKey.from_public_bytes(_b64.b64decode(pin or r.get("public_key_b64"))).verify( _b64.b64decode(r.get("signature_b64", "")), str(r.get("record_hash")).encode()) except Exception: problems.append(f"record {i}: signature does not verify") if r.get("prev_record_hash") != prev: problems.append(f"record {i}: chain link broken") v, checks, _ = evaluate(r.get("subject")) if v != r.get("verdict") or checks != r.get("checks"): problems.append(f"record {i}: re-running the rules gives {v}, the record says {r.get('verdict')}") prev = r.get("record_hash") if run.get("run_hash") != _kit_verifier_sha([r.get("record_hash") for r in recs]): problems.append("run_hash does not match") for p in problems: print(" -", p) if not pin: print("WARNING: no --pinned-key given; the signature was checked against the key inside the file") print("VERDICT:", "VERIFIED" if not problems else "TAMPERED OR INVALID", f"(CAIN-A75-016, {len(recs)} record(s), verdicts {run.get('verdicts')})") return 0 if not problems else 1 if __name__ == "__main__": _sys.exit(_kit_verifier_main(_sys.argv))