silk-graph 0.3.0

"""Audit Silk's verifiable claims across docs, tests, and formal specs.

Scans PROOF.md and INVARIANTS.md for claim identifiers (I-01, Theorem 3, INV-2,
...), then greps Rust tests, Python tests, and TLA+ specs for references to
each. Produces formal/audit.json + a terminal summary. Exits non-zero if any
claim has zero verification surfaces.

Run: python scripts/audit_claims.py
"""

from __future__ import annotations

import json
import os
import re
import sys
from dataclasses import dataclass, asdict
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent

CLAIM_PATTERNS = {
    "invariant": re.compile(r"\bI-0[1-9]\b"),
    "theorem": re.compile(r"\bTheorem [1-9]\b"),
    "inv": re.compile(r"\bINV-[1-9]\b"),
}

# Claims explicitly out of TLA+ scope, with rationale. These won't show up as
# "formalizable gaps" in the summary.
TLA_INELIGIBLE = {
    "I-01": "cryptographic hash integrity; verified by unit test, not structural reasoning",
    "I-06": "quarantine determinism is a corollary of Theorem 3 (PROOF.md §6)",
    "Theorem 4": "composition of two semilattices; proved algebraically in PROOF.md Appendix A (standard result, would not add model-checker value)",
    "Theorem 5": "trivially follows from Theorem 1 (topo sort determinism); proved semi-formally in PROOF.md Appendix A",
}

SOURCES = {
    "invariant": [ROOT / "PROOF.md"],
    "theorem": [ROOT / "PROOF.md"],
    "inv": [ROOT / "INVARIANTS.md"],
}

REFERENCE_SCAN = {
    "rust_tests": sorted(ROOT.glob("src/**/*.rs")) + sorted(ROOT.glob("tests/**/*.rs")),
    "python_tests": sorted(ROOT.glob("pytests/**/*.py")),
    "tla_specs": sorted(ROOT.glob("formal/*.tla")) + sorted(ROOT.glob("formal/*.md")),
}


@dataclass
class ClaimCoverage:
    claim_id: str
    kind: str
    rust_tests: list[str]
    python_tests: list[str]
    tla_specs: list[str]
    tla_eligible: bool
    tla_ineligible_reason: str | None

    @property
    def covered(self) -> bool:
        return bool(self.rust_tests or self.python_tests or self.tla_specs)

    @property
    def surface_count(self) -> int:
        return sum(
            1 for s in (self.rust_tests, self.python_tests, self.tla_specs) if s
        )


def extract_claims() -> dict[str, str]:
    """Return {claim_id: kind} across all sources."""
    claims: dict[str, str] = {}
    for kind, paths in SOURCES.items():
        pattern = CLAIM_PATTERNS[kind]
        for path in paths:
            if not path.exists():
                continue
            text = path.read_text()
            for match in pattern.findall(text):
                claims[match] = kind
    return claims


def scan_references(claim_id: str) -> dict[str, list[str]]:
    """Return references to claim_id grouped by surface."""
    refs: dict[str, list[str]] = {k: [] for k in REFERENCE_SCAN}
    pattern = re.compile(rf"\b{re.escape(claim_id)}\b")
    for surface, paths in REFERENCE_SCAN.items():
        for path in paths:
            try:
                if pattern.search(path.read_text()):
                    refs[surface].append(str(path.relative_to(ROOT)))
            except (UnicodeDecodeError, OSError):
                continue
    return refs


def build_report() -> dict:
    claims = extract_claims()
    coverages = []
    for claim_id, kind in sorted(claims.items()):
        refs = scan_references(claim_id)
        coverages.append(
            ClaimCoverage(
                claim_id=claim_id,
                kind=kind,
                rust_tests=refs["rust_tests"],
                python_tests=refs["python_tests"],
                tla_specs=refs["tla_specs"],
                tla_eligible=claim_id not in TLA_INELIGIBLE,
                tla_ineligible_reason=TLA_INELIGIBLE.get(claim_id),
            )
        )

    total = len(coverages)
    covered = sum(1 for c in coverages if c.covered)
    tla_covered = sum(1 for c in coverages if c.tla_specs)
    test_covered = sum(1 for c in coverages if c.rust_tests or c.python_tests)
    tla_eligible_total = sum(
        1 for c in coverages if c.tla_eligible and c.kind in ("invariant", "theorem")
    )
    tla_eligible_covered = sum(
        1 for c in coverages
        if c.tla_eligible and c.tla_specs and c.kind in ("invariant", "theorem")
    )

    by_kind: dict[str, dict[str, int]] = {}
    for c in coverages:
        bucket = by_kind.setdefault(
            c.kind, {"total": 0, "covered": 0, "tla": 0, "tests": 0}
        )
        bucket["total"] += 1
        if c.covered:
            bucket["covered"] += 1
        if c.tla_specs:
            bucket["tla"] += 1
        if c.rust_tests or c.python_tests:
            bucket["tests"] += 1

    return {
        "summary": {
            "total_claims": total,
            "covered": covered,
            "tla_covered": tla_covered,
            "test_covered": test_covered,
            "coverage_pct": round(100 * covered / total, 1) if total else 0.0,
            "tla_pct": round(100 * tla_covered / total, 1) if total else 0.0,
            "tla_eligible_total": tla_eligible_total,
            "tla_eligible_covered": tla_eligible_covered,
            "tla_eligible_pct": (
                round(100 * tla_eligible_covered / tla_eligible_total, 1)
                if tla_eligible_total else 0.0
            ),
            "by_kind": by_kind,
        },
        "claims": [asdict(c) for c in coverages],
    }


def print_summary(report: dict) -> None:
    s = report["summary"]
    print(f"Silk claim coverage audit")
    print(f"=" * 50)
    print(f"Total claims:   {s['total_claims']}")
    print(f"Any coverage:   {s['covered']}/{s['total_claims']} ({s['coverage_pct']}%)")
    print(f"Test coverage:  {s['test_covered']}/{s['total_claims']}")
    print(
        f"TLA+ (eligible): {s['tla_eligible_covered']}/{s['tla_eligible_total']} "
        f"({s['tla_eligible_pct']}%)"
    )
    print()
    print("By kind:")
    for kind, b in sorted(s["by_kind"].items()):
        print(
            f"  {kind:10s} covered {b['covered']}/{b['total']} "
            f"(tla {b['tla']}, tests {b['tests']})"
        )
    print()

    uncovered = [c for c in report["claims"] if not (c["rust_tests"] or c["python_tests"] or c["tla_specs"])]
    if uncovered:
        print("UNCOVERED:")
        for c in uncovered:
            print(f"  {c['claim_id']} ({c['kind']})")
        print()

    no_tla_eligible = [
        c for c in report["claims"]
        if c["tla_eligible"]
        and not c["tla_specs"]
        and c["kind"] in ("invariant", "theorem")
    ]
    if no_tla_eligible:
        print("Formalizable (not yet in TLA+):")
        for c in no_tla_eligible:
            print(f"  {c['claim_id']}")
        print()

    ineligible = [c for c in report["claims"] if not c["tla_eligible"]]
    if ineligible:
        print("Out of TLA+ scope (by design):")
        for c in ineligible:
            print(f"  {c['claim_id']}: {c['tla_ineligible_reason']}")
        print()


# H7: a CHANGELOG citation that does not resolve buys false confidence at
# exactly the moment someone goes looking for reassurance. The 0.1.7 row cited
# an "integration test in src/python/mod.rs" — a file with zero tests — and the
# Bug 5 row cited a test that contains no call to the mechanism it covers.
# Both survived because nothing checked. This does.
TEST_IDENT = re.compile(r"`([^`]+)`")
PY_TEST = re.compile(r"^(?P<path>(?:pytests|experiments)/[\w/]+\.py)::(?P<name>[\w:]+)$")
RS_TEST = re.compile(r"^(?P<path>src/[\w/]+\.rs)::(?P<name>\w+)$")


def audit_changelog_citations() -> list[str]:
    """Every regression-test citation in the CHANGELOG bug table must name an
    identifier that resolves to a real test. Free text does not resolve and is
    reported, not waved through."""
    changelog = (ROOT / "CHANGELOG.md").read_text().splitlines()
    problems: list[str] = []

    for lineno, line in enumerate(changelog, 1):
        if not line.startswith("|") or line.count("|") < 6:
            continue
        cells = [c.strip() for c in line.strip("|").split("|")]
        if len(cells) < 5 or cells[0] in ("Version", "---------"):
            continue
        if set(cells[0]) <= set("- "):
            continue
        citation = cells[4]
        row_problems: list[str] = []
        resolved = False
        for ident in TEST_IDENT.findall(citation):
            for pattern in (PY_TEST, RS_TEST):
                m = pattern.match(ident)
                if not m:
                    continue
                path = ROOT / m.group("path")
                if not path.is_file():
                    row_problems.append(
                        f"CHANGELOG.md:{lineno}: cited file does not exist: {ident}")
                    continue
                name = m.group("name").split("::")[-1]
                body = path.read_text()
                if f"def {name}" in body or f"fn {name}" in body:
                    resolved = True
                else:
                    row_problems.append(
                        f"CHANGELOG.md:{lineno}: cited test not found: {ident}")
        if not resolved:
            row_problems.append(
                f"CHANGELOG.md:{lineno}: version {cells[0]} cites no resolvable "
                f"test identifier (got: {citation[:70]!r})")
            problems.extend(row_problems)
        # A row with at least one resolving citation is covered; a stale
        # sibling citation in the same row is still reported.
        else:
            problems.extend(p for p in row_problems if "not found" in p or "does not exist" in p)
    return problems


def main() -> int:
    report = build_report()
    out_path = ROOT / "formal" / "audit.json"
    out_path.write_text(json.dumps(report, indent=2) + "\n")
    print_summary(report)
    print(f"Wrote {out_path.relative_to(ROOT)}")

    citation_problems = audit_changelog_citations()
    if citation_problems:
        print("\nCHANGELOG citation failures:")
        for p in citation_problems:
            print(f"  {p}")

    uncovered = [
        c for c in report["claims"]
        if not (c["rust_tests"] or c["python_tests"] or c["tla_specs"])
    ]
    return 1 if (uncovered or citation_problems) else 0


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