1003 lines
42 KiB
Python
1003 lines
42 KiB
Python
#!/usr/bin/env python3
|
|
"""tracecmp.py -- validate shim trace/compare logs (TRACE_FORMAT.md) and report
|
|
divergences; or replay a golden trace against an implementation's output.
|
|
|
|
usage:
|
|
tracecmp.py LOG.jsonl [LOG2.jsonl ...] report per-hook divergences
|
|
tracecmp.py --replay IMPL.jsonl GOLDEN.jsonl [...] diff GOLDEN (ret/side) vs IMPL
|
|
records matched by call_id
|
|
|
|
flags:
|
|
--hook NAME only this hook (repeatable)
|
|
--tolerance SPEC float policy; SPEC = [HOOK=][abs:|rel:|ulp:]NUMBER
|
|
e.g. 1e-6 | rel:1e-5 | Mars::ParseBlock=ulp:2 (repeatable;
|
|
a bare NUMBER without HOOK sets the default for every hook)
|
|
--unordered SPEC treat a list path as a set; SPEC = [HOOK=]PATH (repeatable)
|
|
--ptr exact|ignore pointer policy for every hook (default: ignore)
|
|
--numeric inside `json` values, compare int vs float as numbers
|
|
--first N diffs shown per hook (default 5)
|
|
--json-out PATH write the full report as JSON (convention:
|
|
verify/results/compare/<run>.json)
|
|
--strict-coverage also fail on any undeclared write or unstated hook
|
|
--skip-invalid drop unparsable/invalid records with a warning instead of
|
|
failing the run
|
|
|
|
Every report ends with a `coverage` section stating, per hook, which regions were
|
|
actually compared, which coarse spans were guarded, what the guards saw the original
|
|
write outside those regions, and what the hook's descriptor admits it does not model.
|
|
A clean compare therefore always says what it did not check.
|
|
|
|
exit: 0 = no divergences; 1 = divergences; 2 = invalid input (or usage).
|
|
A hook whose descriptor claims complete coverage while a guard region caught an
|
|
undeclared write is counted as a divergence (it is one: the model is wrong).
|
|
|
|
Stdlib only.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import os
|
|
import re
|
|
import struct
|
|
import sys
|
|
from dataclasses import dataclass, field, replace as dc_replace
|
|
from typing import Any
|
|
|
|
FORMAT_VERSION = 1
|
|
MODES = ("trace", "compare", "replace")
|
|
INT_TYPES = {"i8", "i16", "i32", "i64", "u8", "u16", "u32", "u64"}
|
|
FLOAT_TYPES = {"f32", "f64"}
|
|
TV_TYPES = INT_TYPES | FLOAT_TYPES | {"bool", "str", "wstr", "ptr", "enum", "null",
|
|
"bytes", "list", "set", "struct", "json"}
|
|
RECORD_KEYS = {"ts", "hook", "mode", "call_id", "thread", "depth", "args", "ret", "side",
|
|
"ours", "diverged", "diff", "coverage", "err", "note"}
|
|
COVERAGE_STATES = ("complete", "partial", "unstated")
|
|
INT_RANGE = {"i8": (-2**7, 2**7 - 1), "i16": (-2**15, 2**15 - 1), "i32": (-2**31, 2**31 - 1),
|
|
"i64": (-2**63, 2**63 - 1), "u8": (0, 2**8 - 1), "u16": (0, 2**16 - 1),
|
|
"u32": (0, 2**32 - 1), "u64": (0, 2**64 - 1)}
|
|
_HEX_RE = re.compile(r"^[0-9a-f]*$")
|
|
_SHA_RE = re.compile(r"^[0-9a-f]{64}$")
|
|
_PTR_RE = re.compile(r"^0x[0-9a-fA-F]{1,16}$")
|
|
NONFINITE = {"nan": math.nan, "inf": math.inf, "-inf": -math.inf}
|
|
|
|
|
|
# --- policy ------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class Policy:
|
|
ftol: float = 0.0
|
|
ftol_kind: str = "abs" # abs | rel | ulp
|
|
ptr: str = "ignore" # ignore | exact
|
|
numeric: bool = False # json: int vs float compare as numbers
|
|
unordered: set = field(default_factory=set)
|
|
|
|
def merged(self, d: dict) -> "Policy":
|
|
p = dc_replace(self, unordered=set(self.unordered))
|
|
if "ftol" in d:
|
|
p.ftol = float(d["ftol"])
|
|
if "ftol_kind" in d:
|
|
p.ftol_kind = d["ftol_kind"]
|
|
if "ptr" in d:
|
|
p.ptr = d["ptr"]
|
|
if "numeric" in d:
|
|
p.numeric = bool(d["numeric"])
|
|
if "unordered" in d:
|
|
p.unordered |= set(d["unordered"])
|
|
return p
|
|
|
|
|
|
class PolicyTable:
|
|
"""defaults < log meta per-hook < CLI default < CLI per-hook."""
|
|
|
|
def __init__(self):
|
|
self.default: dict = {}
|
|
self.meta: dict[str, dict] = {}
|
|
self.cli: dict[str, dict] = {}
|
|
|
|
def add_meta(self, hooks: dict) -> None:
|
|
for h, d in (hooks or {}).items():
|
|
self.meta.setdefault(h, {}).update(d)
|
|
|
|
def add_cli(self, hook: str | None, d: dict) -> None:
|
|
if hook is None:
|
|
self.default.update(d)
|
|
else:
|
|
self.cli.setdefault(hook, {}).update(d)
|
|
|
|
def for_hook(self, hook: str) -> Policy:
|
|
p = Policy().merged(self.meta.get(hook, {}))
|
|
p = p.merged(self.default)
|
|
return p.merged(self.cli.get(hook, {}))
|
|
|
|
|
|
def parse_tolerance(spec: str) -> tuple[str | None, dict]:
|
|
hook = None
|
|
if "=" in spec:
|
|
hook, spec = spec.split("=", 1)
|
|
kind = "abs"
|
|
if ":" in spec:
|
|
kind, spec = spec.split(":", 1)
|
|
if kind not in ("abs", "rel", "ulp"):
|
|
raise ValueError(f"bad tolerance kind {kind!r}")
|
|
return hook, {"ftol": float(spec), "ftol_kind": kind}
|
|
|
|
|
|
def parse_unordered(spec: str) -> tuple[str | None, dict]:
|
|
hook = None
|
|
if "=" in spec:
|
|
hook, spec = spec.split("=", 1)
|
|
return hook, {"unordered": [spec]}
|
|
|
|
|
|
# --- validation ----------------------------------------------------------------
|
|
|
|
class Invalid(Exception):
|
|
pass
|
|
|
|
|
|
def _is_int(x) -> bool:
|
|
return isinstance(x, int) and not isinstance(x, bool)
|
|
|
|
|
|
def validate_tv(tv: Any, path: str, errs: list[str]) -> None:
|
|
if not isinstance(tv, dict):
|
|
errs.append(f"{path}: typed value must be an object, got {type(tv).__name__}")
|
|
return
|
|
t = tv.get("t")
|
|
if t not in TV_TYPES:
|
|
errs.append(f"{path}: unknown type {t!r}")
|
|
return
|
|
if "n" in tv and not isinstance(tv["n"], (str, int)):
|
|
errs.append(f"{path}: 'n' must be a string (name) or int (bytes length)")
|
|
if t == "bytes":
|
|
n, sha = tv.get("n"), tv.get("sha256")
|
|
if not _is_int(n) or n < 0:
|
|
errs.append(f"{path}: bytes.n must be int >= 0")
|
|
if not isinstance(sha, str) or not _SHA_RE.match(sha):
|
|
errs.append(f"{path}: bytes.sha256 must be 64 lowercase hex chars")
|
|
if "hex" in tv:
|
|
h = tv["hex"]
|
|
if not isinstance(h, str) or not _HEX_RE.match(h) or (_is_int(n) and len(h) != 2 * n):
|
|
errs.append(f"{path}: bytes.hex must be 2*n lowercase hex chars")
|
|
if "head" in tv and (not isinstance(tv["head"], str) or not _HEX_RE.match(tv["head"])):
|
|
errs.append(f"{path}: bytes.head must be lowercase hex")
|
|
return
|
|
if "v" not in tv:
|
|
errs.append(f"{path}: missing 'v'")
|
|
return
|
|
v = tv["v"]
|
|
if t == "bool":
|
|
if not isinstance(v, bool):
|
|
errs.append(f"{path}: bool.v must be true/false")
|
|
elif t in INT_TYPES:
|
|
iv = None
|
|
if _is_int(v):
|
|
iv = v
|
|
elif isinstance(v, str) and re.match(r"^-?\d+$", v):
|
|
iv = int(v)
|
|
else:
|
|
errs.append(f"{path}: {t}.v must be an integer (or decimal string)")
|
|
if iv is not None:
|
|
lo, hi = INT_RANGE[t]
|
|
if not lo <= iv <= hi:
|
|
errs.append(f"{path}: {t}.v={iv} out of range")
|
|
elif t in FLOAT_TYPES:
|
|
if isinstance(v, bool) or not (isinstance(v, (int, float)) or v in NONFINITE):
|
|
errs.append(f"{path}: {t}.v must be a number or 'nan'/'inf'/'-inf'")
|
|
elif t in ("str", "wstr"):
|
|
if not isinstance(v, str):
|
|
errs.append(f"{path}: {t}.v must be a string")
|
|
elif t == "str" and any(ord(c) > 0xFF for c in v):
|
|
errs.append(f"{path}: str.v carries a code point > U+00FF (bytes must be \\u00XX)")
|
|
elif t == "ptr":
|
|
if not isinstance(v, str) or not _PTR_RE.match(v):
|
|
errs.append(f"{path}: ptr.v must be a hex string like 0x00a36fd0")
|
|
elif t == "enum":
|
|
if not _is_int(v):
|
|
errs.append(f"{path}: enum.v must be an integer")
|
|
elif t == "null":
|
|
if v is not None:
|
|
errs.append(f"{path}: null.v must be null")
|
|
elif t in ("list", "set"):
|
|
if not isinstance(v, list):
|
|
errs.append(f"{path}: {t}.v must be an array")
|
|
else:
|
|
for i, e in enumerate(v):
|
|
validate_tv(e, f"{path}.v[{i}]", errs)
|
|
elif t == "struct":
|
|
if not isinstance(v, dict):
|
|
errs.append(f"{path}: struct.v must be an object")
|
|
else:
|
|
for k, e in v.items():
|
|
validate_tv(e, f"{path}.v.{k}", errs)
|
|
elif t == "json":
|
|
pass # any JSON
|
|
|
|
|
|
def validate_side(side: Any, path: str, errs: list[str], need_before: bool = False) -> None:
|
|
if not isinstance(side, dict):
|
|
errs.append(f"{path}: must be an object")
|
|
return
|
|
for name, ent in side.items():
|
|
if not isinstance(ent, dict) or "after" not in ent:
|
|
errs.append(f"{path}.{name}: must be {{\"before\"?, \"after\"}}")
|
|
continue
|
|
if ent.get("before") is not None:
|
|
validate_tv(ent["before"], f"{path}.{name}.before", errs)
|
|
validate_tv(ent["after"], f"{path}.{name}.after", errs)
|
|
|
|
|
|
def validate_coverage(cov: Any, errs: list[str]) -> None:
|
|
"""Record-level `coverage`: {guards: [name], undeclared: [{region, off, len}], n: int}.
|
|
|
|
Present iff the call declared at least one Guard region, so an empty `undeclared` means
|
|
"watched and nothing moved" -- which is different from, and much stronger than, absent.
|
|
"""
|
|
if not isinstance(cov, dict):
|
|
errs.append("coverage must be an object")
|
|
return
|
|
guards = cov.get("guards")
|
|
if not isinstance(guards, list) or not all(isinstance(g, str) for g in guards):
|
|
errs.append("coverage.guards must be an array of strings")
|
|
und = cov.get("undeclared")
|
|
if not isinstance(und, list):
|
|
errs.append("coverage.undeclared must be an array")
|
|
else:
|
|
for i, w in enumerate(und):
|
|
if (not isinstance(w, dict) or not isinstance(w.get("region"), str)
|
|
or not _is_int(w.get("off")) or not _is_int(w.get("len"))):
|
|
errs.append(f"coverage.undeclared[{i}]: must be {{region, off, len}}")
|
|
if "n" in cov and (not _is_int(cov["n"]) or cov["n"] < 0):
|
|
errs.append("coverage.n must be int >= 0")
|
|
if isinstance(und, list) and _is_int(cov.get("n")) and cov["n"] < len(und):
|
|
errs.append("coverage.n is smaller than the spans it reports")
|
|
|
|
|
|
def validate_meta_coverage(hook: str, cov: Any, warns: list[str]) -> None:
|
|
"""meta.hooks[H].coverage: the descriptor's own admission of what it does not check."""
|
|
if not isinstance(cov, dict):
|
|
warns.append(f"meta.hooks.{hook}.coverage must be an object")
|
|
return
|
|
if cov.get("state") not in COVERAGE_STATES:
|
|
warns.append(f"meta.hooks.{hook}.coverage.state must be one of {COVERAGE_STATES}")
|
|
for i, n in enumerate(cov.get("unmodelled") or []):
|
|
if not isinstance(n, dict) or not isinstance(n.get("what"), str) or not n.get("what"):
|
|
warns.append(f"meta.hooks.{hook}.coverage.unmodelled[{i}]: needs a non-empty 'what'")
|
|
|
|
|
|
def validate_record(rec: Any) -> tuple[list[str], list[str]]:
|
|
"""-> (errors, warnings). Errors make the record invalid."""
|
|
errs: list[str] = []
|
|
warns: list[str] = []
|
|
if not isinstance(rec, dict):
|
|
return ["record is not an object"], warns
|
|
for k in ("ts", "hook", "mode", "call_id", "thread", "args", "ret", "side"):
|
|
if k not in rec:
|
|
errs.append(f"missing required field {k!r}")
|
|
if errs:
|
|
return errs, warns
|
|
if not _is_int(rec["ts"]) and not isinstance(rec["ts"], float):
|
|
errs.append("ts must be a number")
|
|
if not isinstance(rec["hook"], str) or not rec["hook"]:
|
|
errs.append("hook must be a non-empty string")
|
|
if rec["mode"] not in MODES:
|
|
errs.append(f"mode must be one of {MODES}")
|
|
if not _is_int(rec["call_id"]) or rec["call_id"] < 0:
|
|
errs.append("call_id must be int >= 0")
|
|
if not _is_int(rec["thread"]):
|
|
errs.append("thread must be int")
|
|
if "depth" in rec and (not _is_int(rec["depth"]) or rec["depth"] < 0):
|
|
errs.append("depth must be int >= 0")
|
|
if not isinstance(rec["args"], list):
|
|
errs.append("args must be an array")
|
|
else:
|
|
for i, a in enumerate(rec["args"]):
|
|
validate_tv(a, f"args[{i}]", errs)
|
|
if rec["ret"] is not None:
|
|
validate_tv(rec["ret"], "ret", errs)
|
|
validate_side(rec["side"], "side", errs)
|
|
if "coverage" in rec:
|
|
validate_coverage(rec["coverage"], errs)
|
|
if "err" in rec and not isinstance(rec["err"], str):
|
|
errs.append("err must be a string")
|
|
if "note" in rec and not isinstance(rec["note"], str):
|
|
errs.append("note must be a string")
|
|
mode = rec["mode"]
|
|
if mode == "compare":
|
|
ours = rec.get("ours")
|
|
if not isinstance(ours, dict) or "ret" not in ours or "side" not in ours:
|
|
if "err" not in rec:
|
|
errs.append("compare record needs ours={ret, side} (or err)")
|
|
else:
|
|
if ours["ret"] is not None:
|
|
validate_tv(ours["ret"], "ours.ret", errs)
|
|
validate_side(ours["side"], "ours.side", errs)
|
|
if "diverged" not in rec:
|
|
warns.append("compare record without 'diverged'")
|
|
elif not isinstance(rec["diverged"], bool):
|
|
errs.append("diverged must be a bool")
|
|
if "diff" in rec and not isinstance(rec["diff"], list):
|
|
errs.append("diff must be an array")
|
|
else:
|
|
if "ours" in rec:
|
|
warns.append(f"{mode} record carries 'ours' (ignored)")
|
|
for k in rec:
|
|
if k not in RECORD_KEYS:
|
|
warns.append(f"unknown field {k!r}")
|
|
return errs, warns
|
|
|
|
|
|
# --- comparison ------------------------------------------------------------------
|
|
|
|
def _diff(path: str, why: str, a: Any, b: Any, **extra) -> dict:
|
|
d = {"path": path, "why": why, "orig": a, "ours": b}
|
|
d.update(extra)
|
|
return d
|
|
|
|
|
|
def _num(v, width: str = "f64") -> float:
|
|
"""tv float payload -> Python float. f32 values are rounded to float32:
|
|
the emitter's %.9g round-trips the float32, not the exact double."""
|
|
if isinstance(v, str):
|
|
return NONFINITE[v]
|
|
x = float(v)
|
|
if width == "f32" and math.isfinite(x):
|
|
x = struct.unpack("<f", struct.pack("<f", x))[0]
|
|
return x
|
|
|
|
|
|
def _ulp_diff(a: float, b: float, width: str) -> int:
|
|
fmt, ifmt = ("f", "i") if width == "f32" else ("d", "q")
|
|
def to_int(x):
|
|
i = struct.unpack("<" + ifmt, struct.pack("<" + fmt, x))[0]
|
|
return i if i >= 0 else -(i & (2**(31 if fmt == "f" else 63) - 1))
|
|
return abs(to_int(a) - to_int(b))
|
|
|
|
|
|
def floats_equal(a: float, b: float, pol: Policy, width: str = "f64") -> bool:
|
|
if math.isnan(a) or math.isnan(b):
|
|
return math.isnan(a) and math.isnan(b)
|
|
if math.isinf(a) or math.isinf(b):
|
|
return a == b
|
|
if a == b:
|
|
return True
|
|
if pol.ftol <= 0:
|
|
return False
|
|
if pol.ftol_kind == "abs":
|
|
return abs(a - b) <= pol.ftol
|
|
if pol.ftol_kind == "rel":
|
|
return abs(a - b) <= pol.ftol * max(abs(a), abs(b))
|
|
if pol.ftol_kind == "ulp":
|
|
return _ulp_diff(a, b, width) <= pol.ftol
|
|
raise ValueError(pol.ftol_kind)
|
|
|
|
|
|
def _canon(x) -> str:
|
|
return json.dumps(x, sort_keys=True, separators=(",", ":"))
|
|
|
|
|
|
def compare_json(a: Any, b: Any, pol: Policy, path: str, out: list) -> None:
|
|
if isinstance(a, bool) or isinstance(b, bool) or a is None or b is None or \
|
|
isinstance(a, str) or isinstance(b, str):
|
|
if type(a) is not type(b) or a != b:
|
|
out.append(_diff(path, "exact", a, b))
|
|
return
|
|
if isinstance(a, (int, float)) and isinstance(b, (int, float)):
|
|
both_float = isinstance(a, float) and isinstance(b, float)
|
|
if type(a) is not type(b) and not pol.numeric:
|
|
out.append(_diff(path, "type", a, b))
|
|
return
|
|
if both_float or pol.numeric:
|
|
if not floats_equal(float(a), float(b), pol):
|
|
out.append(_diff(path, "ftol" if pol.ftol > 0 else "exact", a, b))
|
|
elif a != b:
|
|
out.append(_diff(path, "exact", a, b))
|
|
return
|
|
if isinstance(a, list) and isinstance(b, list):
|
|
if path in pol.unordered:
|
|
a = sorted(a, key=_canon)
|
|
b = sorted(b, key=_canon)
|
|
if len(a) != len(b):
|
|
out.append(_diff(path, "len", len(a), len(b)))
|
|
return
|
|
for i, (x, y) in enumerate(zip(a, b)):
|
|
compare_json(x, y, pol, f"{path}[{i}]", out)
|
|
return
|
|
if isinstance(a, dict) and isinstance(b, dict):
|
|
ka, kb = set(a), set(b)
|
|
for k in sorted(ka - kb):
|
|
out.append(_diff(f"{path}.{k}", "missing", a[k], None))
|
|
for k in sorted(kb - ka):
|
|
out.append(_diff(f"{path}.{k}", "extra", None, b[k]))
|
|
for k in sorted(ka & kb):
|
|
compare_json(a[k], b[k], pol, f"{path}.{k}", out)
|
|
return
|
|
out.append(_diff(path, "type", a, b))
|
|
|
|
|
|
def compare_tv(a: Any, b: Any, pol: Policy, path: str, out: list) -> None:
|
|
"""Append diff entries for typed values a (orig) vs b (ours)."""
|
|
if a is None or b is None:
|
|
if a is not b:
|
|
out.append(_diff(path, "missing" if b is None else "extra", a, b))
|
|
return
|
|
ta, tb = a.get("t"), b.get("t")
|
|
if ta != tb:
|
|
out.append(_diff(path, "type", a, b))
|
|
return
|
|
t = ta
|
|
if t == "ptr":
|
|
va, vb = a["v"], b["v"]
|
|
za, zb = int(va, 16) == 0, int(vb, 16) == 0
|
|
if za != zb:
|
|
out.append(_diff(path, "exact", a, b))
|
|
elif pol.ptr == "exact" and int(va, 16) != int(vb, 16):
|
|
out.append(_diff(path, "exact", a, b))
|
|
return
|
|
if t == "null":
|
|
return
|
|
if t == "bool" or t == "enum" or t in ("str", "wstr"):
|
|
if a["v"] != b["v"]:
|
|
out.append(_diff(path, "exact", a, b))
|
|
return
|
|
if t in INT_TYPES:
|
|
if int(a["v"]) != int(b["v"]):
|
|
out.append(_diff(path, "exact", a, b))
|
|
return
|
|
if t in FLOAT_TYPES:
|
|
if not floats_equal(_num(a["v"], t), _num(b["v"], t), pol, t):
|
|
out.append(_diff(path, "ftol" if pol.ftol > 0 else "exact", a, b))
|
|
return
|
|
if t == "bytes":
|
|
if a.get("n") != b.get("n"):
|
|
out.append(_diff(path, "len", a, b))
|
|
return
|
|
if a.get("sha256") != b.get("sha256"):
|
|
extra = {}
|
|
if "hex" in a and "hex" in b:
|
|
ha, hb = a["hex"], b["hex"]
|
|
off = next((i for i in range(0, min(len(ha), len(hb)), 2) if ha[i:i+2] != hb[i:i+2]), None)
|
|
if off is not None:
|
|
extra["first_diff_offset"] = off // 2
|
|
out.append(_diff(path, "hash", a, b, **extra))
|
|
elif "hex" in a and "hex" in b and a["hex"] != b["hex"]:
|
|
out.append(_diff(path, "hash", a, b, note="same sha256, different hex: corrupt log"))
|
|
return
|
|
if t in ("list", "set"):
|
|
va, vb = a["v"], b["v"]
|
|
if t == "set" or path in pol.unordered:
|
|
va = sorted(va, key=_canon)
|
|
vb = sorted(vb, key=_canon)
|
|
if len(va) != len(vb):
|
|
out.append(_diff(path, "len", len(va), len(vb)))
|
|
return
|
|
for i, (x, y) in enumerate(zip(va, vb)):
|
|
compare_tv(x, y, pol, f"{path}.v[{i}]", out)
|
|
return
|
|
if t == "struct":
|
|
va, vb = a["v"], b["v"]
|
|
for k in sorted(set(va) - set(vb)):
|
|
out.append(_diff(f"{path}.v.{k}", "missing", va[k], None))
|
|
for k in sorted(set(vb) - set(va)):
|
|
out.append(_diff(f"{path}.v.{k}", "extra", None, vb[k]))
|
|
for k in sorted(set(va) & set(vb)):
|
|
compare_tv(va[k], vb[k], pol, f"{path}.v.{k}", out)
|
|
return
|
|
if t == "json":
|
|
compare_json(a["v"], b["v"], pol, path, out)
|
|
return
|
|
raise AssertionError(t)
|
|
|
|
|
|
def _side_after(side: dict, name: str) -> Any:
|
|
ent = side[name]
|
|
if isinstance(ent, dict) and ent.get("t") is None and "after" in ent:
|
|
return ent["after"]
|
|
return ent # replay input may give the tv directly
|
|
|
|
|
|
def compare_outputs(orig_ret, orig_side: dict, ours_ret, ours_side: dict, pol: Policy) -> list[dict]:
|
|
diffs: list[dict] = []
|
|
compare_tv(orig_ret, ours_ret, pol, "ret", diffs)
|
|
names_a, names_b = set(orig_side), set(ours_side)
|
|
for n in sorted(names_a - names_b):
|
|
diffs.append(_diff(f"side.{n}.after", "missing", _side_after(orig_side, n), None))
|
|
for n in sorted(names_b - names_a):
|
|
diffs.append(_diff(f"side.{n}.after", "extra", None, _side_after(ours_side, n)))
|
|
for n in sorted(names_a & names_b):
|
|
compare_tv(_side_after(orig_side, n), _side_after(ours_side, n), pol, f"side.{n}.after", diffs)
|
|
return diffs
|
|
|
|
|
|
def compare_record(rec: dict, pol: Policy) -> tuple[list[dict], list[str]]:
|
|
"""Recompute a compare record's diff. -> (diffs, warnings)"""
|
|
warns: list[str] = []
|
|
if "err" in rec:
|
|
return [_diff("call", "err", None, rec["err"])], warns
|
|
ours = rec["ours"]
|
|
diffs = compare_outputs(rec["ret"], rec["side"], ours["ret"], ours["side"], pol)
|
|
# snapshot sanity: 'before' on both sides should be identical
|
|
for n in set(rec["side"]) & set(ours["side"]):
|
|
ba, bb = rec["side"][n].get("before"), ours["side"][n].get("before")
|
|
if ba is not None and bb is not None:
|
|
tmp: list = []
|
|
compare_tv(ba, bb, pol, f"side.{n}.before", tmp)
|
|
if tmp:
|
|
warns.append(f"call {rec['call_id']}: side.{n}.before differs between sides (snapshot bug?)")
|
|
if "diverged" in rec and rec["diverged"] != bool(diffs):
|
|
warns.append(f"call {rec['call_id']}: shim said diverged={rec['diverged']}, harness found {len(diffs)} diff(s)")
|
|
return diffs, warns
|
|
|
|
|
|
# --- loading -----------------------------------------------------------------------
|
|
|
|
@dataclass
|
|
class Log:
|
|
path: str
|
|
meta: dict = field(default_factory=dict)
|
|
records: list = field(default_factory=list)
|
|
invalid: list = field(default_factory=list) # (lineno, message)
|
|
warnings: list = field(default_factory=list) # (lineno, message)
|
|
|
|
|
|
def load_log(path: str, policies: PolicyTable | None = None) -> Log:
|
|
log = Log(path)
|
|
seen: dict[int, int] = {}
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for lineno, line in enumerate(f, 1):
|
|
line = line.rstrip("\n")
|
|
if line.endswith("\r"):
|
|
log.warnings.append((lineno, "CRLF line ending"))
|
|
line = line[:-1]
|
|
if not line.strip():
|
|
log.invalid.append((lineno, "blank line"))
|
|
continue
|
|
try:
|
|
rec = json.loads(line)
|
|
except ValueError as e:
|
|
log.invalid.append((lineno, f"not JSON: {e}"))
|
|
continue
|
|
if isinstance(rec, dict) and "meta" in rec and len(rec) == 1:
|
|
m = rec["meta"]
|
|
if not isinstance(m, dict):
|
|
log.invalid.append((lineno, "meta must be an object"))
|
|
continue
|
|
if m.get("format") != FORMAT_VERSION:
|
|
log.invalid.append((lineno, f"unsupported format {m.get('format')!r} (want {FORMAT_VERSION})"))
|
|
continue
|
|
if lineno != 1:
|
|
log.warnings.append((lineno, "meta record not on line 1"))
|
|
log.meta = m
|
|
if policies is not None:
|
|
policies.add_meta(m.get("hooks", {}))
|
|
continue
|
|
errs, warns = validate_record(rec)
|
|
for w in warns:
|
|
log.warnings.append((lineno, w))
|
|
if errs:
|
|
log.invalid.append((lineno, "; ".join(errs)))
|
|
continue
|
|
cid = rec["call_id"]
|
|
if cid in seen:
|
|
log.invalid.append((lineno, f"duplicate call_id {cid} (first at line {seen[cid]})"))
|
|
continue
|
|
seen[cid] = lineno
|
|
rec["_line"] = lineno
|
|
log.records.append(rec)
|
|
return log
|
|
|
|
|
|
def load_impl(path: str) -> tuple[dict[int, dict], list[tuple[int, str]]]:
|
|
"""Implementation output for --replay: records keyed by call_id."""
|
|
out: dict[int, dict] = {}
|
|
invalid: list[tuple[int, str]] = []
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for lineno, line in enumerate(f, 1):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
rec = json.loads(line)
|
|
except ValueError as e:
|
|
invalid.append((lineno, f"not JSON: {e}"))
|
|
continue
|
|
if isinstance(rec, dict) and "meta" in rec and len(rec) == 1:
|
|
continue
|
|
if not isinstance(rec, dict) or not _is_int(rec.get("call_id")):
|
|
invalid.append((lineno, "impl record needs an integer call_id"))
|
|
continue
|
|
errs: list[str] = []
|
|
if rec.get("ret") is not None:
|
|
validate_tv(rec["ret"], "ret", errs)
|
|
side = rec.get("side", {})
|
|
if not isinstance(side, dict):
|
|
errs.append("side must be an object")
|
|
else:
|
|
for n in side:
|
|
validate_tv(_side_after(side, n), f"side.{n}", errs)
|
|
if errs:
|
|
invalid.append((lineno, "; ".join(errs)))
|
|
continue
|
|
if rec["call_id"] in out:
|
|
invalid.append((lineno, f"duplicate call_id {rec['call_id']}"))
|
|
continue
|
|
rec.setdefault("side", {})
|
|
rec.setdefault("ret", None)
|
|
rec["_line"] = lineno
|
|
out[rec["call_id"]] = rec
|
|
return out, invalid
|
|
|
|
|
|
# --- report ------------------------------------------------------------------------
|
|
|
|
MAX_SPANS_REPORTED = 12
|
|
|
|
|
|
def new_hook_stats() -> dict:
|
|
return {"calls": 0, "modes": {}, "compared": 0, "diverged": 0, "errors": 0,
|
|
"diverged_call_ids": [], "diffs": [],
|
|
# --- coverage: what this run actually checked, and what it admits it did not ---
|
|
"coverage": {
|
|
"state": "unstated", # from meta.hooks[H].coverage (absent -> unstated)
|
|
"why": "",
|
|
"unmodelled": [], # [{what, risk, why, mitigation}]
|
|
"checked_regions": [], # side names actually diffed (union over records)
|
|
"guards": [], # guard names the shim watched
|
|
"guarded_calls": 0, # calls that carried a coverage block
|
|
"undeclared_calls": 0, # calls where a guard saw an undeclared write
|
|
"undeclared_writes": 0, # total spans, honest even when the list truncates
|
|
"spans": {}, # mode -> ["region+off:len", ...] (capped)
|
|
}}
|
|
|
|
|
|
def _hook_coverage_from_meta(logs: list) -> dict:
|
|
"""meta.hooks[H].coverage, merged across the input logs (last log wins on conflict)."""
|
|
out: dict[str, dict] = {}
|
|
for log in logs:
|
|
for h, d in (log.meta.get("hooks") or {}).items():
|
|
if isinstance(d, dict) and isinstance(d.get("coverage"), dict):
|
|
out[h] = d["coverage"]
|
|
return out
|
|
|
|
|
|
def _account_coverage(st: dict, rec: dict) -> None:
|
|
"""Fold one record's declared regions and guard findings into the hook's coverage stats."""
|
|
cov = st["coverage"]
|
|
for name in rec.get("side") or {}:
|
|
if name not in cov["checked_regions"]:
|
|
cov["checked_regions"].append(name)
|
|
block = rec.get("coverage")
|
|
if not isinstance(block, dict):
|
|
return
|
|
cov["guarded_calls"] += 1
|
|
for g in block.get("guards") or []:
|
|
if g not in cov["guards"]:
|
|
cov["guards"].append(g)
|
|
spans = block.get("undeclared") or []
|
|
total = block.get("n", len(spans))
|
|
if total:
|
|
cov["undeclared_calls"] += 1
|
|
cov["undeclared_writes"] += total
|
|
bucket = cov["spans"].setdefault(rec["mode"], [])
|
|
for w in spans:
|
|
key = "%s+0x%x:%d" % (w["region"], w["off"], w["len"])
|
|
if key not in bucket and len(bucket) < MAX_SPANS_REPORTED:
|
|
bucket.append(key)
|
|
|
|
|
|
def run_report(logs: list[Log], policies: PolicyTable, hooks: set[str] | None, first: int) -> dict:
|
|
per_hook: dict[str, dict] = {}
|
|
warnings: list[str] = []
|
|
for log in logs:
|
|
for ln, w in log.warnings:
|
|
warnings.append(f"{os.path.basename(log.path)}:{ln}: {w}")
|
|
for rec in log.records:
|
|
h = rec["hook"]
|
|
if hooks and h not in hooks:
|
|
continue
|
|
st = per_hook.setdefault(h, new_hook_stats())
|
|
st["calls"] += 1
|
|
st["modes"][rec["mode"]] = st["modes"].get(rec["mode"], 0) + 1
|
|
_account_coverage(st, rec)
|
|
if rec["mode"] != "compare":
|
|
if "err" in rec:
|
|
st["errors"] += 1
|
|
continue
|
|
st["compared"] += 1
|
|
diffs, w = compare_record(rec, policies.for_hook(h))
|
|
warnings.extend(f"{os.path.basename(log.path)}:{rec['_line']}: {x}" for x in w)
|
|
if diffs:
|
|
st["diverged"] += 1
|
|
if "err" in rec:
|
|
st["errors"] += 1
|
|
st["diverged_call_ids"].append(rec["call_id"])
|
|
if len(st["diffs"]) < first:
|
|
st["diffs"].append({"call_id": rec["call_id"], "line": rec["_line"],
|
|
"file": os.path.basename(log.path), "diff": diffs})
|
|
return _finish(per_hook, logs, warnings, "report")
|
|
|
|
|
|
def run_replay(golden: list[Log], impl: dict[int, dict], policies: PolicyTable,
|
|
hooks: set[str] | None, first: int) -> dict:
|
|
per_hook: dict[str, dict] = {}
|
|
warnings: list[str] = []
|
|
seen: set[int] = set()
|
|
for log in golden:
|
|
for ln, w in log.warnings:
|
|
warnings.append(f"{os.path.basename(log.path)}:{ln}: {w}")
|
|
for rec in log.records:
|
|
h = rec["hook"]
|
|
if hooks and h not in hooks:
|
|
continue
|
|
st = per_hook.setdefault(h, new_hook_stats())
|
|
st["calls"] += 1
|
|
st["modes"][rec["mode"]] = st["modes"].get(rec["mode"], 0) + 1
|
|
_account_coverage(st, rec)
|
|
st["compared"] += 1
|
|
cid = rec["call_id"]
|
|
seen.add(cid)
|
|
ours = impl.get(cid)
|
|
if ours is None:
|
|
diffs = [_diff("call", "missing", None, None, note=f"call_id {cid} absent from impl")]
|
|
else:
|
|
if "hook" in ours and ours["hook"] != h:
|
|
warnings.append(f"impl call_id {cid}: hook {ours['hook']!r} != golden {h!r}")
|
|
diffs = compare_outputs(rec["ret"], rec["side"], ours["ret"], ours["side"], policies.for_hook(h))
|
|
if diffs:
|
|
st["diverged"] += 1
|
|
st["diverged_call_ids"].append(cid)
|
|
if len(st["diffs"]) < first:
|
|
st["diffs"].append({"call_id": cid, "line": rec["_line"],
|
|
"file": os.path.basename(log.path), "diff": diffs})
|
|
extra = sorted(set(impl) - seen)
|
|
if extra:
|
|
warnings.append(f"impl has {len(extra)} call_id(s) not in golden: {extra[:10]}{'…' if len(extra) > 10 else ''}")
|
|
return _finish(per_hook, golden, warnings, "replay")
|
|
|
|
|
|
def _finish(per_hook: dict, logs: list[Log], warnings: list[str], kind: str) -> dict:
|
|
total_div = sum(s["diverged"] for s in per_hook.values())
|
|
invalid = [(os.path.basename(l.path), ln, m) for l in logs for ln, m in l.invalid]
|
|
|
|
# Fold the shim's own coverage declaration in, and decide the coverage verdict per hook.
|
|
#
|
|
# unstated the descriptor never said what it does not check -> the compare's
|
|
# completeness is unknown, and a clean run means nothing on its own.
|
|
# contradicted the descriptor claims complete coverage, but a Guard region caught the
|
|
# original writing outside every declared region. The claim is false, and
|
|
# that counts as a divergence (exit 1) -- it is the exact failure B3 hit.
|
|
# partial the honest normal case: admissions on record, guards agreeing with them.
|
|
# complete claims completeness, and no guard contradicted it.
|
|
meta_cov = _hook_coverage_from_meta(logs)
|
|
unstated, contradicted = [], []
|
|
for h, s in per_hook.items():
|
|
cov = s["coverage"]
|
|
m = meta_cov.get(h)
|
|
if isinstance(m, dict):
|
|
cov["state"] = m.get("state", "unstated")
|
|
cov["why"] = m.get("why", "")
|
|
cov["unmodelled"] = list(m.get("unmodelled") or [])
|
|
for k in ("checked_regions", "guards"):
|
|
cov[k] = sorted(cov[k])
|
|
if cov["state"] == "unstated":
|
|
cov["verdict"] = "unstated"
|
|
unstated.append(h)
|
|
elif cov["state"] == "complete" and cov["undeclared_writes"]:
|
|
cov["verdict"] = "contradicted"
|
|
contradicted.append(h)
|
|
else:
|
|
cov["verdict"] = cov["state"]
|
|
if cov["verdict"] == "unstated":
|
|
warnings.append(f"{h}: no coverage statement in meta -- a clean compare for this hook "
|
|
f"does not say what it did not check")
|
|
if cov["verdict"] == "contradicted":
|
|
warnings.append(f"{h}: claims complete coverage but a guard saw "
|
|
f"{cov['undeclared_writes']} undeclared write(s)")
|
|
|
|
return {
|
|
"kind": kind,
|
|
"format": FORMAT_VERSION,
|
|
"inputs": [l.path for l in logs],
|
|
"meta": [l.meta for l in logs if l.meta],
|
|
"hooks": dict(sorted(per_hook.items())),
|
|
"totals": {"calls": sum(s["calls"] for s in per_hook.values()),
|
|
"compared": sum(s["compared"] for s in per_hook.values()),
|
|
"diverged": total_div,
|
|
"invalid_records": len(invalid),
|
|
"guarded_calls": sum(s["coverage"]["guarded_calls"] for s in per_hook.values()),
|
|
"undeclared_calls": sum(s["coverage"]["undeclared_calls"] for s in per_hook.values()),
|
|
"undeclared_writes": sum(s["coverage"]["undeclared_writes"] for s in per_hook.values()),
|
|
"coverage_unstated": len(unstated),
|
|
"coverage_contradicted": len(contradicted)},
|
|
"coverage_unstated": sorted(unstated),
|
|
"coverage_contradicted": sorted(contradicted),
|
|
"invalid": [{"file": f, "line": ln, "error": m} for f, ln, m in invalid],
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
def _short(x: Any, n: int = 60) -> str:
|
|
s = _canon(x)
|
|
return s if len(s) <= n else s[:n - 1] + "…"
|
|
|
|
|
|
def print_coverage(rep: dict, out=None) -> None:
|
|
"""What this run checked, and what it says it did not.
|
|
|
|
Printed on every run, clean or not. A compare that reports zero divergences bounds only the
|
|
regions it declared; without this section that fact is invisible, which is exactly how B3's
|
|
unposted event survived a clean pass.
|
|
"""
|
|
out = out or sys.stdout
|
|
P = lambda s="": print(s, file=out) # noqa: E731
|
|
if not rep["hooks"]:
|
|
return
|
|
P()
|
|
P("### coverage")
|
|
P()
|
|
P("| hook | verdict | compared regions | guards | undeclared writes | unmodelled |")
|
|
P("|---|---|---|---|---|---|")
|
|
for h, s in rep["hooks"].items():
|
|
c = s["coverage"]
|
|
regions = ", ".join(c["checked_regions"][:6]) or "-"
|
|
if len(c["checked_regions"]) > 6:
|
|
regions += f", +{len(c['checked_regions']) - 6}"
|
|
guards = ", ".join(c["guards"]) or "-"
|
|
und = (f"{c['undeclared_writes']} in {c['undeclared_calls']} call(s)"
|
|
if c["undeclared_writes"] else ("0" if c["guarded_calls"] else "not watched"))
|
|
P(f"| {h} | {c.get('verdict', c['state'])} | {regions} | {guards} | {und} | "
|
|
f"{len(c['unmodelled'])} |")
|
|
for h, s in rep["hooks"].items():
|
|
c = s["coverage"]
|
|
if not c["unmodelled"] and not c["spans"]:
|
|
continue
|
|
P()
|
|
P(f"#### {h} — not checked by this run")
|
|
if c.get("why"):
|
|
P(f"- claims complete coverage: {c['why']}")
|
|
for n in c["unmodelled"]:
|
|
mit = f" [{n['mitigation']}]" if n.get("mitigation") else ""
|
|
P(f"- ({n.get('risk', '?')}) {n.get('what', '?')} — {n.get('why', '')}{mit}")
|
|
for mode, spans in sorted(c["spans"].items()):
|
|
if spans:
|
|
P(f"- guard hits in {mode} mode: {', '.join(spans)}")
|
|
# A write the original makes in compare mode but ours never makes in replace mode (or
|
|
# the reverse) is the asymmetry that a save-hash oracle would eventually catch the hard
|
|
# way. Say it here instead.
|
|
if "compare" in c["spans"] and "replace" in c["spans"]:
|
|
cmp_s, rep_s = set(c["spans"]["compare"]), set(c["spans"]["replace"])
|
|
only_c = sorted(cmp_s - rep_s)
|
|
only_r = sorted(rep_s - cmp_s)
|
|
if only_c:
|
|
P(f"- ONLY the original writes these: {', '.join(only_c)}")
|
|
if only_r:
|
|
P(f"- ONLY ours writes these: {', '.join(only_r)}")
|
|
|
|
|
|
def print_report(rep: dict, out=None) -> None:
|
|
out = out or sys.stdout
|
|
P = lambda s="": print(s, file=out) # noqa: E731
|
|
P(f"## tracecmp {rep['kind']}: {', '.join(os.path.basename(p) for p in rep['inputs'])}")
|
|
P()
|
|
for m in rep["meta"]:
|
|
P(f"- build: {m.get('build', '?')} started: {m.get('started', '?')} inline_max: {m.get('inline_max', '?')}")
|
|
t = rep["totals"]
|
|
P(f"- calls: {t['calls']} compared: {t['compared']} diverged: {t['diverged']} "
|
|
f"invalid records: {t['invalid_records']} warnings: {len(rep['warnings'])}")
|
|
P(f"- coverage: {t.get('guarded_calls', 0)} guarded call(s), "
|
|
f"{t.get('undeclared_writes', 0)} undeclared write(s) in {t.get('undeclared_calls', 0)} call(s); "
|
|
f"{t.get('coverage_unstated', 0)} hook(s) unstated, "
|
|
f"{t.get('coverage_contradicted', 0)} contradicted")
|
|
P()
|
|
P("| hook | calls | modes | compared | diverged | errors |")
|
|
P("|---|---|---|---|---|---|")
|
|
for h, s in rep["hooks"].items():
|
|
modes = " ".join(f"{k}:{v}" for k, v in sorted(s["modes"].items()))
|
|
P(f"| {h} | {s['calls']} | {modes} | {s['compared']} | {s['diverged']} | {s['errors']} |")
|
|
print_coverage(rep, out)
|
|
for h, s in rep["hooks"].items():
|
|
if not s["diffs"]:
|
|
continue
|
|
P()
|
|
P(f"### {h}: first {len(s['diffs'])} of {s['diverged']} divergent call(s)")
|
|
for d in s["diffs"]:
|
|
P(f"- call_id {d['call_id']} ({d['file']}:{d['line']})")
|
|
for e in d["diff"][:8]:
|
|
extra = "".join(f" {k}={v}" for k, v in e.items() if k not in ("path", "why", "orig", "ours"))
|
|
P(f" {e['path']} [{e['why']}] orig={_short(e['orig'])} ours={_short(e['ours'])}{extra}")
|
|
if len(d["diff"]) > 8:
|
|
P(f" … {len(d['diff']) - 8} more")
|
|
more = s["diverged_call_ids"][len(s["diffs"]):]
|
|
if more:
|
|
P(f" other divergent call_ids: {more[:20]}{' …' if len(more) > 20 else ''}")
|
|
if rep["invalid"]:
|
|
P()
|
|
P(f"### invalid records ({len(rep['invalid'])})")
|
|
for e in rep["invalid"][:20]:
|
|
P(f"- {e['file']}:{e['line']}: {e['error']}")
|
|
if rep["warnings"]:
|
|
P()
|
|
P(f"### warnings ({len(rep['warnings'])})")
|
|
for w in rep["warnings"][:20]:
|
|
P(f"- {w}")
|
|
|
|
|
|
# --- main ----------------------------------------------------------------------------
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0], formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
ap.add_argument("logs", nargs="+", metavar="LOG", help="trace/compare JSONL (golden logs in --replay)")
|
|
ap.add_argument("--replay", metavar="IMPL", help="implementation output JSONL to diff against LOG")
|
|
ap.add_argument("--hook", action="append", default=[], help="only this hook (repeatable)")
|
|
ap.add_argument("--tolerance", action="append", default=[], metavar="SPEC", help="[HOOK=][abs:|rel:|ulp:]NUMBER")
|
|
ap.add_argument("--unordered", action="append", default=[], metavar="SPEC", help="[HOOK=]PATH")
|
|
ap.add_argument("--ptr", choices=("ignore", "exact"), default=None)
|
|
ap.add_argument("--numeric", action="store_true")
|
|
ap.add_argument("--first", type=int, default=5, metavar="N")
|
|
ap.add_argument("--json-out", metavar="PATH")
|
|
ap.add_argument("--strict-coverage", action="store_true",
|
|
help="also fail (exit 1) on any undeclared write or unstated hook, not only "
|
|
"on a hook whose 'complete' claim a guard contradicted")
|
|
ap.add_argument("--skip-invalid", action="store_true")
|
|
ap.add_argument("--quiet", action="store_true", help="no markdown report on stdout")
|
|
return ap
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = build_parser()
|
|
a = ap.parse_args(argv)
|
|
policies = PolicyTable()
|
|
try:
|
|
for s in a.tolerance:
|
|
policies.add_cli(*parse_tolerance(s))
|
|
for s in a.unordered:
|
|
policies.add_cli(*parse_unordered(s))
|
|
except ValueError as e:
|
|
ap.error(str(e))
|
|
if a.ptr:
|
|
policies.add_cli(None, {"ptr": a.ptr})
|
|
if a.numeric:
|
|
policies.add_cli(None, {"numeric": True})
|
|
hooks = set(a.hook) or None
|
|
|
|
logs = []
|
|
for p in a.logs:
|
|
if not os.path.exists(p):
|
|
print(f"error: no such file {p}", file=sys.stderr)
|
|
return 2
|
|
logs.append(load_log(p, policies))
|
|
impl_invalid: list = []
|
|
if a.replay:
|
|
if not os.path.exists(a.replay):
|
|
print(f"error: no such file {a.replay}", file=sys.stderr)
|
|
return 2
|
|
impl, impl_invalid = load_impl(a.replay)
|
|
rep = run_replay(logs, impl, policies, hooks, a.first)
|
|
rep["impl"] = a.replay
|
|
rep["invalid"].extend({"file": os.path.basename(a.replay), "line": ln, "error": m} for ln, m in impl_invalid)
|
|
rep["totals"]["invalid_records"] += len(impl_invalid)
|
|
else:
|
|
rep = run_report(logs, policies, hooks, a.first)
|
|
|
|
if a.json_out:
|
|
os.makedirs(os.path.dirname(os.path.abspath(a.json_out)), exist_ok=True)
|
|
with open(a.json_out, "w", encoding="utf-8") as f:
|
|
json.dump(rep, f, indent=1, sort_keys=True)
|
|
f.write("\n")
|
|
if not a.quiet:
|
|
print_report(rep)
|
|
if rep["totals"]["invalid_records"] and not a.skip_invalid:
|
|
return 2
|
|
# A descriptor that claims complete coverage while a guard watched the original write
|
|
# outside every declared region has been proved wrong: that is a divergence between the
|
|
# harness's model and the game, and it fails the run like any other.
|
|
failed = rep["totals"]["diverged"] or rep["totals"]["coverage_contradicted"]
|
|
if a.strict_coverage:
|
|
failed = failed or rep["totals"]["undeclared_writes"] or rep["totals"]["coverage_unstated"]
|
|
return 1 if failed else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|