"""Logic for the `check` domain. Transport-agnostic (D-263). Nothing here prints, calls `sys.exit`, or imports typer. A service must not know it was called from a CLI — that is what lets a test call it directly, lets one domain's service call another's, and leaves a second front end possible without a rewrite. """ from __future__ import annotations import re import sqlite3 from pathlib import Path from tooling.core import config, process from tooling.domains.check.schemas import ( CanvasCheck, CanvasState, DiagramCheck, FactIdCheck, StampCheck, StampState, UnknownFact, VersionCheck, ) # Anchored to line start so the commentary above `version:` (which mentions # earlier versions by number) can never be mistaken for the field itself. _YAML_VERSION = re.compile(r"^version:\s*(\S+)\s*$", re.MULTILINE) _GODOT_VERSION = re.compile(r'^config/version\s*=\s*"([^"]*)"\s*$', re.MULTILINE) def client_version() -> VersionCheck: """Compare the version in project.yaml with the one baked into the client. project.yaml is the version source of truth (CLAUDE.md). The client cannot read it at runtime — an exported build has no repo root — so the value is mirrored into `application/config/version` in client/project.godot, which Godot bakes into the PCK (T-1241). A mirror nobody checks is worse than the bug it replaced: the old code failed LOUDLY in an export ("?.?.?" everywhere), whereas a stale mirror fails SILENTLY — the Atlas disk cache keeps serving canvases under a version that stopped matching the build. That is the T-1239 failure, which cost eight days of a map drawn from a canvas whose generating code no longer existed. """ root = config.repo_root() yaml_version, problem = _read(root / "project.yaml", _YAML_VERSION, "`version:` line") if problem: return VersionCheck(ok=False, problem=problem) godot_path = root / "client" / "project.godot" godot_version, problem = _read(godot_path, _GODOT_VERSION, "`config/version=` line") if problem: return VersionCheck(ok=False, yaml_version=yaml_version, problem=problem) return VersionCheck( ok=yaml_version == godot_version, yaml_version=yaml_version, godot_version=godot_version, ) # --- canvas-generation version pairing ------------------------------------ # # project.yaml's `version:` is the Atlas disk cache's ONLY invalidation signal. # Change how a canvas is generated without moving it and every warm cache keeps # serving canvases built by code that no longer exists — silently, and only on # machines with a warm cache, so the author never sees it. That has shipped five # times; T-1239 is what the last one cost (eight days). # # Deliberately no override flag (T-1242): a false positive costs one round of # cache misses, a false negative costs another week of a wrong map. An escape # hatch would be reached for exactly when someone is sure their change is # harmless — which is the state of mind that produced all five regressions. DEFAULT_BASE = "origin/main" def diff_has_version_bump(diff_text: str) -> bool: """Does this project.yaml diff actually MOVE the `version:` field? Kept pure so the property is testable without constructing git history — which is what tooling/test_canvas_version_check.py exercises, and the reason that test could be written at all. Matched on the diff BODY rather than on the filename appearing in --name-only: project.yaml carries a comment block documenting past bumps, including lines quoting old version NUMBERS, so editing that commentary must not count. Requires the ADDED side — a lone deletion means the field was removed, not moved — and skips `+++` headers, which would otherwise match. """ for line in diff_text.splitlines(): if line.startswith("+++"): continue if line.startswith("+version:"): return True return False def canvas_version(base: str = DEFAULT_BASE, head: str = "HEAD") -> CanvasCheck: """Require a version bump alongside any canvas-generation change.""" from tooling import canvas_sources # Three-dot: what HEAD added since the merge base, matching the convention # the pre-push hook already uses for the systems.db stamp check. commit_range = f"{base}...{head}" if _git("rev-parse", "--verify", base) is None: # No base to compare against (fresh clone, no remote). Skipping is # correct rather than failing: there is no range to judge. return CanvasCheck(state=CanvasState.NO_BASE, commit_range=commit_range, base=base) listing = _git("diff", "--name-only", commit_range) if listing is None: return CanvasCheck( state=CanvasState.DIFF_FAILED, commit_range=commit_range, base=base ) changed = {line for line in listing.splitlines() if line} touched = sorted(changed & set(canvas_sources.relative_paths())) if not touched: return CanvasCheck(state=CanvasState.CLEAN, commit_range=commit_range, base=base) bumped = _git("diff", "-U0", commit_range, "--", "project.yaml") state = ( CanvasState.BUMPED if bumped is not None and diff_has_version_bump(bumped) else CanvasState.NEEDS_BUMP ) return CanvasCheck( state=state, commit_range=commit_range, touched=touched, base=base ) def _git(*args: str) -> str | None: """Run git, returning stdout, or None if it failed. `check=False` on purpose: a git failure here is not an error to report, it is a signal that there is nothing to compare — a fresh clone with no remote is a legitimate state, not a broken one. Still goes through `core.process.run` so the argv-list and missing-binary guards apply. """ result = process.run( ["git", "-C", str(config.repo_root()), *args], check=False, ) return result.stdout if result.returncode == 0 else None # --- systems-db stamp ----------------------------------------------------- # # server/data/systems.db is a committed build artefact. Its meta table records # the SHA-1 of each generator's sources at generation time; if the sources have # moved on and the DB has not, a push would ship a snapshot nobody can reproduce # (T-855, T-857). _SEMVER = re.compile(r"^\d+\.\d+\.\d+$") def systems_db_stamp() -> StampCheck: """Compare systems.db's recorded generator SHAs against the sources on disk.""" # Imported directly rather than through a sys.path insert — the old script # needed the hack because tooling/ was not a package. That is the E402 debt # T-1274 expects to evaporate, doing so. from tooling import generator_sources db_path = config.path("server", "data", "systems.db") if not db_path.exists(): return StampCheck(state=StampState.ABSENT) try: connection = sqlite3.connect(str(db_path)) rows = connection.execute( "SELECT generator_name, schema_version, generator_sha FROM meta" ).fetchall() connection.close() except sqlite3.OperationalError: return StampCheck(state=StampState.UNSTAMPED) if not rows: return StampCheck(state=StampState.UNSTAMPED) bad_version: list[str] = [] unknown: list[str] = [] stale: list[str] = [] versions: dict[str, str] = {} for name, schema_version, stored_sha in rows: versions[name] = schema_version # An old DB may still carry a 40-char SHA here rather than a semver. # Flagged rather than silently accepted, so the answer is "regen" and # not a pass that hides a schema from a different era (T-888). if not _SEMVER.match(schema_version or ""): bad_version.append( f"{name}: schema_version={schema_version!r} " "(expected semver like '1.0.0' — run make regen-db)" ) sources = generator_sources.GENERATOR_SOURCES.get(name) if sources is None: # Fail CLOSED. A branch that adds a generator without registering it # would otherwise pass this gate while checking nothing about it. unknown.append(name) continue try: current = generator_sources.file_sha1(*sources) except FileNotFoundError as exc: return StampCheck(state=StampState.BROKEN, details=[f"{name}: {exc}"]) if current != stored_sha: stale.append(name) if bad_version: return StampCheck(state=StampState.BAD_VERSION, details=bad_version) # Generators disagreeing on schema_version means the DB was partially # regenerated against different source trees — worse than stale, because # nothing about it is internally consistent. if len(set(versions.values())) > 1: return StampCheck( state=StampState.CONFLICT, details=[f"{gen}: {ver}" for gen, ver in sorted(versions.items())], ) if unknown: return StampCheck(state=StampState.UNKNOWN, details=sorted(unknown)) if stale: return StampCheck(state=StampState.STALE, details=sorted(stale)) return StampCheck(state=StampState.OK, generators=len(rows)) # --- dataflow-graph ------------------------------------------------------- # # A diagram that names files goes stale SILENTLY: nothing fails when a path # moves, so the map keeps asserting a layout that is no longer true. This closes # the cheap half of that gap (D-262). It CANNOT check whether an edge still # means what it says — if wiki_sync.py stopped writing body pages tomorrow every # path here would still exist. Edge semantics are a human job. # Hand-authored diagrams whose labels name real repo paths. A diagram absent # from this list is not checked; add one when it starts naming files. CHECKED_DIAGRAMS = ["data-flow/wiki-generator-flow.d2"] # Generated sources are skipped even if listed: their correctness is the # generator's problem (the source-canonical rule), and they name star-system ids # rather than repo paths. GENERATED_PREFIXES = ("design/star-map-",) _LABEL = re.compile(r'"((?:[^"\\]|\\.)*)"') _PATH_TOKEN = re.compile(r"[A-Za-z0-9_.\-*{}/]*/[A-Za-z0-9_.\-*{}/]+") def dataflow_graph() -> DiagramCheck: """Assert every repo path named in a hand-authored diagram still resolves.""" root = config.repo_root() top_level = {entry.name for entry in root.iterdir()} diagram_root = root / "docs" / "diagrams" unresolved: list[str] = [] missing: list[str] = [] checked = 0 considered = [d for d in CHECKED_DIAGRAMS if not d.startswith(GENERATED_PREFIXES)] for diagram in considered: path = diagram_root / diagram if not path.exists(): missing.append(f"{diagram}: diagram not found") continue for label in _LABEL.findall(path.read_text(encoding="utf-8")): for token in _path_tokens(label, top_level): checked += 1 if not _resolves(root, token): unresolved.append(f"{diagram}: path does not resolve: {token}") return DiagramCheck( diagrams=len(considered), checked=checked, unresolved=unresolved, missing=missing, ) def _path_tokens(label: str, top_level: set[str]) -> list[str]: # Labels use \n for line breaks and · to separate sibling files. flat = label.replace("\\n", " ").replace("·", " ") tokens = [] for raw in _PATH_TOKEN.findall(flat): token = raw.strip(".,;:") # A token only counts if its first segment is a real top-level entry. # Without this, legend prose like "dashed one-time or bootstrap" yields # the token one-time/bootstrap and fails the check on nothing. if token and token.split("/", 1)[0] in top_level: tokens.append(token) return tokens def _resolves(root: Path, token: str) -> bool: if (root / token).exists(): return True # bodies/{id}/index.md -> bodies/*/index.md, then glob it. pattern = re.sub(r"\{[^}]*\}", "*", token) if "*" in pattern: try: return any(root.glob(pattern)) except (ValueError, OSError): return False return False # Ported from the bash `check-fact-ids` (D-263: the shell scripts are rewritten, # not wrapped). Deliberately still LINE-MATCHED rather than YAML-parsed: the # original was grep-based, and parsing YAML properly would change which lines # count — anchors, merge keys and multi-document files would all start # contributing ids the old check never saw. That is a different check wearing # the same name, and this port is not the place to make it. _FACT_ID = re.compile(r"^\s*(?:-\s*)?fact_id:\s*(.+?)\s*$") # Its schema is attribute keys, not fact_ids — excluded by the original too. _NOT_A_CATALOG = "entity-attributes.yaml" def fact_ids() -> FactIdCheck: """Validate fact_id references in campaign content against the catalogs.""" root = config.repo_root() canonical = _collect(root / "server" / "content" / "global" / "knowledge", catalogs=True) referenced = _collect(root / "server" / "content" / "campaigns", catalogs=False) if not canonical: return FactIdCheck( advisory=True, canonical_count=0, referenced=sorted(referenced), ) unknown = [ UnknownFact(fact_id=ref, locations=_locate(root, ref)) for ref in sorted(referenced) if ref not in canonical ] return FactIdCheck( advisory=False, canonical_count=len(canonical), referenced=sorted(referenced), unknown=unknown, ) def _collect(directory: Path, *, catalogs: bool) -> set[str]: found: set[str] = set() if not directory.is_dir(): return found for path in directory.rglob("*.yaml"): if catalogs and path.name == _NOT_A_CATALOG: continue for line in path.read_text(encoding="utf-8", errors="replace").splitlines(): if line.lstrip().startswith("#"): continue match = _FACT_ID.match(line) if match: value = _clean(match.group(1)) if value: found.add(value) return found def _clean(value: str) -> str: """Strip a trailing comment and surrounding quotes, as the sed chain did.""" value = re.sub(r"\s+#.*$", "", value).strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in "\"'": value = value[1:-1] return value.strip() def _locate(root: Path, fact_id: str) -> list[str]: """Repo-relative file:line for every reference, so a failure is actionable.""" content = root / "server" / "content" / "campaigns" hits: list[str] = [] if not content.is_dir(): return hits for path in sorted(content.rglob("*.yaml")): for number, line in enumerate( path.read_text(encoding="utf-8", errors="replace").splitlines(), start=1 ): match = _FACT_ID.match(line) if match and _clean(match.group(1)) == fact_id: hits.append(f"{path.relative_to(root)}:{number}") return hits def _read(path: Path, pattern: re.Pattern[str], label: str) -> tuple[str | None, str | None]: """Return (value, problem). Exactly one of the two is ever set.""" if not path.exists(): return None, f"{path} not found" match = pattern.search(path.read_text(encoding="utf-8")) if not match: return None, f"no {label} in {path}" return match.group(1), None