#!/usr/bin/env python3 """Independent verifier of the timeliness corpus (extensions/timeliness.md). Written from the extension's text. It shares no code with wai-rs, which generates the corpus, and never runs it. It takes from the second implementations of the texts this extension builds on what they define: group and object receipts from the JWP corpus's (`../jwp-conformance/jwp_verify.py`, jwp-receipts §2.2, §3); a claim's steps 1 to 5 from the claim corpus's (`../claim-conformance/claim_verify.py`, jwp-receipts §3.1), whose step 4 holds §4.2's rules; a transmission claim's step 6, a schedule's, and a receiver's passes from the carousel corpus's (`carousel_verify.py`: bearer-receipts §4.3, carousel §4.3 and §7); a holdings record's form, signature and holdings from the holdings corpus's (`holdings_verify.py`, holdings §4.4, §5.4); and a WAI1 envelope's manifest from the staged corpus's reader (SPEC §2, §3). Everything else this extension adds is written here: - §4.3: a timeliness claim against its group, the object receipts and envelopes held, and each rung's group and object receipt; - §5: an entry's value at an instant, and its end; - §6: the intents that stand, the records that stand and the held-at statements they give (`held_from_records`, with §2's `before_group`), each intent's lines (`lines`), and a statement offered; - §7: the plan's inputs, each source's arrival, and the choice, the ladder's rung included; - §7.5: a sink's run of its plans against its clock (`run`), each plan reading the claims and records held by its instant. For every case it requires the outcome the corpus states, field for field; that every reason of §9's table, every outcome of §6.3's and every action §7.5.3 states appears in the corpus; then it mutates copies of the corpus, one defect each, and requires every copy to fail; and it drops each of its own checks in turn and requires the corpus to catch the weakened verifier. A bound no vector carries is checked directly. python3 timeliness_verify.py Exit status 0 only if every case verifies, every mutated copy fails, every weakened verifier is caught and every direct check holds. Python 3.8 or later; the `cryptography` package, which the claim verifier imports. """ import copy import json import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, HERE) for d in ("claim-conformance", "jwp-conformance"): sys.path.insert(0, os.path.join(HERE, "..", d)) import claim_verify as CV # noqa: E402 import jwp_verify as JV # noqa: E402 import carousel_verify as KV # noqa: E402 import holdings_verify as HV # noqa: E402 SV = CV.SV # the staged corpus's strict WAI1 reader, which the claim verifier loads EXTENSION = os.path.join(HERE, "..", "extensions", "timeliness.md") SPEC = os.path.join(HERE, "..", "SPEC.md") JWP_RECEIPTS = os.path.join(HERE, "..", "extensions", "jwp-receipts.md") TELEMETRY = os.path.join(HERE, "..", "extensions", "delivery-telemetry.md") FULL = 1_000_000 # §2: a value in parts per million of the full value LAST_INSTANT = 253_402_300_799_999 # §2: 9999-12-31T23:59:59.999Z MAX_INT = (1 << 53) - 1 ACTIONS = ("fetch", "abandon", "present", "passed", "ended") # §7.5.3 T0_PLUS_LATER = 1_793_664_000_000 + 3_600_000 # An instant past every plan of the corpus's runs. # A weakened run drops one of these checks (see `WEAKENINGS`); None checks everything. DROPPED = None def drops(name): return DROPPED == name def text_of(path): with open(path, encoding="utf-8") as f: return f.read() def table_rows(text, heading): """The first backticked cell of each row of the table under `heading`, up to the next heading of its level or above.""" at = text.index("\n" + heading) level = heading.split(" ")[0] ends = [text.find("\n" + h + " ", at + 1) for h in ("#" * n for n in range(2, len(level) + 1))] ends = [e for e in ends if e >= 0] rows = text[at:min(ends) if ends else len(text)].splitlines() return [r.split("`")[1] for r in rows if r.startswith("| `")] def reasons_table(): """§9's codes, in order, each with the section the table says raises it.""" text = text_of(EXTENSION) at = text.index("\n## 9. Reasons") end = text.find("\n## ", at + 1) rows = text[at:end].splitlines() return [(r.split("`")[1], r.split("|")[2].strip()) for r in rows if r.startswith("| `")] def outcomes_table(): """§6.3's outcomes, in order.""" return table_rows(text_of(EXTENSION), "### 6.3 Lines") def refinement_only(): """SPEC §5, *Refinement capabilities*: the capabilities that are refinement-only.""" return set(table_rows(text_of(SPEC), "### Refinement capabilities (staged delivery)")) REFINEMENT_ONLY = refinement_only() def need(ok, reason): """jwp-receipts §3.1 step 6.""" if not ok: raise CV.Reject(6, reason) # ------------------------------------------------------------------ what the verifier holds # Steps 1 to 4 of a claim, a group's steps 1 and 2 with its objects', and a record's reading are # pure in their bytes: each is computed once and kept, which nothing a weakened run drops reads. _CLAIMS, _GROUPS, _RECORDS = {}, {}, {} def claim_steps(wire): """jwp-receipts §3.1 steps 1 to 4 (claim_verify.py). Returns the claim; raises CV.Reject.""" if wire not in _CLAIMS: try: _CLAIMS[wire] = CV.verify_claim(wire) except CV.Reject as r: _CLAIMS[wire] = r v = _CLAIMS[wire] if isinstance(v, CV.Reject): raise v return v def read_record(b): """holdings §4.4 (holdings_verify.py). Returns the record; raises HV.Refused.""" if b not in _RECORDS: try: _RECORDS[b] = HV.read(b) except HV.Refused as r: _RECORDS[b] = r v = _RECORDS[b] if isinstance(v, HV.Refused): raise v return v class Group: """A group given, after jwp-receipts §3 steps 1 and 2, with its object receipts and envelopes.""" def __init__(self, item): self.receipt = JV.read_group(item["group"]) JV.verify_group(self.receipt) self.link = JV.link(self.receipt).hex() self.count = self.receipt["leaf_count"] self.signer = self.receipt["signer_pubkey"] self.start = JV.seconds(self.receipt["group_start"]) self.verdicts = [] # bearer-receipts §6.2 step 2, one for each object receipt given self.pairs = [] # [(object receipt, envelope or None)] for each that reads self.included = {} # position -> content hash of each object receipt that passes step 5 self.envelopes = {} # content hash -> envelope bytes held for o in item["objects"]: try: ob = JV.read_object(o["receipt"]) except JV.Refuse as r: self.verdicts.append("reject reason=" + r.reason) continue env = None if o["envelope"] is None else bytes.fromhex(o["envelope"]) self.pairs.append((ob, env)) try: if env is None: JV.inclusion(ob, self.receipt) self.verdicts.append("inclusion-verified") else: JV.accept(ob, self.receipt, env) self.verdicts.append("delivered-verified") except JV.Refuse as r: self.verdicts.append("reject reason=" + r.reason) try: JV.inclusion(ob, self.receipt) self.included[ob["object_id"]] = ob["content_hash"].hex() except JV.Refuse: pass if env is not None: self.envelopes[JV.blake3(JV.OBJECT + env).hex()] = env def read_groups(items): """§6.2 step 1: each group under jwp-receipts §3 steps 1 and 2, and each of its object receipts as bearer-receipts §6.2 step 2 checks it. Raises JV.Refuse for a group that fails.""" out = [] for item in items: key = json.dumps(item, sort_keys=True) if key not in _GROUPS: try: _GROUPS[key] = Group(item) except JV.Refuse as r: _GROUPS[key] = r g = _GROUPS[key] if isinstance(g, JV.Refuse): raise g out.append(g) return out def manifest(env): """A WAI1 envelope's manifest (SPEC §2, §3), or None for bytes that are not one.""" try: m, _, _ = SV.read_wai1(env) except SV.NotWai1: return None return m def rungs(entry): """§4.1: an entry's ladder, in its order: none for `null`, the one object, or the array's.""" fb = entry["fallback"] return [] if fb is None else fb if isinstance(fb, list) else [fb] def checked_rungs(entry, check): """The rungs §4.3 reads, every one; a weakened run reads only the first.""" return rungs(entry)[:1] if drops(check) else rungs(entry) def is_layer(env): """§4.3 step 6: a WAI1 whose capability is refinement-only (SPEC §5).""" m = manifest(env) return m is not None and m["model_requirement"]["capability"] in REFINEMENT_ONLY def derived_from(env, motif): """§4.3 step 8: a WAI1 whose manifest carries `derived_from` (SPEC §7.1, *Derived renditions*: its four members in their forms) with `motif` the primary's content hash. A WAI2 is none.""" m = manifest(env) df = None if m is None else m.get("derived_from") return (isinstance(df, dict) and all(isinstance(df.get(k), str) and df[k] for k in ("motif", "motif_capability", "renderer", "settings")) and CV.is_hex(df["motif"], 32) and CV.is_hex(df["settings"], 32) and df["motif"] == motif) # ------------------------------------------------------------------ §2: the held-at statement def before_group(word_s, start): """§2: a held-at statement is before its group when the whole seconds of the time its word states, any fraction dropped (`word_s`), are below the seconds jwp-receipts §2.4.3 reads from the group's `group_start` (`start`, None where it is not a date-time, which nothing is before). §4.3 step 4 reads an entry's `from_ms` by the same rule.""" if start is None: return False return word_s <= start if drops("same_second") else word_s < start # ------------------------------------------------------------------ §4.3: step 6 def check_timeliness(c, g, groups): """§4.3 against the group `g` the claim is checked against, which passed jwp-receipts §3 steps 1 and 2, and every group given (`groups`, by link hash) with the object receipts and envelopes held under it.""" need(g.start is not None, "context_invalid") entries = c["body"]["objects"] envelopes = {} for x in groups.values(): envelopes.update(x.envelopes) need(drops("subject") or g.link == c["subject"], "timeliness_subject_mismatch") need(drops("signer") or g.signer == c["signer_pubkey"], "timeliness_signer_mismatch") need(drops("range") or all(e["leaf_index"] < g.count for e in entries), "timeliness_leaf_out_of_range") need(drops("before") or not any(before_group(e["from_ms"] // 1000, g.start) for e in entries), "timeliness_before_group") need(drops("content") or all(g.included.get(e["leaf_index"], e["content_hash"]) == e["content_hash"] for e in entries), "timeliness_content_mismatch") named = [e["content_hash"] for e in entries] + [f["content_hash"] for e in entries for f in checked_rungs(e, "rung_layer")] need(drops("layer") or not any(h in envelopes and is_layer(envelopes[h]) for h in named), "timeliness_on_layer") for f in (f for e in entries for f in checked_rungs(e, "rung_position")): fg = groups.get(f["group"]) if fg is not None and not drops("fallback_position"): need(f["leaf_index"] < fg.count and fg.included.get(f["leaf_index"], f["content_hash"]) == f["content_hash"], "timeliness_fallback_mismatch") for e in entries: for f in checked_rungs(e, "rung_derived"): if f["content_hash"] in envelopes and not drops("derived"): need(derived_from(envelopes[f["content_hash"]], e["content_hash"]), "timeliness_fallback_not_derived") # ------------------------------------------------------------------ §5: value over time def value(entry, h): """§5.1: the entry's value at the instant h, floored.""" by = entry["by_ms"] if h <= by: return FULL pts = [(by, FULL)] + [(p["at_ms"], p["value_ppm"]) for p in entry["decay"]] for (ta, va), (tb, vb) in zip(pts, pts[1:]): if ta < h <= tb: num, den = va * (tb - h) + vb * (h - ta), tb - ta return (2 * num + den) // (2 * den) if drops("floor") else num // den return 0 def end(entry): """§5.2: the `at_ms` of its last point.""" return entry["decay"][-1]["at_ms"] # ------------------------------------------------------------------ §6: judging delivery def claims_step(wires, groups, plan, chained=True): """§6.2 step 2, and §7.2's (`plan`): each claim under jwp-receipts §3.1 steps 1 to 5, each signer's previous claim the last of its claims held before it, then step 6 against the group whose link hash is its subject, or the first group. Returns each claim's label, and each claim accepted (None for one refused). `chained` is False only in a weakened run.""" by_link = {} for g in groups: by_link.setdefault(g.link, g) held, labels, accepted = {}, [], [] for wire in wires: try: c = claim_steps(wire) except CV.Reject as r: labels.append(f"reject step={r.step} reason={r.reason}") accepted.append(None) continue previous = held.get(c["signer_pubkey"]) held[c["signer_pubkey"]] = c try: if chained: CV.check_chain(c, previous) g = by_link.get(c["subject"], groups[0] if groups else None) if c["class"] == "timeliness": need(g is not None, "timeliness_subject_mismatch") check_timeliness(c, g, by_link) elif plan and c["class"] == "transmission" and g is not None: KV.check_transmission(c, g.receipt) elif plan and c["class"] == "schedule" and g is not None: KV.check_schedule(c, g.receipt, g.pairs) elif not drops("context"): raise CV.Reject(6, "context_mismatch") except CV.Reject as r: labels.append(f"reject step={r.step} reason={r.reason}") accepted.append(None) continue labels.append("accepted") accepted.append(c) return labels, accepted def intents_step(accepted, groups): """§6.2 step 3: the entries of the accepted claims by group and `leaf_index`, in the order of the groups given and then of `leaf_index`: [(group, leaf_index, entry, or None when set aside)].""" stated = {} for c in accepted: if c is not None and c["class"] == "timeliness": for e in c["body"]["objects"]: stated.setdefault((c["subject"], e["leaf_index"]), []).append(e) out, seen = [], set() for g in groups: if g.link in seen: continue seen.add(g.link) for i in sorted(i for link, i in stated if link == g.link): es = stated[(g.link, i)] stands = drops("intent_contradiction") or all(e == es[0] for e in es) out.append((g.link, i, es[0] if stands else None)) return out def records_step(records, groups): """§6.2 step 4: each record as holdings §4.4 reads and verifies it, refused whole when its entry for a group given misstates the leaf count or is before the group, two of one holder at one `as_of_ms` that differ set aside. Returns each record's label and the records that are not refused (index -> record).""" by_link = {g.link: g for g in groups} labels, kept = [], {} for k, hexed in enumerate(records): try: rec = read_record(bytes.fromhex(hexed)) mine = [(by_link[g.hex()], n) for g, n, _ in rec[3] if g.hex() in by_link] if not drops("leaf_count") and any(g.count != n for g, n in mine): raise HV.Refused("holdings_leaf_count_mismatch") if not drops("held_before") and any(before_group(rec[1] // 1000, g.start) for g, _ in mine): raise HV.Refused("held_before_group") kept[k] = rec labels.append("stands") except HV.Refused as r: labels.append("reject reason=" + r.reason) at = {} for k, rec in kept.items(): at.setdefault((rec[0], rec[1]), []).append(k) for ks in at.values(): if len({HV.digest(kept[k]) for k in ks}) > 1 and not drops("record_contradiction"): for k in ks: labels[k] = "set-aside reason=holdings_contradicted" return labels, kept def held_from_records(standing, group, index): """§6.2 step 5: the held-at statements the records that stand give for object `index` of `group` (a link hash in hexadecimal): (holder, group, index, instant) for each record that holds it (holdings §5.4), its instant the record's `as_of_ms`.""" g = bytes.fromhex(group) return [(rec[0].hex(), group, index, rec[1]) for rec in standing if HV.holds(rec, g, index) is None] def has_entry(rec, group): return any(g.hex() == group for g, _, _ in rec[3]) def line_holders(group, entry, standing): """§6.3: the holders with a record that stands and has an entry for the intent's group or a rung's group, in the order of their first such record.""" rs = checked_rungs(entry, "rung_holders") out = [] for rec in standing: h = rec[0].hex() if h not in out and (drops("line_holders") or has_entry(rec, group) or any(has_entry(rec, r["group"]) for r in rs)): out.append(h) return out def grade(entry, p, fs): """§6.3's table, the first outcome that holds, for a holder whose least instants for the primary and for each rung of the ladder are p and fs (None where none).""" by = entry["by_ms"] if p is not None and (p < by if drops("deadline_inclusive") else p <= by): return "on_time", FULL # The first rung, in the ladder's order, held by the deadline. in_time = [r for r, f in zip(rungs(entry), fs) if f is not None and (f <= by or drops("fallback_deadline"))] if in_time: r = in_time[-1] if drops("rung_order") else in_time[0] # §5.3: what arrived after the deadline grades nothing, so a later primary of more value # does not displace a rung held by then. if not (drops("sink_rule") and p is not None and value(entry, p) > r["value_ppm"]): return "replaced", r["value_ppm"] if p is not None: return "after_deadline", value(entry, p) return "not_shown", 0 def lines(entry, holders, primary, fallback): """§6.3: an intent's line for each of `holders` (keys in hexadecimal), in increasing order of key bytes, from that holder's own held-at statements for its primary and for the rungs of its fallback, each statement naming its object. *f* is one instant for a ladder of one rung, and the list of them for an array.""" def least(statements, holder, latest=False): instants = [s[3] for s in statements if drops("own_statements") or s[0] == holder] if not instants: return None return max(instants) if latest or drops("least") else min(instants) out = [] for h in (holders if drops("key_order") else sorted(holders, key=bytes.fromhex)): p = least(primary, h) fs = [least([s for s in fallback if (s[1], s[2]) == (r["group"], r["leaf_index"])], h, drops("least_fallback")) for r in rungs(entry)] outcome, v = grade(entry, p, fs) f = fs if isinstance(entry["fallback"], list) else (fs[0] if fs else None) out.append({"holder": h, "held_ms": p, "fallback_ms": f, "outcome": outcome, "value_ppm": v}) return out def intent_lines(link, entry, standing): primary = held_from_records(standing, link, entry["leaf_index"]) fallback = [s for r in rungs(entry) for s in held_from_records(standing, r["group"], r["leaf_index"])] return lines(entry, line_holders(link, entry, standing), primary, fallback) def verdict(case): """§6: the verdict, or `refuse reason=R` for a group that fails.""" try: groups = read_groups(case["groups"]) except JV.Refuse as r: return "refuse reason=" + r.reason labels, accepted = claims_step(case["claims"], groups, plan=False) intents = intents_step(accepted, groups) rlabels, kept = records_step(case["records"], groups) standing = [kept[k] for k in sorted(kept) if rlabels[k] == "stands"] out, found = [], {} for link, i, entry in intents: ls = [] if entry is None else intent_lines(link, entry, standing) found[(link, i)] = (entry, ls) out.append({"group": link, "leaf_index": i, "intent": "stands" if entry is not None else "set-aside reason=timeliness_contradicted", "end_ms": None if entry is None else end(entry), "lines": ls}) statement = None s = case["statement"] if s is not None: entry, ls = found.get((s["group"], s["index"]), (None, [])) if entry is None and not drops("no_intent"): statement = "reject reason=deadline_not_stated" else: line = next((x for x in ls if x["holder"] == s["holder"]), None) statement = ({"outcome": "not_shown", "value_ppm": 0} if line is None else {"outcome": line["outcome"], "value_ppm": line["value_ppm"]}) return {"objects": [g.verdicts for g in groups], "claims": labels, "records": rlabels, "intents": out, "statement": statement} # ------------------------------------------------------------------ §7: planning def sources_valid(sources, loss, h): """§7.1: the sources, each exactly a bearer instance, a holder or the origin.""" if not isinstance(sources, list) or (not sources and not drops("sources_empty")): return False bearers, holders, origins = set(), set(), 0 for s in sources: if not isinstance(s, dict): return False kind = s.get("kind") fetch = s.get("fetch_ms") if kind != "bearer" and not (drops("sources_fetch") or (CV.is_int(fetch) and fetch >= 1)): return False if kind == "bearer": if set(s) != {"kind", "bearer_id"} or not (drops("sources_uri") or CV.is_uri(s["bearer_id"])): return False if s["bearer_id"] in bearers and not drops("sources_twice"): return False bearers.add(s["bearer_id"]) elif kind == "holder": if set(s) != {"kind", "holder", "fetch_ms"} or not (drops("sources_key") or CV.is_hex(s["holder"], 32)): return False if s["holder"] in holders and not drops("sources_twice"): return False holders.add(s["holder"]) elif kind == "origin": if set(s) != {"kind", "fetch_ms"}: return False origins += 1 elif not drops("sources_kind"): return False if origins > 1 and not drops("sources_twice"): return False return not (bearers and (loss is None or h is None)) or drops("sources_loss") def plan_inputs(loss, h, t, sources): """§7.1's checks, in their order: the reason the plan is refused, or None.""" if loss is not None and not drops("loss_bound") and not (CV.is_int(loss) and loss <= 999_999): return "loss_bound_invalid" if h is not None and not drops("overhead") and not (CV.is_int(h) and h <= 2): return "overhead_invalid" if t != "from_ms" and not drops("instant") and not (CV.is_int(t) and t <= LAST_INSTANT): return "instant_invalid" if not sources_valid(sources, loss, h): return "sources_invalid" return None def hash_and_position(v, kind): """§7.1: the `group` of a wanted object or a statement is 64 lowercase hexadecimal digits, and its `index` an integer from 0 to 2^53 - 1.""" if not (drops(f"{kind}_group") or CV.is_hex(v.get("group"), 32)): return False index = v.get("index") if drops(f"{kind}_index"): return True return CV.is_int(index) and not (drops(f"{kind}_greatest") and index == MAX_INT) def wanted_formed(wanted): """§7.1: a list, each entry an object with a `group` and an `index`.""" if not isinstance(wanted, list): return drops("wanted_list") return all(hash_and_position(w, "wanted") if isinstance(w, dict) else drops("wanted_object") for w in wanted) def wanted_list(wanted): """What a verifier without §7.1's rule for the list reads of one: nothing wanted.""" return wanted if isinstance(wanted, list) else [] def named(w): """What a wanted entry names. A verifier without §7.1's rule reads an entry that is no object as naming none, so as an object with no intent.""" return (w.get("group"), w.get("index")) if isinstance(w, dict) else (None, None) def statement_bearer(y, kind): """§7.1: a `bearer_id` that is a string and an absolute URI.""" bearer = y.get("bearer_id") if not isinstance(bearer, str): return drops(f"{kind}_bearer_string") return drops(f"{kind}_uri") or CV.is_uri(bearer) def symbols_formed(symbols): """§7.1: a list, each statement an object with a `bearer_id` that is an absolute URI, a `group`, an `index` and a list of at least one number, each an integer from 0 to 2^53 - 1; and no two of one object on one instance.""" if not isinstance(symbols, list): return drops("symbols_list") seen = set() for y in symbols: if not isinstance(y, dict): if drops("symbols_statement"): continue return False if not statement_bearer(y, "symbols") or not hash_and_position(y, "symbols"): return False if not statement_numbers(y.get("held"), "symbols"): return False key = (y.get("bearer_id"), y.get("group"), y.get("index")) if key in seen and not drops("symbols_twice"): return False seen.add(key) return True def statement_numbers(held, kind): """§7.1: a list of at least one number, each an integer from 0 to 2^53 - 1.""" if not isinstance(held, list): return drops(f"{kind}_held_list") if not held and not drops(f"{kind}_empty"): return False if not drops(f"{kind}_integer") and any(type(n) is not int for n in held): return False return drops(f"{kind}_number") or all(CV.is_int(n) for n in held if type(n) is int) def entry_on(schedules, bearer_id, group, index): """§7.2: the entry for an object in the schedule that stands for its group on an instance, or None.""" body = schedules.get((group, bearer_id)) return None if body is None else next((x for x in body["objects"] if x["leaf_index"] == index), None) def holding_fits(e, held): """§7.2, by carousel §7's own check: one number for each source block, and no more of a `no-code` block than it has. Whatever else the plan's loss bound is does not enter into it.""" p = KV.plan({"entry": e, "loss_ppm": 0, "overhead": 0, "at_ms": 0, "held": held}) return p.get("reason") != "holding_invalid" def schedules_step(labels, accepted): """§7.2: the declarations (bearer-receipts §6.2 step 4) and the schedules (carousel §6.2 step 4), each of its group. Relabels the claims; returns the schedules that stand, by (group, bearer_id).""" def key(c): return c["body"]["bearer_id"] if drops("per_group") else (c["subject"], c["body"]["bearer_id"]) def by_instance(cls): out = {} for i, c in enumerate(accepted): if c is not None and c["class"] == cls: out.setdefault(key(c), []).append(i) return out declared, contradicted = {}, set() for k, idx in by_instance("transmission").items(): bodies = [accepted[i]["body"] for i in idx] same = all(b == bodies[0] for b in bodies) if same: declared[k] = bodies[0]["bearer"] else: contradicted.add(k) for i in idx: labels[i] = "declared" if same else "set-aside reason=bearer_contradicted" schedules = {} for k, idx in by_instance("schedule").items(): bodies = [accepted[i]["body"] for i in idx] if not KV.monotone(bodies) and not drops("schedule_contradicted"): aside = "schedule_contradicted" elif k in contradicted: aside = "bearer_contradicted" elif k not in declared and not drops("undeclared"): aside = "bearer_not_declared" elif declared.get(k) == "bidirectional" and not drops("bidirectional"): aside = "schedule_on_bidirectional" else: aside = None # The last of an instance's schedules is the one that stands (carousel §6.2 step 4). last = bodies[0] if drops("schedule_last") else bodies[-1] for i in idx: if aside is not None: labels[i] = "set-aside reason=" + aside else: labels[i] = "stands" if accepted[i]["body"] == last else "extended" schedules[(accepted[i]["subject"], accepted[i]["body"]["bearer_id"])] = last return schedules def standing_at(kept, rlabels, s): """§7.2: the records that stand at s, by holdings §5.1 to §5.3 over those §6.2 step 4 did not refuse, and not set aside there.""" live = [rec for k, rec in kept.items() if drops("record_at_s") or rec[1] <= s < rec[2]] out = [] for holder in {rec[0] for rec in live}: mine = [rec for rec in live if rec[0] == holder] latest = max(rec[1] for rec in mine) top = [rec for rec in mine if rec[1] == latest] if len({HV.digest(rec) for rec in top}) == 1: out.append(top[0]) aside = {HV.digest(kept[k]) for k in kept if rlabels[k] != "stands"} return [rec for rec in out if HV.digest(rec) not in aside] def arrival(src, group, index, s, ctx): """§7.2: a source's arrival of object `index` of `group` planned from s, or None.""" kind = src["kind"] if kind == "origin": return s + src["fetch_ms"] if kind == "holder": recs = standing_at(ctx["kept"], ctx["rlabels"], s) held = any((drops("holder_record") or rec[0].hex() == src["holder"]) and HV.holds(rec, bytes.fromhex(group), index) is None for rec in recs) return s + src["fetch_ms"] if held else None if kind != "bearer": return None e = entry_on(ctx["schedules"], src["bearer_id"], group, index) if e is None: return None blocks = sum(n for _, n in CV.schedule_blocks(e)) held = [0] * blocks for y in ctx["symbols"]: same_instance = y["bearer_id"] == src["bearer_id"] or drops("symbols_instance") if same_instance and y["group"] == group and y["index"] == index and not drops("symbols_ignored"): held = y["held"] break p = KV.plan({"entry": e, "loss_ppm": ctx["loss"], "overhead": ctx["h"], "at_ms": s, "held": held}) return p["complete_ms"] def choose(entry, primary, ladder): """§7.3 for an object whose intent stands, from each source's arrivals of the primary and, for each rung of the ladder in its order, of that rung.""" by = entry["by_ms"] def choice(outcome, obj, r, i, arrived, v): return {"outcome": outcome, "object": obj, "rung": r, "source": i, "arrival_ms": arrived, "value_ppm": v} timely = [i for i, a in enumerate(primary) if a is not None and a <= by] if timely: i = min(timely, key=lambda j: (primary[j], j)) if drops("plan_order") else timely[0] return choice("on_time", "primary", None, i, primary[i], FULL) late = [(value(entry, a), i) for i, a in enumerate(primary) if a is not None] best = max((v for v, _ in late), default=0) # Rule 2: the first rung of at least the best late value that arrives by the deadline, from the # first such source in the asker's order. order = list(enumerate(rungs(entry))) if drops("plan_rung_first"): order = order[:1] if drops("plan_rung_order"): order.reverse() for r, fb in order: at_least = fb["value_ppm"] > best if drops("fallback_tie") else fb["value_ppm"] >= best if not (at_least or drops("fallback_value")): continue arrivals = ladder[r] in_time = [i for i, a in enumerate(arrivals) if a is not None and (a < by if drops("plan_fallback_deadline") else a <= by)] if in_time and drops("fallback_order"): in_time.sort(key=lambda j: (arrivals[j], j)) if in_time: return choice("replaced", "fallback", r, in_time[0], arrivals[in_time[0]], fb["value_ppm"]) if late and (best >= 1 or drops("late_zero")): ties = [i for v, i in late if v == best] i = ties[-1] if drops("late_tie") else ties[0] return choice("after_deadline", "primary", None, i, primary[i], best) return choice("no_source", None, None, None, None, 0) def plan(case): """§7: the plan, or `refuse reason=R`.""" loss, h, t, sources = case["loss_ppm"], case["overhead"], case["at_ms"], case["sources"] bad = plan_inputs(loss, h, t, sources) if drops("wanted_before_sources") and not wanted_formed(case["wanted"]): return "refuse reason=wanted_invalid" if bad: return "refuse reason=" + bad if drops("wanted_before_symbols") and not wanted_formed(case["wanted"]): return "refuse reason=wanted_invalid" symbols = case.get("symbols", []) if not symbols_formed(symbols): return "refuse reason=symbols_invalid" if not wanted_formed(case["wanted"]): return "refuse reason=wanted_invalid" # A verifier without those rules reads what it cannot count as nothing held; a formed list stays whole. symbols = [y for y in symbols if isinstance(y, dict) and all(k in y for k in ("bearer_id", "group", "index")) and isinstance(y.get("held"), list) and not any(isinstance(n, str) for n in y["held"])] \ if isinstance(symbols, list) else [] try: groups = read_groups(case["groups"]) except JV.Refuse as r: return "refuse reason=" + r.reason labels, accepted = claims_step(case["claims"], groups, plan=True, chained=not case.get("unchained")) intents = {(link, i): entry for link, i, entry in intents_step(accepted, groups)} rlabels, kept = records_step(case["records"], groups) schedules = schedules_step(labels, accepted) # §7.2: a statement against the entry that stands for its object on its instance; one with no such # entry is compared with nothing. if not drops("symbols_holding"): for y in symbols: e = entry_on(schedules, y["bearer_id"], y["group"], y["index"]) if e is not None and not holding_fits(e, y["held"]): return "refuse reason=holding_invalid" ctx = {"kept": kept, "rlabels": rlabels, "schedules": schedules, "loss": loss, "h": h, "symbols": symbols} choices = [] for w in wanted_list(case["wanted"]): entry = intents.get(named(w)) if entry is None: choices.append("reject reason=deadline_not_stated") continue s = entry["from_ms"] if t == "from_ms" else t primary = [arrival(src, w["group"], w["index"], s, ctx) for src in sources] ladder = [[arrival(src, r["group"], r["leaf_index"], s, ctx) for src in sources] for r in rungs(entry)] choices.append(choose(entry, primary, ladder)) return {"claims": labels, "records": rlabels, "choices": choices} # ------------------------------------------------------------------ §7.5: running a plan def run_inputs(case): """§7.5.1's checks, in their order: the reason the run is refused, or None.""" plans = case["plans"] if not plans and not drops("run_plans_none"): return "plans_invalid" if any(p["at_ms"] == "from_ms" for p in plans) and not drops("run_plans_from_ms"): return "plans_invalid" # §7.5.1 check 1: one that is not an integer reads as above every other, and §7.1 refuses it. instants = [p["at_ms"] if type(p["at_ms"]) is int and p["at_ms"] >= 0 else float("inf") for p in plans] if not drops("run_plans_order") and any(a >= b for a, b in zip(instants, instants[1:])): return "plans_invalid" if drops("run_wanted_before_twice") and not all(wanted_formed(p["wanted"]) for p in plans): return "wanted_invalid" if not drops("run_plans_twice"): for p in plans: wanted = wanted_list(p["wanted"]) # An entry §7.1 refuses names no object: that is check 2's, not this one's. objects = [json.dumps(w, sort_keys=True) for w in wanted] if drops("run_wanted_malformed_twice") \ else [named(w) for w in wanted if wanted_formed([w])] if len(set(objects)) != len(objects): return "plans_invalid" for p in plans: if drops("run_wanted_before_sources") and not wanted_formed(p["wanted"]): return "wanted_invalid" bad = plan_inputs(case["loss_ppm"], case["overhead"], p["at_ms"], p["sources"]) if bad: return bad if not drops("run_wanted_late") and not wanted_formed(p["wanted"]): return "wanted_invalid" arrivals = case["arrivals"] if not drops("run_arrivals_last") and any(not CV.is_int(a["at_ms"]) or a["at_ms"] > LAST_INSTANT for a in arrivals): return "arrivals_invalid" if not drops("run_arrivals_order") and any(a["at_ms"] > b["at_ms"] for a, b in zip(arrivals, arrivals[1:])): return "arrivals_invalid" if not drops("run_arrivals_twice") and len({(a["group"], a["index"]) for a in arrivals}) != len(arrivals): return "arrivals_invalid" def formed(kind, items, key, seconds): if key not in case or (case[key] is None and drops(f"run_{kind}_null")): return True return dated_formed(kind, items, case[key], seconds) checks = [("claims_invalid", formed("claims", case["claims"], "claims_at_ms", claim_seconds)), ("records_invalid", formed("records", case["records"], "records_at_ms", record_seconds)), ("groups_invalid", formed("groups", [json.dumps(g["group"], sort_keys=True) for g in case["groups"]], "groups_at_ms", group_seconds) and ("receipts_at_ms" not in case or receipts_formed(case["groups"], case["receipts_at_ms"]))), ("symbols_invalid", symbols_trace_formed(case.get("symbols", [])))] # Checks 4 to 7, and each weakening that swaps two neighbours of them. for k, name in enumerate(("run_dated_order", "run_order_records", "run_order_groups")): if drops(name): checks[k], checks[k + 1] = checks[k + 1], checks[k] for reason, ok in checks: if not ok: return reason return None def group_seconds(key): """§7.5.1 check 6: the whole seconds jwp-receipts §2.4.3 reads from the `group_start` of a group that passes jwp-receipts §3 steps 1 and 2, or None: one that fails them starts at no instant it can be held to.""" item = json.loads(key) try: r = JV.read_group(item) if not drops("run_groups_unverified"): JV.verify_group(r) return JV.seconds(r["group_start"]) except JV.Refuse: return None def receipts_formed(groups, at): """§7.5.1 check 6: a list for each group, an entry for each of its object receipts, none past the last instant.""" if not isinstance(at, list): return drops("run_receipts_not_list") if len(at) != len(groups) and not drops("run_receipts_groups"): return False for g, a in zip(groups, at): if not isinstance(a, list): if drops("run_receipts_member_not_list"): continue return False if len(a) != len(g["objects"]) and not drops("run_receipts_objects"): return False if not drops("run_receipts_last") and any(t is not None and (not CV.is_int(t) or t > LAST_INSTANT) for t in a): return False return True def symbols_trace_formed(trace): """§7.5.1 check 7: each instant an integer to the last instant, none below the one before it; each statement as §7.1 forms it; of one object on one instance, each statement at a later instant than the one before it, of as many source blocks, and with no number below that one's.""" if not isinstance(trace, list): return drops("run_symbols_list") or (trace is None and drops("run_symbols_null")) last, previous = {}, 0 for s in trace: if not isinstance(s, dict): if drops("run_symbols_statement"): continue return False at, held = s.get("at_ms"), s.get("held") if not drops("run_symbols_last") and (not CV.is_int(at) or at > LAST_INSTANT): return False if at < previous and not drops("run_symbols_order"): return False previous = at if not statement_bearer(s, "run_symbols") or not hash_and_position(s, "run_symbols"): return False if not statement_numbers(held, "run_symbols"): return False key = (s.get("bearer_id"), s.get("group"), s.get("index")) if key in last: was_at, was = last[key] if was_at == at and not drops("run_symbols_twice"): return False if len(was) != len(held) and not drops("run_symbols_blocks"): return False if any(b < a for a, b in zip(was, held)) and not drops("run_symbols_fall"): return False last[key] = (at, held) return True def claim_seconds(wire): """§7.5.1 check 4: the whole seconds jwp-receipts §2.4.3 reads from the `issued_at` of a claim that passes steps 1 to 4, or None: one that fails them is not known to be its signer's.""" try: return JV.seconds(claim_steps(wire)["issued_at"]) except CV.Reject: return JV.seconds(json.loads(wire)["issued_at"]) if drops("run_claims_unverified") else None def record_seconds(hexed): """§7.5.1 check 5: the whole seconds of a record's `as_of_ms`, where holdings §4.4 reads and verifies it, or None.""" try: return read_record(bytes.fromhex(hexed))[1] // 1000 except HV.Refused: return None def dated_formed(kind, items, at, seconds): """§7.5.1 checks 4 to 6: one entry for each; none past the last instant; no statement given twice; none held in a whole second before its own.""" if not isinstance(at, list): return drops(f"run_{kind}_not_list") if len(at) != len(items) and not drops(f"run_{kind}_length"): return False if not drops(f"run_{kind}_last") and any(a is not None and (not CV.is_int(a) or a > LAST_INSTANT) for a in at): return False if not drops(f"run_{kind}_twice") and len(set(items)) != len(items): return False if drops(f"run_{kind}_signed"): return True for item, a in zip(items, at): since = seconds(item) if a is None or since is None: continue # A statement's own second admits an instant anywhere in it. if (a < since * 1000 + 999 if drops(f"run_{kind}_signed_ms") else a // 1000 < since): return False return True def read_by(at, i, t, kind): """§7.5.1: what is held by t: its instant is at most t, or it has none.""" if not isinstance(at, list) or i >= len(at) or at[i] is None or drops(f"run_{kind}_undated"): return True return at[i] < t if drops(f"run_{kind}_strict") else at[i] <= t def read_at(case, t): """§7.5.1: a run's inputs for its plan at t: the claims and records held by then, in the order given, and their positions.""" claims = [i for i, _ in enumerate(case["claims"]) if read_by(case.get("claims_at_ms"), i, t, "claims")] records = [i for i, _ in enumerate(case["records"]) if read_by(case.get("records_at_ms"), i, t, "records")] return claims, records def envelope_held(case, g, o, t): """§7.5.1: an envelope is held from its object's arrival; one whose object no arrival names, from before the first plan. A group or receipt that does not read names no object.""" if drops("run_envelopes_undated"): return True try: link, index = JV.link(JV.read_group(g["group"])).hex(), JV.read_object(o["receipt"])["object_id"] except JV.Refuse: return True a = next((a for a in case["arrivals"] if a["group"] == link and a["index"] == index), None) if a is None: return not drops("run_envelopes_no_arrival") return a["at_ms"] < t if drops("run_envelopes_strict") else a["at_ms"] <= t def groups_read(case, t): """§7.5.1: the groups a plan at t reads. A run that dates none reads each whole, envelopes and all; one that dates either reads what the sink held by t.""" at_groups, at_receipts = case.get("groups_at_ms"), case.get("receipts_at_ms") if at_groups is None and at_receipts is None and not drops("run_envelopes_opt_in"): return case["groups"] out = [] for i, g in enumerate(case["groups"]): if not read_by(at_groups, i, t, "groups"): continue at = at_receipts[i] if isinstance(at_receipts, list) and i < len(at_receipts) else None objects = [dict(o, envelope=o["envelope"] if envelope_held(case, g, o, t) else None) for k, o in enumerate(g["objects"]) if read_by(at, k, t, "receipts")] out.append({"group": g["group"], "objects": objects}) return out def symbols_read(case, t): """§7.5.1: what the sink holds at t: of each object on each instance, the last statement by t, as §7.1 gives it.""" read = {} trace = case.get("symbols", []) for s in [s for s in trace if isinstance(s, dict)] if isinstance(trace, list) else []: if not (drops("run_symbols_undated") or (s["at_ms"] < t if drops("run_symbols_strict") else s["at_ms"] <= t)): continue key = (s.get("bearer_id"), s.get("group"), s.get("index")) if key not in read or not drops("run_symbols_latest"): read[key] = {k: s[k] for k in ("bearer_id", "group", "index", "held") if k in s} return list(read.values()) def not_held(n, read, labels): """§7.5.3: each of n statements' labels, `not-held` for one the plan did not read.""" out = ["not-held"] * n for i, label in zip(read, labels): out[i] = label return out def same_source(a, b): """§7.5.2: both name one bearer instance or one holder's key, or both are the origin, whatever their `fetch_ms`.""" if a["kind"] != b["kind"]: return False if drops("run_same_fetch_ms") and a.get("fetch_ms") != b.get("fetch_ms"): return False return a.get("bearer_id") == b.get("bearer_id") and a.get("holder") == b.get("holder") class Fetch: """§7.5.2: the object a fetch takes, its source, and the plan it serves.""" def __init__(self, obj, rung, takes, plan, index, source): self.obj, self.rung, self.takes, self.plan, self.index, self.source = obj, rung, takes, plan, index, source def takes(key, entry, choice): """The object a choice takes: the primary, or the rung it names.""" if choice["object"] == "primary": return key r = rungs(entry)[choice["rung"]] return (r["group"], r["leaf_index"]) def run(case): """§7.5: the run of a sink's plans against its arrivals, or `refuse reason=R`.""" bad = run_inputs(case) if bad: return "refuse reason=" + bad # Each plan derived by §7.2 and §7.3 from what the sink held by its instant, with its own # sources and wanted objects; a group that fails refuses the run as it refuses a plan. plans, read, intents_of = [], [], [] dated = ("claims_at_ms" in case or "records_at_ms" in case or ("groups_at_ms" in case and not drops("run_groups_read")) or ("receipts_at_ms" in case and not drops("run_receipts_read"))) unchained = dated and drops("run_claims_chain") for p in case["plans"]: cs, rs = read_at(case, p["at_ms"]) claims, records = [case["claims"][i] for i in cs], [case["records"][i] for i in rs] held_groups = groups_read(case, p["at_ms"]) derived = plan(dict(case, at_ms=p["at_ms"], sources=p["sources"], wanted=p["wanted"], claims=claims, records=records, groups=held_groups, symbols=symbols_read(case, p["at_ms"]), unchained=unchained)) if isinstance(derived, str): return derived plans.append(derived["choices"]) read.append({"claims": not_held(len(case["claims"]), cs, derived["claims"]), "records": not_held(len(case["records"]), rs, derived["records"])}) groups = read_groups(held_groups) _, accepted = claims_step(claims, groups, plan=True, chained=not unchained) intents_of.append({(link, i): entry for link, i, entry in intents_step(accepted, groups)}) def objects_run(k): """§7.5.2: what plan k wants whose choice is not refused, in its wanted order, each with its choice and the entry that plan read.""" return [((w["group"], w["index"]), c, intents_of[k][(w["group"], w["index"])]) for w, c in zip(wanted_list(case["plans"][k]["wanted"]), plans[k]) if isinstance(c, dict)] instants = {p["at_ms"] for p in case["plans"]} | {a["at_ms"] for a in case["arrivals"]} for k in range(len(plans)): for _, _, e in objects_run(k): instants |= {e["by_ms"], end(e)} held, settled, passed, fetches, force, out = {}, set(), set(), {}, None, [] def decide(n, k, action, key, obj=None, rung=None, source=None, v=None): out.append({"at_ms": n, "plan": k, "action": action, "group": key[0], "index": key[1], "object": obj, "rung": rung, "source": source, "value_ppm": v}) def abandon(n, key): f = fetches.pop(key) decide(n, f.plan, "abandon", key, f.obj, f.rung, f.index) for n in sorted(instants): # Step 1. for a in case["arrivals"]: if a["at_ms"] == n: arrived = (a["group"], a["index"]) held[arrived] = n if not drops("run_arrival_completes"): for key in [key for key, f in fetches.items() if f.takes == arrived]: del fetches[key] # Step 2. new = next((k for k, p in enumerate(case["plans"]) if p["at_ms"] == n), None) if new is not None: sources = case["plans"][new]["sources"] choices = {key: (c, e) for key, c, e in objects_run(new)} def continues(key, f): c, e = choices.get(key, (None, None)) if c is None or c["object"] is None or drops("run_continue"): return False return takes(key, e, c) == f.takes and same_source(sources[c["source"]], f.source) if force is not None: for key, _, _ in objects_run(force): if key not in settled and key in fetches and not continues(key, fetches[key]): if drops("run_drop_abandon") and key not in choices: continue abandon(n, key) force = new for key, c, e in objects_run(new): if (key in settled and not drops("run_settled_refetch")) or c["object"] is None: continue if drops("run_symbols_hold") and c["arrival_ms"] == n and sources[c["source"]]["kind"] == "bearer": # §7.5.1: symbols are not an arrival, however many are stated. held[takes(key, e, c)] = n if takes(key, e, c) in held and not drops("run_held_fetch"): continue f = fetches.get(key) if f is not None and continues(key, f): if not drops("run_continued_plan"): f.plan, f.index = new, c["source"] continue fetches[key] = Fetch(c["object"], c["rung"], takes(key, e, c), new, c["source"], sources[c["source"]]) decide(n, new, "fetch", key, c["object"], c["rung"], c["source"]) if force is None: continue current = objects_run(force) if drops("run_passed_keeps_running"): # §7.5.4: an object a later plan has no intent for is no longer run, passed or not. run_now = {key for key, _, _ in current} current = current + [x for k in range(force) for x in objects_run(k) if x[0] in passed and x[0] not in run_now] # Step 3, by what the sink holds by `by_ms`. for key, _, e in current: if key in settled or key in passed or e["by_ms"] > n: continue limit = n if drops("run_by_deadline") else e["by_ms"] holds = [(r, f) for r, f in enumerate(rungs(e)) if held.get((f["group"], f["leaf_index"]), limit + 1) <= limit] if held.get(key, limit + 1) <= limit: settled.add(key) decide(n, force, "present", key, "primary", v=FULL) elif holds: r, f = holds[-1] if drops("run_rung_order") else holds[0] settled.add(key) decide(n, force, "present", key, "fallback", r, v=f["value_ppm"]) else: passed.add(key) decide(n, force, "passed", key) if key in fetches: after_present = key in settled and not drops("run_present_abandon") after_passed = key in passed and fetches[key].obj == "fallback" and not drops("run_passed_abandon") if after_present or after_passed: abandon(n, key) # Step 4, at the primary's value at its arrival. for key, _, e in current: if key in passed and key not in settled and key in held: v = value(e, n if drops("run_late_value") else held[key]) if v >= 1 or drops("run_late_zero"): settled.add(key) decide(n, force, "present", key, "primary", v=v) # Step 5. for key, _, e in current: if key in passed and key not in settled and end(e) <= n: if key in fetches and not drops("run_end_abandon"): abandon(n, key) settled.add(key) if not drops("run_ended"): decide(n, force, "ended", key) return {"plans": plans, "read": read, "decisions": out} if dated else {"plans": plans, "decisions": out} # ------------------------------------------------------------------ the corpus def claim_outcome(c): try: claim_steps(c["claim"]) return "accept" except CV.Reject as r: return f"reject step={r.step} reason={r.reason}" def value_outcome(c): """§5 for an entry read alone, as §3.1 steps 2 and 4 read an entry.""" e = c["entry"] CV.timeliness_entry_body(e) CV.timeliness_entry_rules(e, None) return {"value_ppm": value(e, c["at_ms"]), "end_ms": end(e)} RUNS = (("claims", claim_outcome), ("values", value_outcome), ("verdicts", verdict), ("plans", plan), ("runs", run)) KINDS = tuple(k for k, _ in RUNS) def strings(v, out): if isinstance(v, str): out.append(v) elif isinstance(v, list): for x in v: strings(x, out) elif isinstance(v, dict): for x in v.values(): strings(x, out) def check(doc, quiet=False): """The failures of a corpus document: each case's outcome, every reason of §9, every outcome of §6.3 and every action of §7.5.3.""" fails = [] for kind, run in RUNS: for c in doc[kind]: try: got = run(c) except CV.Reject as r: got = {"error": f"reject step={r.step} reason={r.reason}"} except (KeyError, TypeError, ValueError, IndexError, AttributeError, StopIteration, ZeroDivisionError) as e: got = {"error": repr(e)} if got != c["outcome"]: fails.append(c["name"]) if not quiet: print(f"FAIL {c['name']}\n stated {json.dumps(c['outcome'])}\n got {json.dumps(got)}") stated = [] for kind in KINDS: for c in doc[kind]: strings(c["outcome"], stated) for code, _ in reasons_table(): if not any(s == code or s.endswith("reason=" + code) for s in stated): fails.append("no case gives " + code) for outcome in outcomes_table(): if outcome not in stated: fails.append("no case gives the outcome " + outcome) for action in ACTIONS: if action not in stated: fails.append("no case gives the action " + action) return fails def mutations(doc): """Copies of the corpus, one defect each, every one of which must fail.""" def case(d, name): return next(c for k in KINDS for c in d[k] if c["name"] == name) def m_instant(d): # A millisecond before the end is 75; at the end, 0. case(d, "example_last_ms")["at_ms"] += 1 def m_end(d): case(d, "hard_one_ms")["outcome"]["end_ms"] -= 1 def m_floor(d): case(d, "step_mid")["outcome"]["value_ppm"] += 1 def m_accept(d): case(d, "decay_near_collinear")["outcome"] = "reject step=4 reason=timeliness_decay_invalid" def m_line(d): case(d, "holder_one_ms_after")["outcome"]["intents"][0]["lines"][0]["value_ppm"] = 999_801 def m_earliest(d): c = case(d, "earliest_record_counts") c["records"] = c["records"][1:] def m_statement(d): case(d, "statement_line")["statement"]["holder"] = d["keys"]["holder_c"] def m_envelope(d): c = case(d, "on_layer") c["groups"][0]["objects"][3]["envelope"] = None def m_order(d): c = case(d, "order_over_earliest") c["sources"] = list(reversed(c["sources"])) def m_deadline(d): case(d, "on_time_at_deadline")["at_ms"] += 1 def m_fetch(d): # Both late sources at 300,000: the first in order, the origin. case(d, "late_choice_by_value")["sources"][1]["fetch_ms"] = 66_000 def m_loss(d): # 70%: 3 symbols a pass of the 12 needed, a fourth pass. case(d, "every_pass_from_first")["loss_ppm"] = 700_000 def m_from_ms(d): case(d, "publisher_check_fails")["at_ms"] = d["plans"][0]["at_ms"] def m_object(d): case(d, "fallback_from_carousel")["outcome"]["choices"][0]["object"] = "primary" def m_reason_gone(d): d["verdicts"].remove(case(d, "record_before_group")) def m_extended(d): case(d, "plan_bearer_extended")["outcome"]["claims"][1] = "stands" def m_withdrawn(d): # The withdrawn pass's schedule given first and last alike: one body, which stands. c = case(d, "plan_bearer_withdrawn") c["claims"][2] = c["claims"][1] def m_rung_extended(d): # The rung above the one the extended schedule reaches. case(d, "ladder_rung_on_extended_schedule")["outcome"]["choices"][0]["rung"] = 0 def m_rung_value(d): # The lower rung presented at the higher's value. case(d, "ladder_lower_rung_alone")["outcome"]["intents"][0]["lines"][0]["value_ppm"] = 700_000 def m_rung_instant(d): # The lower rung a millisecond late: nothing held by the deadline. r = case(d, "ladder_lower_rung_alone") r["outcome"]["intents"][0]["lines"][0]["fallback_ms"][1] += 1 def m_rung_form(d): # A ladder's line reports its list. r = case(d, "ladder_first_rung_replaces") r["outcome"]["intents"][0]["lines"][0]["fallback_ms"] = r["outcome"]["intents"][0]["lines"][0]["fallback_ms"][0] def m_plan_rung(d): case(d, "ladder_lower_rung")["outcome"]["choices"][0]["rung"] = 0 def m_arrival_late(d): # The trace a millisecond later is a late object at another value. case(d, "run_late_object")["arrivals"][0]["at_ms"] += 1 def m_arrival_gone(d): case(d, "run_on_time")["arrivals"] = [] def m_decision_value(d): case(d, "run_late_object")["outcome"]["decisions"][-1]["value_ppm"] += 1 def m_decision_action(d): case(d, "run_every_rung_passed_no_primary")["outcome"]["decisions"][1]["action"] = "ended" def m_plan_dropped(d): c = case(d, "run_replaced_changes_source") c["plans"] = c["plans"][:1] def m_plan_instant(d): case(d, "run_replaced_at_deadline")["plans"][1]["at_ms"] += 1 def m_decision_order(d): c = case(d, "run_in_wanted_order") c["outcome"]["decisions"][0], c["outcome"]["decisions"][1] = c["outcome"]["decisions"][1], c["outcome"]["decisions"][0] def m_dates_dropped(d): # The schedules given undated: every plan reads the extension, and the fetches differ. c = case(d, "run_extension_reaches_mid_run") del c["claims_at_ms"] def m_date_later(d): # Past the plan that read it. case(d, "run_extension_reaches_mid_run")["claims_at_ms"][3] = T0_PLUS_LATER def m_read_label(d): case(d, "run_extension_reaches_mid_run")["outcome"]["read"][1]["claims"][3] = "not-held" def m_dated_fetch(d): # The rung fetched at T0, as the undated run does. first = case(d, "run_extension_reaches_mid_run")["outcome"]["decisions"][0] first["object"], first["rung"], first["source"] = "fallback", 1, 1 def m_withdrawn_read(d): # The withdrawal never read: every schedule stands in the second plan. c = case(d, "run_contradiction_mid_run") c["claims_at_ms"][3] = T0_PLUS_LATER def m_parent_first(d): # The parent held with its child: nothing is refused. c = case(d, "run_claim_parent_not_held_yet") c["claims_at_ms"][0] = c["claims_at_ms"][1] def m_before_signed(d): # Within the claim's own second: no refusal. case(d, "claims_held_before_signed")["claims_at_ms"][0] += 1 def m_record_date(d): del case(d, "run_record_reaches_mid_run")["records_at_ms"] def m_claims_reason_gone(d): d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=claims_invalid"] def m_records_reason_gone(d): d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=records_invalid"] def m_reason_gone_run(d): d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=arrivals_invalid"] def m_symbols_gone(d): # Holding nothing, no rung is in reach: the rung the case states is not chosen. del case(d, "run_symbols_reach_a_rung")["symbols"] def m_symbols_later(d): # Past the only plan. case(d, "run_symbols_reach_a_rung")["symbols"][0]["at_ms"] = T0_PLUS_LATER def m_symbols_fewer(d): # 6 of 12 takes two passes, so the rung comes a pass later. case(d, "run_symbols_reach_a_rung")["symbols"][0]["held"] = [6] def m_symbols_instance(d): case(d, "run_symbols_reach_a_rung")["symbols"][0]["bearer_id"] = "urn:example:bearer:b3" def m_symbols_arrival_stated(d): case(d, "run_symbols_reach_a_rung")["outcome"]["plans"][0][0]["arrival_ms"] += 1 def m_symbols_fetch_stated(d): d_ = case(d, "run_symbols_reach_a_rung")["outcome"]["decisions"] d_[:] = d_[1:] def m_plan_symbols_gone(d): del case(d, "plan_symbols_reach_a_rung")["symbols"] def m_symbols_reason_gone(d): for k in ("plans", "runs"): d[k] = [c for c in d[k] if c["outcome"] != "refuse reason=symbols_invalid"] def m_group_date_dropped(d): del case(d, "run_group_reaches_mid_run")["groups_at_ms"] def m_group_date_later(d): case(d, "run_group_reaches_mid_run")["groups_at_ms"][0] = T0_PLUS_LATER def m_envelope_read_early(d): # With no list the envelope is read throughout, so the schedule is refused from T0. del case(d, "run_envelope_read_from_its_arrival")["groups_at_ms"] def m_envelope_arrival_gone(d): # An envelope whose object no arrival names was held from before the first plan. case(d, "run_envelope_read_from_its_arrival")["arrivals"] = [] def m_receipt_date_dropped(d): del case(d, "run_object_receipt_reaches_mid_run")["receipts_at_ms"] def m_groups_reason_gone(d): d["runs"] = [c for c in d["runs"] if c["outcome"] != "refuse reason=groups_invalid"] def m_passed_then_presented(d): # A primary arriving after an entry is set aside is not presented. case(d, "run_entry_contradicted_after_passed")["outcome"]["decisions"].append( case(d, "run_late_object")["outcome"]["decisions"][-1]) def m_contradiction_early(d): # Read from before the first plan, the entry is set aside before anything is fetched. case(d, "run_entry_contradicted_after_passed")["claims_at_ms"][1] = None def m_enough_then_arrives(d): # With an arrival the sink holds the object, and presents it at the deadline. case(d, "run_symbols_enough_no_arrival")["arrivals"] = [{"group": d["groups"]["F"], "index": 0, "at_ms": 1_793_664_021_000}] def restored(name, pointer, value): """A copy in which one case written over holds its well-formed member: its refusal goes.""" def m(d): target = case(d, name) for key in pointer[:-1]: target = target[key] target[pointer[-1]] = value m.__name__ = "m_restored_" + name return m bearer_b2 = "urn:example:bearer:b2" def statement(d): return [{"bearer_id": bearer_b2, "group": d["groups"]["F"], "index": 0, "at_ms": 1_793_664_020_000, "held": [10]}] def symbols_restored(name): def m(d): case(d, name)["symbols"] = statement(d) m.__name__ = "m_restored_" + name return m def plan_symbols_restored(name): def m(d): case(d, name)["symbols"] = [{k: v for k, v in statement(d)[0].items() if k != "at_ms"}] m.__name__ = "m_restored_" + name return m def wanted_restored(name, pointer, as_list=False): """A copy in which one case written over names a wanted object again.""" def m(d): target = case(d, name) for key in pointer[:-1]: target = target[key] formed = {"group": d["groups"]["G"], "index": 0} target[pointer[-1]] = [formed] if as_list else formed m.__name__ = "m_restored_" + name return m def statement_restored(name, instant): """A copy in which one case written over holds a formed statement again.""" def m(d): formed = {"bearer_id": "urn:example:bearer:b3", "group": d["groups"]["F"], "index": 0, "held": [10]} case(d, name)["symbols"][0] = dict(formed, at_ms=1_793_664_020_000) if instant else formed m.__name__ = "m_restored_" + name return m def m_wanted_reason_gone(d): for k in ("plans", "runs"): d[k] = [c for c in d[k] if c["outcome"] != "refuse reason=wanted_invalid"] def m_wanted_twice_reason(d): # One object wanted twice stays the plans' reason. case(d, "run_wanted_twice_and_malformed")["outcome"] = "refuse reason=wanted_invalid" def m_not_an_instant_restored(d): for name, key in (("claims_at_not_an_instant", "claims_at_ms"), ("records_at_not_an_instant", "records_at_ms"), ("groups_at_not_an_instant", "groups_at_ms")): case(d, name)[key][0] = 1_793_664_002_000 return [m_instant, m_end, m_floor, m_accept, m_line, m_earliest, m_statement, m_envelope, m_order, m_deadline, m_fetch, m_loss, m_from_ms, m_object, m_reason_gone, m_extended, m_withdrawn, m_rung_extended, m_rung_value, m_rung_instant, m_rung_form, m_plan_rung, m_arrival_late, m_arrival_gone, m_decision_value, m_decision_action, m_plan_dropped, m_plan_instant, m_decision_order, m_reason_gone_run, m_dates_dropped, m_date_later, m_read_label, m_dated_fetch, m_withdrawn_read, m_parent_first, m_before_signed, m_record_date, m_claims_reason_gone, m_records_reason_gone, m_symbols_gone, m_symbols_later, m_symbols_fewer, m_symbols_instance, m_symbols_arrival_stated, m_symbols_fetch_stated, m_plan_symbols_gone, m_symbols_reason_gone, m_group_date_dropped, m_group_date_later, m_envelope_read_early, m_envelope_arrival_gone, m_receipt_date_dropped, m_groups_reason_gone, m_passed_then_presented, m_contradiction_early, m_not_an_instant_restored, m_enough_then_arrives, restored("claims_at_not_a_list", ["claims_at_ms"], [None]), restored("claims_at_null", ["claims_at_ms"], [None]), restored("records_at_not_a_list", ["records_at_ms"], [None]), restored("groups_at_not_a_list", ["groups_at_ms"], [None]), restored("receipts_at_not_a_list", ["receipts_at_ms"], [[None] * 4]), restored("receipts_at_one_not_a_list", ["receipts_at_ms", 0], [None] * 4), restored("symbols_held_a_string", ["symbols", 0, "held"], [10]), restored("symbols_held_a_fraction", ["symbols", 0, "held"], [10]), restored("symbols_held_not_a_list", ["symbols", 0, "held"], [10]), symbols_restored("symbols_not_a_list"), symbols_restored("symbols_null"), restored("plan_symbols_held_a_string", ["symbols", 0, "held"], [10]), restored("plan_symbols_held_a_fraction", ["symbols", 0, "held"], [10]), restored("plan_symbols_held_not_a_list", ["symbols", 0, "held"], [10]), plan_symbols_restored("plan_symbols_not_a_list"), statement_restored("plan_symbols_statement_not_an_object", False), statement_restored("plan_symbols_bearer_id_not_a_string", False), statement_restored("plan_symbols_group_uppercase", False), statement_restored("plan_symbols_index_a_string", False), statement_restored("symbols_statement_not_an_object", True), statement_restored("symbols_bearer_id_not_a_string", True), statement_restored("symbols_group_uppercase", True), statement_restored("symbols_index_a_string", True), wanted_restored("plan_wanted_not_an_object", ["wanted", 0]), wanted_restored("plan_wanted_group_uppercase", ["wanted", 0]), wanted_restored("plan_wanted_index_a_string", ["wanted", 0]), wanted_restored("plan_wanted_not_a_list", ["wanted"], as_list=True), wanted_restored("wanted_not_an_object", ["plans", 0, "wanted", 0]), wanted_restored("wanted_group_uppercase", ["plans", 0, "wanted", 0]), wanted_restored("wanted_index_a_string", ["plans", 0, "wanted", 0]), wanted_restored("wanted_not_a_list", ["plans", 0, "wanted"], as_list=True), m_wanted_reason_gone, m_wanted_twice_reason] WEAKENINGS = ["subject", "signer", "range", "before", "same_second", "content", "layer", "fallback_position", "derived", "context", "intent_contradiction", "leaf_count", "held_before", "record_contradiction", "line_holders", "least", "least_fallback", "own_statements", "key_order", "deadline_inclusive", "fallback_deadline", "sink_rule", "floor", "no_intent", "loss_bound", "overhead", "instant", "sources_empty", "sources_kind", "sources_uri", "sources_key", "sources_fetch", "sources_twice", "sources_loss", "per_group", "schedule_contradicted", "schedule_last", "undeclared", "bidirectional", "record_at_s", "holder_record", "plan_order", "fallback_value", "fallback_tie", "fallback_order", "plan_fallback_deadline", "late_zero", "late_tie", "rung_layer", "rung_position", "rung_derived", "rung_holders", "rung_order", "plan_rung_first", "plan_rung_order", "run_plans_none", "run_plans_from_ms", "run_plans_order", "run_plans_twice", "run_arrivals_last", "run_arrivals_order", "run_arrivals_twice", "run_same_fetch_ms", "run_arrival_completes", "run_continue", "run_drop_abandon", "run_settled_refetch", "run_held_fetch", "run_continued_plan", "run_by_deadline", "run_rung_order", "run_present_abandon", "run_passed_abandon", "run_late_value", "run_late_zero", "run_end_abandon", "run_ended", "run_claims_undated", "run_claims_strict", "run_claims_length", "run_claims_last", "run_claims_twice", "run_claims_signed", "run_claims_signed_ms", "run_claims_unverified", "run_claims_chain", "run_records_undated", "run_records_strict", "run_records_length", "run_records_last", "run_records_twice", "run_records_signed", "run_records_signed_ms", "run_dated_order", "symbols_uri", "symbols_empty", "symbols_number", "symbols_twice", "symbols_holding", "symbols_ignored", "symbols_instance", "run_symbols_last", "run_symbols_order", "run_symbols_uri", "run_symbols_empty", "run_symbols_number", "run_symbols_twice", "run_symbols_blocks", "run_symbols_fall", "run_symbols_undated", "run_symbols_strict", "run_symbols_latest", "run_groups_undated", "run_groups_strict", "run_groups_length", "run_groups_last", "run_groups_twice", "run_groups_signed", "run_groups_signed_ms", "run_groups_unverified", "run_groups_read", "run_receipts_undated", "run_receipts_strict", "run_receipts_groups", "run_receipts_objects", "run_receipts_last", "run_receipts_read", "run_envelopes_undated", "run_envelopes_strict", "symbols_statement", "symbols_bearer_string", "symbols_group", "symbols_index", "symbols_greatest", "run_symbols_statement", "run_symbols_bearer_string", "run_symbols_group", "run_symbols_index", "run_symbols_greatest", "wanted_list", "wanted_object", "wanted_group", "wanted_index", "wanted_greatest", "wanted_before_sources", "wanted_before_symbols", "run_wanted_before_twice", "run_wanted_malformed_twice", "run_wanted_before_sources", "run_wanted_late", "symbols_list", "symbols_held_list", "symbols_integer", "run_symbols_list", "run_symbols_null", "run_symbols_held_list", "run_symbols_integer", "run_symbols_hold", "run_claims_not_list", "run_claims_null", "run_records_not_list", "run_groups_not_list", "run_receipts_not_list", "run_receipts_member_not_list", "run_envelopes_opt_in", "run_envelopes_no_arrival", "run_order_records", "run_order_groups", "run_passed_keeps_running"] def direct_checks(): """What no vector carries, and what this text states of the texts beside it.""" fails = [] # §7.1: the greatest fetch estimate is admitted; the corpus carries only the one past it. if not sources_valid([{"kind": "origin", "fetch_ms": MAX_INT}], None, None): fails.append("a fetch_ms of 2^53 - 1 is refused") # §4.2 rule 3.2 where the products pass 2^64: three points on one line are refused, and a # part per million off it is not. A float of 64 bits cannot tell them apart. half = (LAST_INSTANT - 1) // 2 for v1, ok in ((500_000, False), (500_001, True)): e = {"leaf_index": 0, "content_hash": "00" * 32, "from_ms": 0, "by_ms": 0, "fallback": None, "decay": [{"at_ms": half, "value_ppm": v1}, {"at_ms": 2 * half, "value_ppm": 0}]} try: CV.timeliness_entry_rules(e, None) got = True except CV.Reject: got = False if got != ok: fails.append(f"a point of {v1} at {half} is {'refused' if ok else 'accepted'}") # §7.5.3 states each action a run decides. decisions = text_of(EXTENSION) decisions = decisions[decisions.index("\n#### 7.5.3 Decisions"):decisions.index("\n#### 7.5.4 ")] for action in ACTIONS: if f"`{action}`" not in decisions: fails.append(f"§7.5.3 does not state {action}") # jwp-receipts §3.1 lists §4.2's codes at step 4 and §4.3's at step 6; delivery-telemetry §5.1 # lists §8.6's switch reason. jwp = text_of(JWP_RECEIPTS) for code, where in reasons_table(): step = {"§4.2": 4, "§4.3": 6}.get(where.split(" ")[0]) if step is not None and f"| {step} | `{code}` |" not in jwp: fails.append(f"jwp-receipts §3.1 does not list {code} at step {step}") if "- `not-held-by-deadline`:" not in text_of(TELEMETRY): fails.append("delivery-telemetry §5.1 does not list not-held-by-deadline") if REFINEMENT_ONLY != {"wai.image.int_refine", "wai.video.int_refine", "wai.audio.int_refine", "wai.splat.int_refine", "wai.field.int_refine", "wai.image.int_cond_refine"}: fails.append(f"SPEC §5's refinement capabilities read as {sorted(REFINEMENT_ONLY)}") return fails def main(argv): global DROPPED if len(argv) != 2: print(__doc__) return 2 doc = json.load(open(os.path.join(argv[1], "vectors.json"), encoding="utf-8")) extra = sorted(set(os.listdir(argv[1])) - {"vectors.json", "README.md"}) fails = check(doc) if extra: fails.append(f"files the corpus does not hold: {extra}") fails += direct_checks() print(f"{sum(len(doc[k]) for k in KINDS)} cases ({', '.join(f'{len(doc[k])} {k}' for k in KINDS)}): " f"{len(fails)} failure(s)") caught = 0 for m in mutations(doc): d = copy.deepcopy(doc) m(d) if check(d, quiet=True): caught += 1 else: fails.append(f"mutation {m.__name__} was not caught") print(f"{caught} of {len(mutations(doc))} mutated copies fail") weak = 0 for w in WEAKENINGS: DROPPED = w if check(doc, quiet=True): weak += 1 else: fails.append(f"a verifier without its {w} check passes the corpus") DROPPED = None print(f"{weak} of {len(WEAKENINGS)} weakened verifiers are caught") for f in fails: print("FAIL", f) return 1 if fails else 0 if __name__ == "__main__": sys.exit(main(sys.argv))