mikkeli
2401663b4b
Decisions, topics, doc references, deployed binaries and front matter, each checked against the repo that owns the fact. A reusable workflow runs it from every repo on push. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz
265 lines
12 KiB
Python
265 lines
12 KiB
Python
#!/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-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(".")
|
|
if root in ws.repos 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())
|