tai/taicpp/tools/parity_ledger.py

#!/usr/bin/env python3
"""Parity ledger: every tai.py symbol against the C++ tree and the differential results.

Usage: tools/parity_ledger.py [--check] [--publish]

Writes build/parity/ledger.json, ledger.md and PARITY.md (all under build/).
--publish also copies the generated PARITY.md over taicpp/PARITY.md (the committed copy).
--check exits 1 while ANY symbol is missing or unverified, a case names a symbol the
ledger does not know, or the differential results are absent or stale.

Status per symbol (strongest first):
  verified   a passing differential case (tools/parity_diff.py, unfiltered run, tai.py and
             detz hashes equal to the current files) names the symbol. Facets list WHAT the
             case verified (`parse`, `output`, ...); verified never means "all behavior".
  name-only  an identifier in src/ or main.cpp matches the normalized name. Behavior unproven.
  missing    no identifier matches.
Only a differential case sets `verified`; nothing here infers it.

Symbol ids are `kind:name`, optionally `#facet` in case files:
  class:Store  method:Store.load  function:main  tool:shell  flag:--version
  setting:model  command:/help
"""
import ast
import hashlib
import json
import os
import re
import sys

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
REPO = os.path.dirname(ROOT)
PY = os.path.join(REPO, "tai.py")
OUT = os.path.join(ROOT, "build", "parity")
DETZ = os.environ.get("DETZ_BIN", os.path.join(ROOT, "build", "detz"))
CASES_DIR = os.path.join(ROOT, "tests", "diff", "cases")
PARITY_MD = os.path.join(ROOT, "PARITY.md")
KINDS = ("class", "method", "function", "tool", "flag", "setting", "command")


def norm(name):
    return re.sub(r"[^a-z0-9]", "", name.lower())


def sha256_file(path):
    if not os.path.isfile(path):
        return "missing"
    digest = hashlib.sha256()
    with open(path, "rb") as fh:
        for block in iter(lambda: fh.read(1 << 20), b""):
            digest.update(block)
    return digest.hexdigest()


class PySymbols:
    """Symbols extracted from tai.py with ast."""

    def __init__(self, path):
        with open(path, encoding="utf-8") as fh:
            self.tree = ast.parse(fh.read())

    def code(self):
        out = []
        for node in self.tree.body:
            if isinstance(node, ast.ClassDef):
                out.append(("class", node.name, node.lineno))
                for sub in node.body:
                    if isinstance(sub, (ast.FunctionDef, ast.AsyncFunctionDef)):
                        out.append(("method", "%s.%s" % (node.name, sub.name), sub.lineno))
            elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                out.append(("function", node.name, node.lineno))
        return out

    def tools(self):
        out = []
        for node in ast.walk(self.tree):
            if isinstance(node, ast.Call) and getattr(node.func, "id", "") == "tool_schema" and node.args:
                arg = node.args[0]
                if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                    out.append(("tool", arg.value, node.lineno))
        return out

    def flags(self):
        out = []
        for node in ast.walk(self.tree):
            if isinstance(node, ast.Call) and getattr(node.func, "attr", "") == "add_argument":
                for arg in node.args:
                    if isinstance(arg, ast.Constant) and isinstance(arg.value, str) and arg.value.startswith("--"):
                        out.append(("flag", arg.value, node.lineno))
        return out

    def settings(self):
        out = []
        for node in self.tree.body:
            if isinstance(node, ast.Assign) and any(getattr(t, "id", "") == "SETTINGS_CATALOG" for t in node.targets):
                for sub in ast.walk(node):
                    if isinstance(sub, ast.Call) and getattr(sub.func, "id", "") == "setting_spec" and sub.args:
                        arg = sub.args[0]
                        if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
                            out.append(("setting", arg.value, sub.lineno))
        return out

    def commands(self):
        out = []
        for node in self.tree.body:
            if isinstance(node, ast.Assign) and any(getattr(t, "id", "") == "HELP_TEXT" for t in node.targets):
                text = "".join(c.value for c in ast.walk(node.value) if isinstance(c, ast.Constant) and isinstance(c.value, str))
                seen = set()
                for line in text.splitlines():
                    m = re.match(r"\s+(/[a-z]+)", line)
                    if m and m.group(1) not in seen:
                        seen.add(m.group(1))
                        out.append(("command", m.group(1), node.lineno))
        return out

    def all(self):
        return self.code() + self.tools() + self.flags() + self.settings() + self.commands()


class CppIndex:
    """Normalized identifiers found in the C++ tree."""

    def __init__(self):
        self.idents = set()
        files = [os.path.join(ROOT, "main.cpp")]
        for base, _dirs, names in os.walk(os.path.join(ROOT, "src")):
            files += [os.path.join(base, n) for n in names if n.endswith((".cpp", ".h", ".hpp"))]
        for path in files:
            if os.path.isfile(path):
                with open(path, encoding="utf-8", errors="replace") as fh:
                    self.idents.update(norm(t) for t in re.findall(r"[A-Za-z_][A-Za-z0-9_]*", fh.read()))

    def has(self, kind, name):
        if kind in ("tool", "setting"):
            return norm(name) in self.idents
        if kind == "flag":
            return norm(name) in self.idents
        if kind == "command":
            return norm(name.lstrip("/")) in self.idents and len(norm(name)) > 3
        leaf = name.split(".")[-1]
        if norm(name.replace(".", "")) in self.idents:
            return True
        return kind != "class" and len(norm(leaf)) > 5 and norm(leaf) in self.idents


class Evidence:
    """Passing differential cases, and whether they may be trusted for the current files."""

    def __init__(self):
        self.path = os.path.join(OUT, "diff-results.json")
        self.problems = []
        self.by_symbol = {}   # "kind:name" -> {"cases": [ids], "facets": set}
        self.cases = {}       # id -> result dict
        self.meta = None
        if not os.path.isfile(self.path):
            self.problems.append("no differential results (run tools/parity_diff.py)")
            return
        with open(self.path) as fh:
            doc = json.load(fh)
        self.meta = doc["meta"]
        now_py, now_detz = sha256_file(PY), sha256_file(self.meta.get("detz_path", DETZ))
        if self.meta["tai_sha"] != now_py:
            self.problems.append("differential results are stale: tai.py changed since the run")
        if self.meta["detz_sha"] != now_detz:
            self.problems.append("differential results are stale: detz binary changed or missing since the run")
        if self.meta.get("no_llm"):
            self.problems.append("last differential run skipped LLM cases (--no-llm)")
        trusted = not any("stale" in p for p in self.problems)
        for res in doc["results"]:
            self.cases[res["id"]] = res
            if res["status"] != "pass" or not trusted:
                continue
            for sym in res["symbols"]:
                base, _, facet = sym.partition("#")
                slot = self.by_symbol.setdefault(base, {"cases": [], "facets": set()})
                slot["cases"].append(res["id"])
                slot["facets"].add(facet or "all")


class Ledger:
    def __init__(self):
        self.py = PySymbols(PY)
        self.cpp = CppIndex()
        self.evidence = Evidence()
        self.rows = []
        self.case_symbol_errors = []
        self._build()

    def _build(self):
        known = set()
        for kind, name, line in self.py.all():
            sid = "%s:%s" % (kind, name)
            if sid in known:
                continue
            known.add(sid)
            ev = self.evidence.by_symbol.get(sid)
            if ev:
                status, facets, cases = "verified", sorted(ev["facets"]), ev["cases"]
            else:
                status, facets, cases = ("name-only" if self.cpp.has(kind, name) else "missing"), [], []
            self.rows.append({"id": sid, "kind": kind, "name": name, "line": line, "status": status, "facets": facets, "cases": cases})
        for cid, res in self.evidence.cases.items():
            for sym in res["symbols"]:
                if sym.partition("#")[0] not in known:
                    self.case_symbol_errors.append("case %s names unknown symbol %s" % (cid, sym))

    def count(self, kind=None):
        rows = [r for r in self.rows if kind is None or r["kind"] == kind]
        tally = {"total": len(rows), "verified": 0, "name-only": 0, "missing": 0}
        for r in rows:
            tally[r["status"]] += 1
        return tally

    def problems(self):
        out = list(self.evidence.problems) + self.case_symbol_errors
        t = self.count()
        if t["missing"]:
            out.append("%d symbols missing" % t["missing"])
        if t["name-only"]:
            out.append("%d symbols name-only (unverified)" % t["name-only"])
        return out


class Writer:
    def __init__(self, ledger):
        self.ledger = ledger

    def tally_table(self):
        lines = ["| kind | total | verified | name-only | missing |", "|---|---|---|---|---|"]
        for kind in KINDS + (None,):
            t = self.ledger.count(kind)
            lines.append("| %s | %d | %d | %d | %d |" % (kind or "**all**", t["total"], t["verified"], t["name-only"], t["missing"]))
        return lines

    def parity_md(self):
        led, ev = self.ledger, self.ledger.evidence
        meta = ev.meta or {}
        lines = ["# Parity detz vs tai.py (generated, do not edit)", "",
                 "Generated by `tools/parity_ledger.py` from the symbol ledger and the last differential run.",
                 "Only a passing differential case (`tools/parity_diff.py`) makes a symbol `verified`; a facet",
                 "after the symbol says what that case proved (for example `parse` means argument parsing only).",
                 "`name-only` means an identifier matches in the C++ tree and NOTHING about behavior is known.", "",
                 "- tai.py sha256: `%s`" % sha256_file(PY)[:16],
                 "- detz sha256: `%s`" % sha256_file(meta.get("detz_path", DETZ))[:16],
                 "- differential run: %s (tai.py `%s`, detz `%s`)" % (meta.get("ts", "none"), meta.get("tai_sha", "-")[:16], meta.get("detz_sha", "-")[:16]), ""]
        if ev.problems:
            lines += ["## Evidence problems", ""] + ["- " + p for p in ev.problems] + [""]
        lines += ["## Summary", ""] + self.tally_table() + [""]
        results = list(ev.cases.values())
        counts = {}
        for r in results:
            counts[r["status"]] = counts.get(r["status"], 0) + 1
        lines += ["## Differential cases", "", "Counts: " + (", ".join("%s=%d" % kv for kv in sorted(counts.items())) or "none"), ""]
        failing = [r for r in results if r["status"] in ("fail", "xfail", "xpass")]
        if failing:
            lines += ["| case | status | reason |", "|---|---|---|"]
            for r in sorted(failing, key=lambda x: x["id"]):
                reason = r["expected_fail"] or (r["problems"][0] if r["problems"] else "")
                lines.append("| %s | %s | %s |" % (r["id"], r["status"], reason.replace("|", "/")[:120]))
            lines.append("")
        tm = os.path.join(OUT, "tests-map.json")
        if os.path.isfile(tm):
            with open(tm) as fh:
                cov = json.load(fh)["summary"]
            lines += ["## Python test-suite coverage", "",
                      "%d test classes, %d tests; tests reproduced by a passing diff case: %d (in %d classes); classes with no covered test: %d."
                      % (cov["classes"], cov["tests"], cov["tests_covered"], cov["classes_covered"], cov["classes_uncovered"]),
                      "Full table: `build/parity/tests-map.md`.", ""]
        for kind in ("flag", "tool", "setting", "command"):
            rows = [r for r in led.rows if r["kind"] == kind]
            lines += ["## %ss (%d)" % (kind, len(rows)), "", "| name | status | facets | cases |", "|---|---|---|---|"]
            for r in sorted(rows, key=lambda x: x["name"]):
                lines.append("| `%s` | %s | %s | %s |" % (r["name"], r["status"], ",".join(r["facets"]), ",".join(r["cases"][:3])))
            lines.append("")
        miss = {}
        for r in led.rows:
            if r["kind"] == "method" and r["status"] == "missing":
                miss.setdefault(r["name"].split(".")[0], 0)
                miss[r["name"].split(".")[0]] += 1
        top = sorted(miss.items(), key=lambda kv: -kv[1])[:25]
        lines += ["## Classes with the most missing methods (top 25 of %d)" % len(miss), "", "| class | missing methods |", "|---|---|"]
        lines += ["| %s | %d |" % kv for kv in top]
        lines.append("")
        return "\n".join(lines)

    def ledger_md(self):
        lines = ["# Parity ledger (generated, do not edit)", ""] + self.tally_table() + [""]
        lines += ["| id | status | facets | cases |", "|---|---|---|---|"]
        for r in self.ledger.rows:
            lines.append("| `%s` | %s | %s | %s |" % (r["id"], r["status"], ",".join(r["facets"]), ",".join(r["cases"][:3])))
        return "\n".join(lines) + "\n"

    def write(self, publish):
        os.makedirs(OUT, exist_ok=True)
        with open(os.path.join(OUT, "ledger.json"), "w") as fh:
            json.dump(self.ledger.rows, fh, indent=1)
        with open(os.path.join(OUT, "ledger.md"), "w") as fh:
            fh.write(self.ledger_md())
        text = self.parity_md()
        with open(os.path.join(OUT, "PARITY.md"), "w") as fh:
            fh.write(text)
        if publish:
            with open(PARITY_MD, "w") as fh:
                fh.write(text)


def main():
    ledger = Ledger()
    Writer(ledger).write("--publish" in sys.argv)
    print("\n".join(Writer(ledger).tally_table()))
    problems = ledger.problems()
    for p in problems:
        print("PROBLEM: %s" % p)
    if "--check" in sys.argv:
        if problems:
            print("parity-check: FAIL (%d problems)" % len(problems))
            return 1
        print("parity-check: OK")
    return 0


if __name__ == "__main__":
    sys.exit(main())