"""CAIN-A75-003 offline verifier (rules embedded verbatim). CAIN-A75-003 discovery-loop rules. Standard library only, no CAIN imports: embedded verbatim in the offline verifier. A closed-loop discovery campaign is a sequence of cycles. In each cycle the agent runs experiments (measurements), states claims about them, and revises its strategy for the next cycle. The rules decide, per cycle, whether the agent's record keeps facts apart and whether its learning rests on physical evidence: every claim has exactly one fact class: OBSERVED, VERIFIED, DERIVED, UNVERIFIED or COUNTERFACTUAL OBSERVED cites one measurement of THIS cycle that has a raw-data SHA-256, and its assertion holds on that value VERIFIED cites >= 2 measurements of this cycle from independent sources (different instrument or a named non-agent verifier), all of which satisfy the assertion and agree within the stated tolerance DERIVED cites earlier claims (no cycle in the derivation graph, none COUNTERFACTUAL) and a named method UNVERIFIED allowed, but can never be the only basis of a strategy revision COUNTERFACTUAL never cited as evidence by any other claim or by a revision no measurement from a later cycle may be cited (no hindsight); every measurement of the cycle is either cited or explicitly excluded with a reason (no silent cherry-picking) and keeps its raw-data hash even when excluded; a strategy revision must cite at least one OBSERVED, VERIFIED or DERIVED claim that the cycle produced or inherited. Verdict per cycle: SUPPORTED, or NOT_SUPPORTED with every failing check named. Malformed input is NOT_SUPPORTED. The rules check the record, not the chemistry: a fabricated measurement with a plausible hash passes the record checks. They catch an agent that learns from nothing, overstates evidence, or hides inconvenient results. """ from __future__ import annotations import math import re CLASSES = {"OBSERVED", "VERIFIED", "DERIVED", "UNVERIFIED", "COUNTERFACTUAL"} OPS = {"<", "<=", ">", ">=", "=="} _HEX = re.compile(r"^[0-9a-f]{64}$") _ID = re.compile(r"^[A-Za-z0-9_.:\-]{1,80}$") class Bad(ValueError): pass def _id(v, what): if not isinstance(v, str) or not _ID.match(v): raise Bad(f"{what} must be an identifier") return v def _num(v, what): if isinstance(v, bool) or not isinstance(v, (int, float)) or not math.isfinite(v): raise Bad(f"{what} must be a finite number") return float(v) def _holds(value, op, target, tol): if op == "<": return value < target if op == "<=": return value <= target + tol if op == ">": return value > target if op == ">=": return value >= target - tol return abs(value - target) <= tol def evaluate(cycle): """cycle = {schema, campaign, cycle, agent, measurements:[...], claims:[...], exclusions:[...], revision:{...}, prior_claims:[...] (claims of earlier cycles, carried forward)}. Returns (verdict, checks, summary).""" checks = [] def c(cid, ok, detail=""): checks.append({"id": cid, "ok": bool(ok), "detail": detail}) return ok try: if not isinstance(cycle, dict) or cycle.get("schema") != "cain42.a75.discovery.cycle.v1": raise Bad("not a cain42.a75.discovery.cycle.v1 record") n = int(_num(cycle.get("cycle"), "cycle")) agent = _id(cycle.get("agent"), "agent") meas = cycle.get("measurements") claims = cycle.get("claims") if not isinstance(meas, list) or not isinstance(claims, list): raise Bad("measurements and claims must be lists") M = {} for m in meas: if not isinstance(m, dict): raise Bad("a measurement must be an object") mid = _id(m.get("id"), "measurement.id") if mid in M: raise Bad(f"duplicate measurement id {mid}") _num(m.get("value"), f"{mid}.value") _id(m.get("metric"), f"{mid}.metric") if int(_num(m.get("cycle"), f"{mid}.cycle")) > n: raise Bad(f"{mid} is from cycle {m.get('cycle')}, after this cycle {n}") M[mid] = m prior = cycle.get("prior_claims") if isinstance(cycle.get("prior_claims"), list) else [] C = {} for cl in prior + claims: if not isinstance(cl, dict): raise Bad("a claim must be an object") C[_id(cl.get("id"), "claim.id")] = cl own = {cl["id"] for cl in claims} for cl in claims: cid, fc = cl["id"], cl.get("class") if not c(f"{cid}.class", fc in CLASSES, f"class {fc!r}"): continue a = cl.get("assert") if isinstance(cl.get("assert"), dict) else None refs = cl.get("measurements") if isinstance(cl.get("measurements"), list) else [] if fc in ("OBSERVED", "VERIFIED"): if not c(f"{cid}.assertion", a is not None and a.get("op") in OPS and isinstance(a.get("metric"), str), "OBSERVED/VERIFIED claims carry {metric, op, value, tolerance}"): continue tgt, tol = _num(a.get("value"), f"{cid}.assert.value"), _num(a.get("tolerance", 0), f"{cid}.assert.tolerance") known = [M.get(r) for r in refs] if not c(f"{cid}.measurements_exist", refs and all(known), f"cites {refs}"): continue c(f"{cid}.same_cycle", all(int(m["cycle"]) == n for m in known), "observed and verified claims cite this cycle's measurements") c(f"{cid}.raw_data_hash", all(isinstance(m.get("raw_sha256"), str) and _HEX.match(m["raw_sha256"]) for m in known), "every cited measurement needs the SHA-256 of its raw data") c(f"{cid}.metric_matches", all(m["metric"] == a["metric"] for m in known), f"claim metric {a['metric']!r}") c(f"{cid}.assertion_holds", all(_holds(float(m["value"]), a["op"], tgt, tol) for m in known), "; ".join(f"{m['id']}={m['value']}" for m in known) + f" vs {a['op']} {tgt} (tol {tol})") if fc == "OBSERVED": c(f"{cid}.one_measurement", len(known) == 1, "an OBSERVED claim cites exactly one measurement") else: sources = {(m.get("instrument"), m.get("verified_by")) for m in known} indep = len({m.get("instrument") for m in known}) >= 2 or any( isinstance(m.get("verified_by"), str) and m["verified_by"] and m["verified_by"] != agent for m in known) c(f"{cid}.independent_sources", len(known) >= 2 and indep and len(sources) >= 2, "VERIFIED needs >= 2 measurements from different instruments, or a named non-agent verifier") vals = [float(m["value"]) for m in known] c(f"{cid}.replicates_agree", max(vals) - min(vals) <= tol * 2, f"spread {max(vals) - min(vals):.4g} vs 2x tolerance {tol * 2:.4g}") elif fc == "DERIVED": src = cl.get("from") if isinstance(cl.get("from"), list) else [] c(f"{cid}.sources_exist", src and all(s in C for s in src), f"derived from {src}") c(f"{cid}.no_counterfactual_source", all(C.get(s, {}).get("class") != "COUNTERFACTUAL" for s in src), "") c(f"{cid}.no_unverified_only", not src or any(C.get(s, {}).get("class") in ("OBSERVED", "VERIFIED", "DERIVED") for s in src), "a derivation needs at least one evidential source") c(f"{cid}.method_named", isinstance(cl.get("method"), str) and len(cl["method"]) >= 3, "name the computation") elif fc == "COUNTERFACTUAL": c(f"{cid}.no_measurement_refs", not refs, "a counterfactual is not an observation") # derivation graph must be acyclic def cyc(start): seen, stack = set(), [start] while stack: x = stack.pop() for s in (C.get(x, {}).get("from") or []): if s == start: return True if s not in seen: seen.add(s); stack.append(s) return False for cl in claims: if cl.get("class") == "DERIVED": c(f"{cl['id']}.acyclic", not cyc(cl["id"]), "derivation must not cite itself") for cl in claims: # nobody may lean on a counterfactual for s in (cl.get("from") or []): if C.get(s, {}).get("class") == "COUNTERFACTUAL": c(f"{cl['id']}.cites_counterfactual", False, f"cites counterfactual {s}") # cherry-picking: every measurement of this cycle is cited or excluded with a reason cited = {r for cl in claims for r in (cl.get("measurements") or [])} excl = cycle.get("exclusions") if isinstance(cycle.get("exclusions"), list) else [] excluded = {e.get("measurement") for e in excl if isinstance(e, dict) and isinstance(e.get("reason"), str) and len(e["reason"]) >= 5} silent = sorted(mid for mid, m in M.items() if int(m["cycle"]) == n and mid not in cited and mid not in excluded) c("cycle.no_silent_drops", not silent, f"uncited, unexcluded measurements: {silent}" if silent else "every measurement accounted for") unhashed = sorted(mid for mid, m in M.items() if int(m["cycle"]) == n and not (isinstance(m.get("raw_sha256"), str) and _HEX.match(m["raw_sha256"]))) c("cycle.raw_data_kept", not unhashed, f"measurements without a raw-data SHA-256 (excluded ones included): {unhashed}" if unhashed else "every measurement of the cycle keeps its raw-data hash, excluded ones too") # learning from physical evidence rev = cycle.get("revision") if rev is not None: if not isinstance(rev, dict): raise Bad("revision must be an object") basis = rev.get("based_on") if isinstance(rev.get("based_on"), list) else [] c("revision.basis_exists", basis and all(b in C for b in basis), f"based on {basis}") c("revision.evidential", any(C.get(b, {}).get("class") in ("OBSERVED", "VERIFIED", "DERIVED") for b in basis), "a strategy revision must rest on at least one OBSERVED, VERIFIED or DERIVED claim") c("revision.no_counterfactual", all(C.get(b, {}).get("class") != "COUNTERFACTUAL" for b in basis), "") c("revision.uses_this_cycle", any(b in own for b in basis) or any(C.get(b, {}).get("class") == "DERIVED" for b in basis), "the revision must use what this cycle produced (or a derivation over it)") except Bad as e: checks.append({"id": "input.well_formed", "ok": False, "detail": str(e)}) return "NOT_SUPPORTED", checks, {} except Exception as e: # noqa: BLE001 checks.append({"id": "input.error", "ok": False, "detail": f"{type(e).__name__}: {e}"}) return "NOT_SUPPORTED", checks, {} summary = {"cycle": n, "measurements": len(meas), "claims": len(claims), "by_class": {k: sum(1 for cl in claims if cl.get("class") == k) for k in sorted(CLASSES)}} return ("SUPPORTED" if all(x["ok"] for x in checks) else "NOT_SUPPORTED"), 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.discovery.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-003, {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))