#!/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/.json) --skip-invalid drop unparsable/invalid records with a warning instead of failing the run exit: 0 = no divergences; 1 = divergences; 2 = invalid input (or usage). 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", "err", "note"} 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_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 "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(" 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 ------------------------------------------------------------------------ def new_hook_stats() -> dict: return {"calls": 0, "modes": {}, "compared": 0, "diverged": 0, "errors": 0, "diverged_call_ids": [], "diffs": []} 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 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 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] 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)}, "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_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() 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']} |") 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("--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 return 1 if rep["totals"]["diverged"] else 0 if __name__ == "__main__": sys.exit(main())