rune/henxels_checks.py

from pathlib import Path

from henxels import statement


@statement("ci_runs_all_canaries", help="every src/rune/**/canary_*.py is listed in .gitea/workflows/ci.yml")
def ci_runs_all_canaries(param, file, root):
    if not (file.startswith("src/rune/") and Path(file).name.startswith("canary_")):
        return None
    if not file.endswith(".py"):
        return None
    ci = root / ".gitea" / "workflows" / "ci.yml"
    if not ci.exists():
        return f"{file} is missing from .gitea/workflows/ci.yml (no ci.yml found)"
    if Path(file).name in ci.read_text():
        return None
    return f"{file} is missing from .gitea/workflows/ci.yml — register it in the canary job's for-loop list"


# --- Umbrella impact-map contracts (T32): the map is the change-propagation
# graph. It must be complete (bidirectional), reference real canaries, and
# stay acyclic — otherwise the impact gate lies.

def _umbrella_manifest(root):
    import json
    p = root / "config" / "umbrellas.json"
    if not p.exists():
        return None
    return json.loads(p.read_text()).get("umbrellas", {})


@statement(
    "umbrella_manifest_is_complete",
    help="every src/rune/*.py and shared/*.py module is in exactly one umbrella; every entry references a real file (bidirectional)",
)
def umbrella_manifest_is_complete(param, file, root):
    manifest = _umbrella_manifest(root)
    if manifest is None:
        return "config/umbrellas.json is missing — the impact gate has no map"
    mapped = {}
    for umb, data in manifest.items():
        for mod in data.get("modules", []):
            mapped.setdefault(mod, []).append(umb)
    dupes = {m: u for m, u in mapped.items() if len(u) > 1}
    if dupes:
        return f"modules in multiple umbrellas (map must be a partition): {dupes}"
    # 'unmapped' preserves the historical top-level contract; 'phantom' checks
    # existence across the whole tree (a mapped name is phantom only if NO
    # file with that stem exists anywhere under src/rune or shared).
    actual = {f.stem for f in (root / "src" / "rune").glob("*.py")
              if not f.stem.startswith(("canary_", "test_", "__"))}
    actual |= {f.stem for f in (root / "shared").glob("*.py")}
    existing = {f.stem for f in (root / "src" / "rune").rglob("*.py")}
    existing |= {f.stem for f in (root / "shared").rglob("*.py")}
    unmapped = sorted(actual - set(mapped))
    phantom = sorted(set(mapped) - existing)
    debt = root / "docs" / "module-coverage-debt.md"
    debt_ok = debt.exists()
    if unmapped and not debt_ok:
        return f"unmapped modules (add to config/umbrellas.json or docs/module-coverage-debt.md): {unmapped}"
    if phantom:
        return f"phantom manifest entries (files gone — remove from the map): {phantom}"
    return None


@statement(
    "umbrella_canaries_exist",
    help="every canary referenced by the umbrella manifest exists in src/rune/ (CI registration enforced by ci_runs_all_canaries)",
)
def umbrella_canaries_exist(param, file, root):
    manifest = _umbrella_manifest(root)
    if manifest is None:
        return None
    phantom = []
    for umb, data in manifest.items():
        for c in data.get("canaries", []):
            if not (root / "src" / "rune" / f"{c}.py").exists():
                phantom.append(f"{umb}:{c}")
    if phantom:
        return f"manifest references canaries that do not exist: {phantom}"
    return None


@statement(
    "umbrella_dag_acyclic",
    help="the umbrella depends_on graph is acyclic with known targets — reflexive updates cannot cycle",
)
def umbrella_dag_acyclic(param, file, root):
    manifest = _umbrella_manifest(root)
    if manifest is None:
        return None
    def has_cycle(node, seen, stack):
        if node in stack:
            return True
        if node in seen:
            return False
        seen.add(node)
        return any(has_cycle(d, seen, stack | {node})
                   for d in manifest.get(node, {}).get("depends_on", []))
    for u in manifest:
        if has_cycle(u, set(), frozenset()):
            return f"dependency cycle through umbrella '{u}' — reflexive updates would recurse forever"
        for d in manifest[u].get("depends_on", []):
            if d not in manifest:
                return f"umbrella '{u}' depends on unknown umbrella '{d}'"
    return None


# --- Pipeline-expansion contracts (issue #204 doctrine): when a new process,
# feature, workflow, or backend surface is added, a governing contract must
# exist for it. These statements make "ungoverned surface" a check failure.

@statement(
    "workflows_are_governed",
    help="every .gitea/workflows/*.yml is referenced by docs/pipeline-registry.md — new pipeline machinery ships with documentation",
)
def workflows_are_governed(param, file, root):
    if not file.startswith(".gitea/workflows/") or not file.endswith(".yml"):
        return None
    registry = root / "docs" / "pipeline-registry.md"
    if not registry.exists():
        return f"docs/pipeline-registry.md is missing — create it and register {file} (every workflow documents trigger/purpose/owner)"
    if Path(file).name not in registry.read_text():
        return f"{file} is not registered in docs/pipeline-registry.md — new pipeline processes ship with a registry entry (trigger, purpose, failure mode)"
    return None


@statement(
    "tui_panels_are_governed",
    help="new TUI surfaces (src/rune/*tui*.py panels/widgets) are referenced in docs/tui-registry.md — no silent UI growth",
)
def tui_panels_are_governed(param, file, root):
    if Path(file).stem.startswith("canary_"):
        return None
    if "tui" not in Path(file).name.lower():
        return None
    registry = root / "docs" / "tui-registry.md"
    if not registry.exists():
        return f"docs/tui-registry.md is missing — create it and register {file} (every TUI surface documents what it renders and who owns it)"
    if Path(file).name in registry.read_text():
        return None
    # registry exists but this file is new/unregistered — check for a known-exempt core list
    core = {"rune_tui.py"}
    if Path(file).name in core:
        return None
    return f"{Path(file).name} is not registered in docs/tui-registry.md — new TUI surfaces ship with a registry entry"


def _tests_corpus(root) -> str:
    tests_dir = root / "tests"
    if not tests_dir.exists():
        return ""
    parts = []
    for tf in sorted(tests_dir.glob("test_*.py")):
        try:
            parts.append(tf.read_text(errors="replace"))
        except OSError:
            continue
    return "\n".join(parts)


@statement(
    "new_rune_modules_have_canary_or_test_coverage",
    help="every src/rune/*.py module is referenced by at least one test file or has a canary — no ungoverned code surfaces",
)
def new_rune_modules_have_canary_or_test_coverage(param, file, root):
    if not file.startswith("src/rune/") or not file.endswith(".py"):
        return None
    name = Path(file).stem
    if name.startswith(("canary_", "test_", "__")):
        return None
    tests_dir = root / "tests"
    if (tests_dir / f"test_{name}.py").exists():
        return None
    canary = root / "src" / "rune" / f"canary_{name}.py"
    if canary.exists():
        return None
    corpus = _tests_corpus(root)
    if name in corpus:
        return None
    debt = root / "docs" / "module-coverage-debt.md"
    if debt.exists():
        debt_text = debt.read_text(errors="replace")
        if name in debt_text:
            return None  # tracked debt — visible in docs/module-coverage-debt.md (list only shrinks)
    return (
        f"{name} is referenced by no test file and has no canary — new modules ship with coverage "
        f"(a test importing/exercising it, or a canary) per docs/bugfix-contract.md; "
        f"legacy debt goes in docs/module-coverage-debt.md (tracked, not silent)"
    )