#!/usr/bin/env python3 """Stress the reference implementation with random inputs. Exit status 0 means every property held. Non-normative tooling; it needs Python 3.14 and the packages pixi installs (`jsonschema`, `cryptography`). Usage: python3 tools/stress.py [--seed N] [--rounds N] The vectors pin values; this program checks the invariants the chapters state, over inputs drawn from a seeded random generator, so a failure is reproducible from the seed it prints. Canonical JSON round-trips and sorts keys by UTF-16 units; normalization is idempotent; notation folding never raises; random corrections keep the five values equal to a full recomputation, never reuse an ID and apply deterministically; anchors made before those corrections always resolve to one of the five outcomes; Merkle roots agree with a second implementation; spot-check samples do not depend on input order; error bounds are the least six-decimal value with the stated tail; differently stamped pages share a fingerprint; the zip reader and the package validator refuse or accept mutated input without raising and never write outside their folders; rights combine the same in any order; and spot-check chains hundreds deep settle without recursion. """ import argparse import copy import json import os import random import shutil import sys import tempfile import time import zipfile from fractions import Fraction from math import comb ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from tools.oracle import align as AL # noqa: E402 from tools.oracle import anchors as AN # noqa: E402 from tools.oracle import corrections as CO # noqa: E402 from tools.oracle import crypto, fingerprint as FP, hashing, jcs, receipts as RC, rights as RI, targets as TG # noqa: E402 from tools.oracle import package as PK # noqa: E402 from tools.oracle.model import LABELLED, LEAVES, Doc # noqa: E402 from tools.oracle.notation import fold_math # noqa: E402 from tools.oracle.text import NODE_ID_RE, RECORD_ID_RE, letter_first, norm, record_id, sha256_hex, is_ws # noqa: E402 from tools.oracle.tree import Refusal, validate_structure # noqa: E402 from tools.build import ROSTER_BY_ID, SCHEMAS, secret, sentence_doc, tiny_package # noqa: E402 from tools.check import STORE, reference_values # noqa: E402 FAILS = [] COUNTS = {} RULES = {r["id"] for r in json.load(open(os.path.join(ROOT, "registry", "rules.json"), encoding="utf-8"))["rules"]} def ok(section, cond, msg): COUNTS[section] = COUNTS.get(section, 0) + 1 if not cond and len(FAILS) < 200: FAILS.append("%s: %s" % (section, msg)) TEXT_POOL = list("abcdefghijklmnopqrstuvwxyzABCDEFXYZ0123456789 .,;:-()[]") + [ "é", "Å", "fi", "µ", "−", "–", " ", "\t", "\n", "
", "€", "\U0001d49c", "\U0001f600", "é", "H₂O", "10³", "cm⁻¹", " "] KEY_POOL = list("abcxyz") + ["€", "\U0001f600", "fi", "A", "é", "", "a b", "\u0001"] def rand_text(rng, lo=0, hi=12, pool=TEXT_POOL): return "".join(rng.choice(pool) for _ in range(rng.randint(lo, hi))) def rand_json(rng, depth=0): r = rng.random() if depth > 3 or r < 0.4: choice = rng.randint(0, 6) if choice == 0: return None if choice == 1: return rng.random() < 0.5 if choice == 2: return rng.randint(-2**53, 2**53) if choice == 3: return rng.uniform(-1, 1) * 10.0 ** rng.randint(-12, 24) if choice == 4: return rng.choice([0, -0.0, 1e21, 1e-7, 0.000001, 123456789.125, 5e-324, 1.7976931348623157e308]) return rand_text(rng) if r < 0.7: return [rand_json(rng, depth + 1) for _ in range(rng.randint(0, 4))] return {rand_text(rng, 0, 3, KEY_POOL): rand_json(rng, depth + 1) for _ in range(rng.randint(0, 4))} def _as_doubles(text): """Parse JSON as an ECMAScript engine does, which is how canonical JSON is read (vectors/README.md).""" return json.loads(text, parse_int=lambda x: int(x) if abs(int(x)) <= 2**53 else float(x)) def stress_canonical_json(rng, rounds): for _ in range(rounds): v = rand_json(rng) try: s = jcs.dumps(v) except ValueError as e: ok("canonical-json", False, "dumps refused a plain value: %r (%s)" % (v, e)) continue ok("canonical-json", jcs.dumpb(v) == s.encode("utf-8"), "dumpb is dumps encoded") back = _as_doubles(s) ok("canonical-json", back == v, "round trip changed the value: %r" % (v,)) ok("canonical-json", jcs.dumps(back) == s, "canonical form is not idempotent for %r" % (v,)) pairs = _as_doubles(s) if not s.startswith(("{", "[")) else json.loads(s, object_pairs_hook=lambda kv: kv) def ordered(x): if isinstance(x, list) and x and isinstance(x[0], tuple): keys = [k for k, _ in x] return keys == sorted(keys, key=lambda k: k.encode("utf-16-be")) and all(ordered(v) for _, v in x) if isinstance(x, list): return all(ordered(y) for y in x) return True ok("canonical-json", ordered(pairs), "object members are not sorted by UTF-16 code units in %s" % s[:80]) def stress_normalization(rng, rounds): for _ in range(rounds): s = rand_text(rng, 0, 30) n = norm(s) ok("normalize", norm(n) == n, "normalization is not idempotent for %r" % s) ok("normalize", not n or (not is_ws(n[0]) and not is_ws(n[-1])), "normalized text keeps edge whitespace") ok("normalize", " " not in n and all(ch == " " or not is_ws(ch) for ch in n), "normalized text keeps a whitespace run or a non-space whitespace: %r" % n) TEX_TOKENS = list("xyzKrHO0123456789+-=()'") + ["_", "^", "{", "}", "\\alpha", "\\mathrm{H}", "\\frac{a}{b}", " ", "\\,", "\\cdot", "H_2O", "^{14}", "{}", "\\left(", "\\right)", "\\\\", "µ", "−", "''", "_{", "^{"] TYPST_TOKENS = list("xyzKr0123456789+-=()'") + ["_", "^", "(", ")", "alpha", "upright(H)", "\"cm\"", " ", "dot", "plus.minus", "<=", "/", "epsilon.alt", "_(", "^(", "\"a\\b\""] def stress_notation(rng, rounds): for _ in range(rounds): for notation, pool in (("tex", TEX_TOKENS), ("typst", TYPST_TOKENS)): s = "".join(rng.choice(pool) for _ in range(rng.randint(0, 8))) try: a = fold_math(notation, s) b = fold_math(notation, s) except Exception as e: # noqa: BLE001 ok("notation", False, "fold_math raised %r on %s %r" % (e, notation, s)) continue ok("notation", a == b and (a is None or isinstance(a, str)), "fold_math is not deterministic on %r" % s) ok("notation", a is None or not any(is_ws(ch) for ch in a), "a canonical form holds whitespace: %r" % a) ok("notation", a is None or "$" not in a, "a canonical form holds a dollar sign: %r" % a) def _leaf_ids(doc, kinds): return [nid for nid in doc.preorder() if doc.nodes[nid]["kind"] in kinds] def random_correction(rng, st, doc): """One random proposal against `st`, or None.""" sentences = _leaf_ids(doc, {"sentence"}) cells = _leaf_ids(doc, {"cell"}) op = rng.choice(["replace", "replace", "split", "merge", "move", "relabel", "set", "delete", "reclassify", "insert-text", "insert-node"]) evidence = [{"kind": "crop", "sha256": "0" * 64}] if rng.random() < 0.8 else [{"kind": "person", "by": "ana"}] c = {"op": op.split("-")[0], "evidence": evidence} if op == "replace" and (sentences or cells): nid = rng.choice(sentences + cells) text = hashing.plain(doc, nid) if not text: return None i = rng.randrange(len(text)) j = min(len(text), i + rng.randint(1, 6)) c.update(target=nid, find=text[i:j], replace=rand_text(rng, 0, 8)) elif op == "split" and sentences: nid = rng.choice(sentences) c.update(target=nid, offset=rng.randint(0, max(1, len(hashing.plain(doc, nid))))) elif op == "merge" and len(sentences) > 1: nid = rng.choice(sentences) sibs = doc.kids(doc.nodes[nid]["parent"]) k = sibs.index(nid) other = sibs[k + 1] if k + 1 < len(sibs) else rng.choice(sentences) c.update(target=nid, **{"with": other}, joiner=rng.choice([" ", ""])) elif op == "move" and sentences: paragraphs = _leaf_ids(doc, {"paragraph"}) if not paragraphs: return None c.update(target=rng.choice(sentences), parent=rng.choice(paragraphs), place_after=None) elif op == "relabel": labelled = [nid for nid in doc.preorder() if doc.nodes[nid]["kind"] in LABELLED] if not labelled: return None c.update(target=rng.choice(labelled), label=rng.choice([None, "2", "Table 7", "a", rand_text(rng, 1, 5)])) elif op == "set" and cells: c.update(target=rng.choice(cells), attr=rng.choice(["colspan", "rowspan", "role", "column", "level"]), value=rng.choice([1, 2, 3, "2", -1, "data", "row-header", "boss", None])) elif op == "delete" and len(doc.order) > 8: c.update(target=rng.choice(doc.order[1:])) elif op == "reclassify": c.update(target=rng.choice(doc.order), kind=rng.choice(["footnote", "quote", "paragraph", "caption", "section"])) elif op == "insert-text" and sentences: nid = rng.choice(sentences) c.update(target=nid, run=rng.randint(0, len(doc.nodes[nid]["content"])), offset=rng.randint(0, 20), text=rand_text(rng, 1, 6)) elif op == "insert-node": paragraphs = _leaf_ids(doc, {"paragraph"}) if not paragraphs: return None parent = rng.choice(paragraphs) kids = doc.kids(parent) node = {"kind": "sentence", "content": [{"kind": "text", "text": rand_text(rng, 1, 20) or "x"}]} if rng.random() < 0.2: node["foo"] = 1 c.update(parent=parent, place_after=rng.choice(kids + [None]), node=node) else: return None return c def stress_corrections(rng, rounds): """A random walk of corrections from the tiny document; returns the states it passed through.""" tiny = tiny_package(None, receipts=False) st = tiny["final"] base = {"author": {"key": "ana"}, "reason": "Stress.", "at": "2026-10-11T12:00:00Z"} seen = {n["id"] for n in st.nodes} history, accepted = [st], 0 for _ in range(rounds): doc = st.doc() c = random_correction(rng, st, doc) if c is None: continue c = dict(base, **c) try: st1, rec = CO.apply(st, copy.deepcopy(c), store=SCHEMAS) except Refusal as e: ok("corrections", e.rule in RULES, "a refusal names the unregistered rule %s" % e.rule) continue except Exception as e: # noqa: BLE001 ok("corrections", False, "apply raised %r on %s" % (e, json.dumps(c)[:160])) continue accepted += 1 try: validate_structure(st1.doc()) ok("corrections", st1.index().values == reference_values(st1.doc()), "incremental values differ from the full computation after %s" % c["op"]) except Refusal as e: ok("corrections", False, "an accepted correction left the tree invalid: %s" % e.rule) new_ids = {n["id"] for n in st1.nodes} - {n["id"] for n in st.nodes} ok("corrections", not (new_ids & seen), "an ID was reused: %s" % sorted(new_ids & seen)) seen |= new_ids table = {e["id"] for e in st1.ids} ok("corrections", set(rec.get("retired", [])) <= table, "a retired ID is not in the ID table") ok("corrections", st1.ids == sorted(st1.ids, key=CO.id_order), "ids are not in the order of spec/04 §4.4") ok("corrections", rec["id"] == record_id(sha256_hex(jcs.dumpb({k: v for k, v in rec.items() if k != "id"}))), "the record's id does not derive from its content") ok("corrections", rec["index"] == st.rev + 1 == st1.rev, "the record's index is not the next revision") st2, rec2 = CO.apply(st, copy.deepcopy(c), store=SCHEMAS) ok("corrections", rec2 == rec and st2.nodes == st1.nodes and st2.ids == st1.ids, "apply is not deterministic") try: _, rec3 = CO.gate(st, copy.deepcopy(c), SCHEMAS) ok("corrections", rec3 == rec, "the gate records something else than apply") except Refusal as e: ok("corrections", e.rule == "correction.schema", "the gate refused an applied correction with %s" % e.rule) st = st1 history.append(st) ok("corrections", accepted >= rounds // 10, "too few random corrections were accepted: %d of %d" % (accepted, rounds)) return history def stress_anchors(rng, history, rounds): st0, final = history[0], history[-1] doc0, vals0 = st0.doc(), hashing.compute(st0.doc()) outcomes = {"current", "changed-below", "moved", "reattached", "stale"} seen_outcomes = set() anchors = [] for nid in doc0.preorder(): anchors.append(AN.make_anchor(doc0, vals0, nid, 0, store="smithlab")) if doc0.nodes[nid]["kind"] in LEAVES: t = hashing.plain(doc0, nid) if len(t) > 2: i = rng.randrange(len(t) - 1) j = rng.randint(i + 1, len(t)) anchors.append(AN.make_anchor(doc0, vals0, nid, 0, span=[i, j], store="smithlab")) for _ in range(rounds // 4): a = copy.deepcopy(rng.choice(anchors)) a["node"] = letter_first(os.urandom(32), 7) if rng.random() < 0.5 else a["node"] a.pop("kind", None) anchors.append(a) for state in (final, history[len(history) // 2], history[min(1, len(history) - 1)]): values = hashing.compute(state.doc()) for a in anchors: try: r = AN.resolve(state, a) except Exception as e: # noqa: BLE001 ok("anchors", False, "resolve raised %r on %s" % (e, json.dumps(a)[:120])) continue ok("anchors", r["result"] in outcomes, "unknown outcome %s" % r.get("result")) seen_outcomes.add(r["result"]) if r["result"] != "stale" or r.get("node"): ok("anchors", r.get("node") in state.doc().nodes, "a non-stale result names no live node") if r["result"] == "current": ok("anchors", values[r["node"]]["node_hash"] == a["node_hash"], "current with another node_hash") if "span" in r: t = hashing.plain(state.doc(), r["node"]) ok("anchors", 0 <= r["span"][0] <= r["span"][1] <= len(t), "a span outside the node's text") ok("anchors", {"current", "stale"} <= seen_outcomes, "the random walk produced only %s" % sorted(seen_outcomes)) def _merkle_reference(items): leaves = [sha256_hex(b"\x00" + jcs.dumpb(x)) for x in items] def mth(ls): if len(ls) == 1: return ls[0] k = 1 while 2 * k < len(ls): k *= 2 return sha256_hex(b"\x01" + bytes.fromhex(mth(ls[:k])) + bytes.fromhex(mth(ls[k:]))) return mth(leaves) def stress_merkle(rng, rounds): """Receipt items are lists, `["node", id, hash]` or `[kind, record id, hash]` (spec/08 §8.5).""" for _ in range(rounds): n = rng.randint(1, 40) items = [[rng.choice(["node", "annotation", "digitization", "checklist"]), letter_first(os.urandom(32), 7), sha256_hex(os.urandom(8))] for _ in range(n)] if rng.random() < 0.3: items.append([rand_text(rng), [rand_text(rng)], rng.randint(0, 2**53)]) ok("merkle", crypto.merkle_root(items) == _merkle_reference(items), "Merkle roots differ for %d items" % len(items)) def stress_spotcheck(rng, rounds): for _ in range(rounds): n = rng.randint(1, 60) ids = sorted({letter_first(os.urandom(32), 7) for _ in range(n)}) receipt = {"nodes": [[i, sha256_hex(i)] for i in ids]} seed = os.urandom(64) rate, minimum = rng.choice(["0.05", "0.1", "0.5", "1"]), rng.randint(1, 15) sample = RC.sample(receipt, seed, rate, minimum) size = RC.sample_size(len(ids), rate, minimum) ok("spot-check", len(sample) == size == min(len(ids), max(minimum, -(-Fraction(rate) * len(ids) // 1))), "sample size %d is not the rule's for n=%d" % (len(sample), len(ids))) ok("spot-check", sample == sorted(sample) and set(sample) <= set(ids), "sample is not sorted or not from the receipt") shuffled = {"nodes": rng.sample(receipt["nodes"], len(receipt["nodes"]))} ok("spot-check", RC.sample(shuffled, seed, rate, minimum) == sample, "the sample depends on input order") def stress_bounds(rng, rounds): alpha = Fraction(5, 100) for _ in range(rounds // 4): n = rng.randint(1, 120) k = rng.randint(0, n) b = RC.bound_string(k, n) ok("bounds", len(b.split(".")[1]) == 6, "bound has not six decimals: %s" % b) if k >= n: ok("bounds", b == "1.000000", "k = n must give 1") continue p = Fraction(b) def tail(q): return sum(comb(n, i) * q**i * (1 - q)**(n - i) for i in range(k + 1)) ok("bounds", tail(p) <= alpha, "the tail at the bound is above alpha for k=%d n=%d" % (k, n)) if p > 0: ok("bounds", tail(p - Fraction(1, 10**6)) > alpha, "the bound is not the least six-decimal value for k=%d n=%d" % (k, n)) if k + 1 < n: ok("bounds", Fraction(RC.bound_string(k + 1, n)) >= p, "the bound is not monotone in k") def stress_fingerprints(rng, rounds): rules = json.load(open(os.path.join(ROOT, "registry", "stamps", "scienting-stamps-1.json"), encoding="utf-8"))["rules"] for _ in range(rounds): lines = [{"text": rand_text(rng, 1, 40)} for _ in range(rng.randint(1, 8))] page = {"size": [612.0, 792.0], "lines": lines, "images": [sha256_hex(rand_text(rng)) for _ in range(rng.randint(0, 2))]} stamped = copy.deepcopy(page) stamped["lines"].insert(rng.randint(0, len(lines)), {"text": "Downloaded via EXAMPLE UNIV on %s" % rand_text(rng, 1, 10, list("abcXYZ0123456789-:"))}) fp, removed = FP.fingerprint(page, rules) fp2, removed2 = FP.fingerprint(stamped, rules) ok("fingerprints", fp == fp2 and len(removed2) == len(removed) + 1, "a stamped download has another fingerprint") digit_lines = [i for i, l in enumerate(lines) if any(ch in "0123456789" for ch in l["text"])] if digit_lines: changed = copy.deepcopy(page) i = rng.choice(digit_lines) t = changed["lines"][i]["text"] j = next(j for j, ch in enumerate(t) if ch in "0123456789") changed["lines"][i]["text"] = t[:j] + str((int(t[j]) + 1) % 10) + t[j + 1:] ok("fingerprints", FP.fingerprint(changed, rules)[0] != fp, "a changed digit kept the fingerprint") spaced = copy.deepcopy(page) spaced["lines"] = [{"text": " ".join(l["text"])} for l in spaced["lines"]] ok("fingerprints", FP.fingerprint(spaced, rules)[0] == fp, "inserted spaces changed the fingerprint") def stress_zip(rng, rounds): src = os.path.join(ROOT, "fixtures", "packages", "tiny2026example" + PK.ZIP_EXTENSION) with open(src, "rb") as f: good = f.read() accepted = 0 with tempfile.TemporaryDirectory() as tmp: for i in range(rounds): data = bytearray(good) mode = rng.random() if mode < 0.6: for _ in range(rng.randint(1, 3)): data[rng.randrange(len(data))] = rng.randrange(256) elif mode < 0.8: data = data[:rng.randrange(len(data))] else: data = data[:rng.randrange(len(data))] + os.urandom(rng.randint(1, 64)) + data[rng.randrange(len(data)):] path = os.path.join(tmp, "z%d.zip" % i) with open(path, "wb") as f: f.write(data) try: rep = PK.validate_zip(path, STORE) except Exception as e: # noqa: BLE001 ok("zip", False, "validate_zip raised %r on mutation %d" % (e, i)) continue accepted += bool(rep) dest = os.path.join(tmp, "d%d" % i) try: PK.read_zip(path, dest) except Refusal as e: ok("zip", e.rule in RULES, "read_zip names the unregistered rule %s" % e.rule) except Exception as e: # noqa: BLE001 ok("zip", False, "read_zip raised %r on mutation %d" % (e, i)) stray = [f for f in os.listdir(tmp) if not (f.startswith("z") and f.endswith(".zip")) and not f.startswith("d")] ok("zip", not stray, "the zip reader wrote outside its folder: %s" % stray) ok("zip", accepted <= rounds // 20 + 1, "%d of %d mutated archives were accepted" % (accepted, rounds)) def stress_packages(rng, rounds): src = os.path.join(ROOT, "fixtures", "packages", "tiny2026example") files = sorted(os.path.relpath(os.path.join(d, f), src) for d, _, fs in os.walk(src) for f in fs) with tempfile.TemporaryDirectory() as tmp: for i in range(rounds): pkg = os.path.join(tmp, "p%d" % i) shutil.copytree(src, pkg) mode = rng.choice(["flip", "flip", "truncate", "delete", "add", "rename-case", "symlink"]) target = rng.choice(files) full = os.path.join(pkg, target) if mode == "flip": data = bytearray(open(full, "rb").read()) if data: k = rng.randrange(len(data)) data[k] = (data[k] + rng.randrange(1, 256)) % 256 open(full, "wb").write(data) elif mode == "truncate": data = open(full, "rb").read() open(full, "wb").write(data[:rng.randrange(len(data) + 1)]) elif mode == "delete": os.remove(full) elif mode == "add": open(os.path.join(pkg, rand_text(rng, 1, 8, list("abcxyz")) + ".txt"), "wb").write(b"stray\n") elif mode == "rename-case": if target.lower() == target.upper(): continue os.rename(full, os.path.join(pkg, target.upper())) else: os.remove(full) os.symlink(os.path.join(src, target), full) try: rep = PK.validate(pkg, STORE) except Exception as e: # noqa: BLE001 ok("packages", False, "validate raised %r after %s of %s" % (e, mode, target)) continue ok("packages", not rep, "a package mutated by %s of %s validated" % (mode, target)) ok("packages", all(e["rule"] in RULES for e in rep.errors), "an unregistered rule was reported") shutil.rmtree(pkg) def stress_rights(rng, rounds): tri = [True, False, "unknown"] keys = list(RI.USES) + ["ai/" + p for p in RI.AI_PAIRS] for _ in range(rounds): flats = [{k: rng.choice(tri) for k in keys} for _ in range(rng.randint(1, 4))] combined = RI.combine(flats) shuffled = RI.combine(rng.sample(flats, len(flats))) ok("rights", combined == shuffled, "combining depends on order") for k in keys: vals = [f[k] for f in flats] want = False if False in vals else ("unknown" if "unknown" in vals else True) ok("rights", combined[k] == want, "combining %s gave %r for %r" % (k, combined[k], vals)) def _align_items(rng, prefix, texts): items = [] for i, t in enumerate(texts): runs = [{"kind": "text", "text": t}] doc, s = sentence_doc(runs) strict, cx = hashing.strict(doc, s) items.append({"id": "%s%d" % (prefix, i), "container": "p%d" % (i // 3), "context": "paragraph", "runs": runs, "strict": strict, "lossy": hashing.lossy_from_strict(strict), "complex": cx}) return items def stress_alignment(rng, rounds): for _ in range(rounds // 4): texts = [rand_text(rng, 5, 60) or "x" for _ in range(rng.randint(1, 12))] a = _align_items(rng, "a", texts) try: same = AL.align(a, _align_items(rng, "b", texts)) b_texts = [t for t in texts if rng.random() > 0.3] + [rand_text(rng, 5, 30)] diff = AL.align(a, _align_items(rng, "b", b_texts)) ok("alignment", AL.align(a, _align_items(rng, "b", b_texts)) == diff, "alignment is not deterministic") except Exception as e: # noqa: BLE001 ok("alignment", False, "align raised %r" % (e,)) continue ok("alignment", len(same["pairs"]) == len(a) and all(p["agree"] for p in same["pairs"]) and not [s for s in same["suspects"] if s["kind"] not in ("structure", "join")], "aligning two equal readings left disagreements or unpaired sentences: %s" % same["suspects"][:3]) ids_a = {x["id"] for x in a} ok("alignment", all(p["a"] in ids_a for p in diff["pairs"]), "a pair names an unknown sentence") paired = {p["a"] for p in diff["pairs"]} dropped = {s["a"] for s in diff["suspects"] if s["kind"] == "dropped"} ok("alignment", paired | dropped == ids_a and not (paired & dropped), "every sentence of a reading is paired or dropped, not both or neither") def stress_addresses(rng, rounds): for _ in range(rounds): store = rng.choice([None, "smithlab", "a1", "x.y-z"]) key = rng.choice(["hale2026harmonic", "k", "a.b_c-d"]) node = letter_first(os.urandom(32), 7) pin = sha256_hex(rand_text(rng))[:6] if rng.random() < 0.5 else None span = sorted([rng.randint(0, 50), rng.randint(0, 50)]) if rng.random() < 0.5 else None s = AN.format_address(store, key, node, pin, span) parsed = AN.parse_address(s) ok("addresses", parsed and AN.format_address(parsed["store"], parsed["key"], parsed["node"], parsed["pin"], parsed["span"]) == s, "address does not round trip: %s" % s) junk = rand_text(rng, 0, 20) try: AN.parse_address(junk) except Exception as e: # noqa: BLE001 ok("addresses", False, "parse_address raised %r on %r" % (e, junk)) for _ in range(rounds): d = os.urandom(32) ok("ids", NODE_ID_RE.match(letter_first(d, 7)) and RECORD_ID_RE.match(record_id(d.hex())), "ID form") def stress_targets(rng, rounds): st = PK.read_state(os.path.join(ROOT, "fixtures", "packages", "hale2026harmonic")) doc = st.doc() for _ in range(rounds // 4): anchored = rng.sample(doc.order, rng.randint(0, 6)) sets = {} for rule in TG.DIGIT_RULES: got = TG.target_set(doc, anchored, rule) ids = [n for n, _ in got] pos = {nid: i for i, nid in enumerate(doc.order)} ok("targets", ids == sorted(ids, key=pos.get) and len(set(ids)) == len(ids), "targets are not in document order") ok("targets", all(doc.nodes[n]["kind"] in LEAVES for n in ids), "a target is not a leaf") sets[rule] = set(ids) ok("targets", sets["none"] <= sets["anchored-sections"] <= sets["all"], "the digit rules do not nest") try: TG.target_set(doc, anchored + ["zzzzzzz"], "all") ok("targets", False, "an unknown anchored node was accepted") except ValueError: ok("targets", True, "") def stress_receipt_chains(rng, rounds): """Chains of spot-checks hundreds deep settle without recursion, and alternate as spec/08 §8.6 implies.""" depth = max(50, rounds) ids = sorted({letter_first(os.urandom(32), 7) for _ in range(20)}) rec = RC.seal({"format": "scienting-receipt/1", "doc": "hale2026harmonic", "rev": 0, "level": "F3", "scope": "all", "nodes": [[i, sha256_hex(i)] for i in ids], "evidence": [], "by": "ana", "at": "2026-10-11T12:00:00Z"}, secret("ana"), "ana") records = [rec] for i in range(depth): orig = records[-1] seed = RC.make_seed(orig, secret("ben-spot")) sample = RC.sample(orig, seed) hashes = dict(orig["nodes"]) q = RC.seal({"format": "scienting-receipt/1", "doc": "hale2026harmonic", "rev": 0, "level": "F3", "scope": "qa:" + orig["id"], "nodes": [[n, hashes[n]] for n in sample], "evidence": [], "by": "ben", "at": "2026-10-11T12:%02d:%02dZ" % (i // 60 % 60, i % 60), "qa": {"of": orig["id"], "key": "ben-spot", "seed": __import__("base64").b64encode(seed).decode(), "rate": "0.05", "minimum": 10, "sample": sample, "errors": [{"node": sample[0]}]}}, secret("ben"), "ben") records.append(q) t0 = time.time() try: status = RC.statuses(records, ROSTER_BY_ID) except RecursionError: ok("receipts", False, "a chain of %d spot-checks overflowed the stack" % depth) return ok("receipts", time.time() - t0 < 30, "statuses of a %d-deep chain took too long" % depth) # The last spot-check nobody checks is valid; each receipt it checks is suspended; and so on, alternating. want = "valid" for r in reversed(records): ok("receipts", status[r["id"]] == want, "receipt at depth %d is %s, not %s" % (records.index(r), status[r["id"]], want)) want = "suspended" if want == "valid" else "valid" def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ap.add_argument("--seed", type=int, default=20261014, help="the random seed (default 20261014)") ap.add_argument("--rounds", type=int, default=200, help="cases per property (default 200)") a = ap.parse_args() rng = random.Random(a.seed) print("seed %d, %d rounds" % (a.seed, a.rounds)) stress_canonical_json(rng, a.rounds) stress_normalization(rng, a.rounds) stress_notation(rng, a.rounds) history = stress_corrections(rng, a.rounds) stress_anchors(rng, history, a.rounds) stress_merkle(rng, a.rounds) stress_spotcheck(rng, a.rounds) stress_bounds(rng, a.rounds) stress_fingerprints(rng, a.rounds) stress_zip(rng, a.rounds) stress_packages(rng, a.rounds // 4) stress_rights(rng, a.rounds) stress_alignment(rng, a.rounds) stress_addresses(rng, a.rounds) stress_targets(rng, a.rounds) stress_receipt_chains(rng, a.rounds) for k in sorted(COUNTS): print("%-17s %6d checks" % (k, COUNTS[k])) if FAILS: print("\n%d failures (seed %d):" % (len(FAILS), a.seed)) for f in FAILS[:60]: print(" " + f) sys.exit(1) print("\nEvery property held.") if __name__ == "__main__": main()