settled-reach/tooling/domains/check/service.py

"""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
from tooling.domains.check.schemas import (
    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,
    )


# --- 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