tool-crosscheck/crosscheck.py

#!/usr/bin/env python3
"""crosscheck — mechanical consistency checks across the yukart repositories.

Facts in yukart are stated in more than one repo: a decision's status, a topic, a
service list, a path in another repo's docs. This finds the places where those
copies no longer agree with their source. No model, no network: it reads the
tracked files of every repo under ROOT and reports file:line findings.

    crosscheck.py [ROOT]          ROOT holds the repos side by side (default: .)

Exit status 1 when anything is found. Checks, and the source each one trusts:

  decision   "decision N" names a row of project-yukart's decisions table, and a
             line calling it proposed does not contradict an Accepted row
  topic      every yukart/<...> topic matches contracts' topics.json; concrete
             vehicle and host signals exist in signals.json with that source
  docref     "<repo> `path.md`" names a file that exists in that repo; relative
             markdown links resolve
  binaries   REPOS.md's row for a repo names every deployed [[bin]] in its Cargo.toml;
             ORIN-RUNTIME.md names every service a repo deploys
  frontmatter  every docs/**/*.md starts with front matter whose type is known
"""

import json
import os
import re
import subprocess
import sys
from pathlib import Path

DOC_TYPES = {"guidelines", "plan", "report", "reference"}
# Test-only topic roots, deliberately outside contracts (svc-core bus-bench).
IGNORED_TOPIC_ROOTS = {"bench"}
TEXT_SUFFIXES = {".md", ".rs", ".dart", ".py", ".json", ".service", ".sh", ".yaml", ".yml", ".toml"}

findings: list[str] = []


def find(path: str, line: int, check: str, msg: str) -> None:
    findings.append(f"{path}:{line}: [{check}] {msg}")


def tracked(repo: Path) -> list[str]:
    out = subprocess.run(["git", "-C", str(repo), "ls-files"], capture_output=True, text=True, check=True)
    return [f for f in out.stdout.splitlines() if Path(f).suffix in TEXT_SUFFIXES]


def lines_outside_fences(text: str):
    """(line number, line) for prose lines; fenced code blocks are skipped."""
    fenced = False
    for n, line in enumerate(text.splitlines(), 1):
        if line.lstrip().startswith("```"):
            fenced = not fenced
            continue
        if not fenced:
            yield n, line


class Workspace:
    def __init__(self, root: Path):
        self.root = root
        self.repos = sorted(p.name for p in root.iterdir() if (p / ".git").exists())
        self.files: dict[str, list[str]] = {r: tracked(root / r) for r in self.repos}
        self.texts: dict[tuple[str, str], str] = {}

    def text(self, repo: str, rel: str) -> str:
        key = (repo, rel)
        if key not in self.texts:
            try:
                self.texts[key] = (self.root / repo / rel).read_text(encoding="utf-8")
            except (OSError, UnicodeDecodeError):
                self.texts[key] = ""
        return self.texts[key]

    def each(self, suffixes=None):
        for repo in self.repos:
            for rel in self.files[repo]:
                if suffixes is None or Path(rel).suffix in suffixes:
                    yield repo, rel, self.text(repo, rel)


# ── decisions ────────────────────────────────────────────────────────────────

DECISION_ROW = re.compile(r"^\|\s*(\d+)\s*\|.*\|\s*\**(Accepted|Proposed|Rejected|Superseded)[^|]*\|", re.I)
DECISION_REF = re.compile(r"\b[Dd]ecisions?\s+(\d+(?:\s*(?:,|and|or)\s*\d+)*)")


def check_decisions(ws: Workspace) -> None:
    arch = ("project-yukart", "docs/ARCHITECTURE.md")
    table = {}
    for n, line in enumerate(ws.text(*arch).splitlines(), 1):
        m = DECISION_ROW.match(line)
        if m:
            table[int(m.group(1))] = m.group(2).capitalize()
    if not table:
        find("/".join(arch), 1, "decision", "no decisions table found")
        return
    for repo, rel, text in ws.each({".md", ".rs", ".dart", ".py"}):
        for n, line in enumerate(text.splitlines(), 1):
            if DECISION_ROW.match(line):
                continue
            for m in DECISION_REF.finditer(line):
                for num in map(int, re.findall(r"\d+", m.group(1))):
                    status = table.get(num)
                    if status is None:
                        find(f"{repo}/{rel}", n, "decision", f"decision {num} is not in the decisions table")
                    elif status == "Accepted" and re.search(r"\bproposed\b", line, re.I):
                        find(f"{repo}/{rel}", n, "decision", f"says proposed, but decision {num} is Accepted")


# ── topics ───────────────────────────────────────────────────────────────────

TOPIC = re.compile(r"(?<![\w/:.@-])yukart/([a-z][a-z0-9_-]*)((?:/[A-Za-z0-9_<>+#*.-]*)*)")
# Same shape, not a topic: Flutter platform channel names, systemd directory names.
NOT_TOPIC = re.compile(r"MethodChannel\(|EventChannel\(|Directory=")


def segment_ok(mention: str, pattern: str) -> bool:
    if pattern.startswith("<"):
        return True
    return mention == pattern or mention == "+" or mention.startswith("<") or "*" in mention


def topic_matches(segs: list[str], pattern: list[str]) -> bool:
    for i, s in enumerate(segs):
        if s == "#":
            return True
        if s == "":  # a prefix mention, "yukart/vehicle/"
            return i == len(segs) - 1 and len(pattern) > i
        if i >= len(pattern) or not segment_ok(s, pattern[i]):
            return False
    return len(segs) == len(pattern)


def check_topics(ws: Workspace) -> None:
    topics = json.loads(ws.text("contracts", "schema/topics.json"))["topics"]
    signals = json.loads(ws.text("contracts", "schema/signals.json"))["signals"]
    patterns = [t.split("/") for t in topics]
    sources = {"vehicle": {"obd2", "z27ag"}, "host": {"host"}}
    for repo, rel, text in ws.each():
        if (repo, rel) == ("contracts", "schema/topics.json"):
            continue
        for n, line in enumerate(text.splitlines(), 1):
            if NOT_TOPIC.search(line):
                continue
            for m in TOPIC.finditer(line):
                root, rest = m.group(1), m.group(2).rstrip(".")
                # Topic levels never contain '-'; repo names (yukart/svc-core) usually do.
                if root in ws.repos or "-" in root or root in IGNORED_TOPIC_ROOTS:
                    continue
                segs = ["yukart", root] + rest.split("/")[1:]
                if rest == "":
                    segs.append("")
                if not any(topic_matches(segs, p) for p in patterns):
                    find(f"{repo}/{rel}", n, "topic", f"'{m.group(0)}' matches no topic in contracts topics.json")
                    continue
                if root in sources and len(segs) == 3 and "*" in segs[2]:
                    glob = re.compile(re.escape(segs[2]).replace(r"\*", ".*"))
                    if not any(glob.fullmatch(k) and v.get("source") in sources[root] for k, v in signals.items()):
                        find(f"{repo}/{rel}", n, "topic", f"no {root} signal in contracts signals.json matches '{segs[2]}'")
                elif root in sources and len(segs) == 3 and re.fullmatch(r"[a-z0-9_]+", segs[2]):
                    sig = signals.get(segs[2])
                    if sig is None:
                        find(f"{repo}/{rel}", n, "topic", f"signal '{segs[2]}' is not in contracts signals.json")
                    elif sig.get("source") not in sources[root]:
                        find(f"{repo}/{rel}", n, "topic",
                             f"signal '{segs[2]}' has source '{sig.get('source')}', not a {root} source")


# ── document references ──────────────────────────────────────────────────────

LINK = re.compile(r"\[[^\]]*\]\(([^)\s]+)\)")


def check_docrefs(ws: Workspace) -> None:
    names = "|".join(map(re.escape, ws.repos))
    xref = re.compile(rf"\b({names})\b[^`\n]{{0,3}}`([^`\s]+\.md)(?:#[^`]*)?`")
    basenames = {r: {Path(f).name for f in ws.files[r]} for r in ws.repos}
    for repo, rel, text in ws.each({".md"}):
        for n, line in lines_outside_fences(text):
            for m in xref.finditer(line):
                target, path = m.group(1), m.group(2)
                ok = path in ws.files[target] if "/" in path else path in basenames[target]
                if not ok:
                    find(f"{repo}/{rel}", n, "docref", f"{target} has no {path}")
            for m in LINK.finditer(line):
                href = m.group(1)
                if re.match(r"^[a-z][a-z0-9+.-]*:", href) or href.startswith("#"):
                    continue  # a URL (http:, gitea:, mailto:) or an anchor on this page
                path = href.split("#", 1)[0]
                resolved = os.path.normpath(os.path.join(os.path.dirname(rel), path))
                if resolved.startswith(".."):
                    continue  # leaves the repo: not checkable from here
                if not (ws.root / repo / resolved).exists():
                    find(f"{repo}/{rel}", n, "docref", f"link target {href} does not exist")


# ── binaries and services ────────────────────────────────────────────────────


def check_binaries(ws: Workspace) -> None:
    repos_md = ("project-yukart", "docs/REPOS.md")
    runtime = ws.text("project-yukart", "docs/ORIN-RUNTIME.md")
    rows = {}
    for n, line in enumerate(ws.text(*repos_md).splitlines(), 1):
        m = re.match(r"^\|\s*\[([\w-]+)\]", line)
        if m:
            rows[m.group(1)] = (n, line)
    for repo in ws.repos:
        if "Cargo.toml" in ws.files[repo]:
            bins = re.findall(r'\[\[bin\]\]\s*\nname\s*=\s*"([^"]+)"', ws.text(repo, "Cargo.toml"))
            if bins and repo not in rows:
                find("/".join(repos_md), 1, "binaries", f"no row for {repo}")
            elif bins:
                n, line = rows[repo]
                # Only what is installed somewhere: a test tool needs no place in the overview.
                deployed = [b for b in bins if f"deploy/yukart-{b}.service" in ws.files[repo]]
                missing = [b for b in deployed if f"`{b}`" not in line]
                if missing:
                    find("/".join(repos_md), n, "binaries",
                         f"{repo} row does not name {', '.join(missing)} (from its Cargo.toml)")
        for rel in ws.files[repo]:
            if rel.startswith("deploy/") and rel.endswith(".service"):
                unit = Path(rel).stem
                service = unit.removeprefix("yukart-")
                if service not in runtime and repo not in runtime:
                    find("project-yukart/docs/ORIN-RUNTIME.md", 1, "binaries",
                         f"{repo} deploys {unit}, which the startup tiers do not mention")


# ── front matter ─────────────────────────────────────────────────────────────


def check_frontmatter(ws: Workspace) -> None:
    for repo, rel, text in ws.each({".md"}):
        if not rel.startswith("docs/"):
            continue
        m = re.match(r"---\n(.*?)\n---\n", text, re.S)
        if not m:
            find(f"{repo}/{rel}", 1, "frontmatter", "no front matter")
            continue
        t = re.search(r"^type:\s*(\S+)", m.group(1), re.M)
        if not t:
            find(f"{repo}/{rel}", 1, "frontmatter", "front matter has no type")
        elif t.group(1) not in DOC_TYPES:
            find(f"{repo}/{rel}", 1, "frontmatter", f"type '{t.group(1)}' is not one of {sorted(DOC_TYPES)}")


def main() -> int:
    root = Path(sys.argv[1] if len(sys.argv) > 1 else ".").resolve()
    ws = Workspace(root)
    missing = {"project-yukart", "contracts"} - set(ws.repos)
    if missing:
        print(f"crosscheck: {root} lacks {', '.join(sorted(missing))}; nothing to check against", file=sys.stderr)
        return 2
    for check in (check_decisions, check_topics, check_docrefs, check_binaries, check_frontmatter):
        check(ws)
    for f in findings:
        print(f)
    print(f"crosscheck: {len(ws.repos)} repos, {len(findings)} finding(s)", file=sys.stderr)
    return 1 if findings else 0


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