"""Code-domain invariants engine for Muse.

Evaluates semantic rules against code snapshots.  Rules are declared in
``.muse/code_invariants.toml`` and evaluated at commit time, merge time, or
on-demand via ``muse code-check``.

Rule types
----------

``max_complexity``
    Detects functions / methods whose estimated cyclomatic complexity exceeds
    *threshold*.  Complexity is approximated by counting control-flow branch
    points (``if``, ``elif``, ``for``, ``while``, ``except``, ``with``,
    ``and``, ``or``) inside each symbol's body.  This correlates well with
    real cyclomatic complexity for Python and is language-agnostic for other
    tree-sitter-parsed languages.

``no_circular_imports``
    Detects import cycles among Python files in the snapshot.  Builds a
    directed graph (file → files it imports) and runs DFS cycle detection.
    Reports each cycle as one violation at the root file of the cycle.

``no_dead_exports``
    Detects top-level functions and classes that are never imported by any
    other file in the snapshot (dead exports / unreachable public API).
    Only applies to semantic files; test files and ``__init__.py`` are exempt.

``test_coverage_floor``
    Requires that at least *min_ratio* of non-test functions have a
    corresponding test function (detected by ``test_`` prefix convention).
    Reports the actual vs required coverage ratio when the floor is not met.

TOML example
------------
::

    [[rule]]
    name = "complexity_gate"
    severity = "error"
    scope = "function"
    rule_type = "max_complexity"
    [rule.params]
    threshold = 15

    [[rule]]
    name = "no_cycles"
    severity = "error"
    scope = "file"
    rule_type = "no_circular_imports"

    [[rule]]
    name = "dead_exports"
    severity = "warning"
    scope = "file"
    rule_type = "no_dead_exports"

    [[rule]]
    name = "test_coverage"
    severity = "warning"
    scope = "repo"
    rule_type = "test_coverage_floor"
    [rule.params]
    min_ratio = 0.30

Public API
----------
- :class:`CodeInvariantRule`    — code-specific rule declaration.
- :class:`CodeViolation`        — violation with file + symbol address.
- :class:`CodeInvariantReport`  — full report for one commit.
- :class:`CodeChecker`          — satisfies :class:`~muse.core.invariants.InvariantChecker`.
- :func:`load_invariant_rules`  — load from TOML with built-in defaults.
- :func:`run_invariants`        — top-level runner.
"""

from __future__ import annotations

import ast
import logging
import pathlib
from typing import Literal, TypedDict

import msgpack

from muse.core.invariants import (
    BaseReport,
    BaseViolation,
    InvariantSeverity,
    load_rules_toml,
    make_report,
)
from muse.core.object_store import read_object
from muse.core.store import (
    Manifest,
    get_commit_snapshot_manifest,
    read_msgpack_file,
)
from muse.core.validation import MAX_AST_BYTES

logger = logging.getLogger(__name__)

type ComplexityMap = dict[str, int]             # symbol_address → branch-count score
type FileDataMap = dict[str, _FileData]          # content_hash → parsed file data
type ImportGraph = dict[str, set[str]]           # file_path → set of imported file_paths
type ParamsDict = dict[str, str | int | float]   # rule params (dynamic keys)

_DEFAULT_RULES_FILE = ".muse/code_invariants.toml"
_FILE_CACHE_NAME = "code_invariants_cache.msgpack"
_FILE_CACHE_VERSION = 1

# ---------------------------------------------------------------------------
# Per-file AST cache — two-level: intra-run (in-memory) + cross-run (msgpack)
# ---------------------------------------------------------------------------


class _FileData(TypedDict):
    """All AST-derived data for one Python file, derived from a single parse.

    Keyed by file content hash in :class:`_InvariantFileCache` so a file that
    has not changed between ``muse code invariants`` runs costs zero parse time.

    ``raw_module_imports``
        Module names from ``import X`` statements.  e.g. ``"import os"`` → ``["os"]``.
    ``from_module`` / ``from_name``
        Parallel lists for ``from M import N`` statements.  Same index = one import pair.
    ``top_level_fns``
        Public (non-underscore, non-``main``) top-level function names.
    ``top_level_classes``
        Public top-level class names.
    ``has_all``
        ``True`` when the file declares ``__all__``.
    ``complexity``
        ``{symbol_address: score}`` — branch-count cyclomatic complexity proxy.
    """

    raw_module_imports: list[str]
    from_module: list[str]
    from_name: list[str]
    top_level_fns: list[str]
    top_level_classes: list[str]
    has_all: bool
    complexity: ComplexityMap


class _InvariantCacheDoc(TypedDict):
    """On-disk document shape for the invariant file cache (msgpack)."""

    version: int
    entries: FileDataMap


def _empty_file_data() -> _FileData:
    return _FileData(
        raw_module_imports=[],
        from_module=[],
        from_name=[],
        top_level_fns=[],
        top_level_classes=[],
        has_all=False,
        complexity={},
    )


def _parse_file_info(source: bytes, file_path: str) -> _FileData:
    """Parse one Python file and extract everything all rules need.

    A single ``ast.parse`` call populates all fields.  Non-Python files and
    files that fail to parse return an empty :class:`_FileData`.
    """
    if not file_path.endswith((".py", ".pyi")):
        return _empty_file_data()
    if len(source) > MAX_AST_BYTES:
        return _empty_file_data()
    try:
        tree = ast.parse(source, filename=file_path)
    except SyntaxError:
        return _empty_file_data()

    branch_nodes = (
        ast.If, ast.For, ast.While, ast.ExceptHandler,
        ast.With, ast.AsyncWith, ast.AsyncFor,
        ast.BoolOp, ast.Assert, ast.comprehension,
    )

    raw_module_imports: list[str] = []
    from_module: list[str] = []
    from_name: list[str] = []
    top_level_fns: list[str] = []
    top_level_classes: list[str] = []
    has_all = False
    complexity: ComplexityMap = {}

    def _score(node: ast.FunctionDef | ast.AsyncFunctionDef, prefix: str) -> None:
        qualified = f"{prefix}{node.name}" if prefix else node.name
        addr = f"{file_path}::{qualified}"
        score = 1
        for child in ast.walk(node):
            if isinstance(child, branch_nodes):
                score += 1
        complexity[addr] = score

    # Single walk: collect imports, complexity, top-level names.
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                raw_module_imports.append(alias.name)
        elif isinstance(node, ast.ImportFrom) and node.module:
            for alias in node.names:
                from_module.append(node.module)
                from_name.append(alias.name)
        elif isinstance(node, ast.ClassDef):
            for item in ast.walk(node):
                if isinstance(item, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    _score(item, f"{node.name}.")
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            _score(node, "")

    # Top-level names only (not recursive — iter_child_nodes).
    for node in ast.iter_child_nodes(tree):
        if isinstance(node, ast.Assign):
            if any(
                isinstance(t, ast.Name) and t.id == "__all__"
                for t in node.targets
            ):
                has_all = True
        elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            if not node.name.startswith("_") and node.name != "main":
                top_level_fns.append(node.name)
        elif isinstance(node, ast.ClassDef):
            if not node.name.startswith("_"):
                top_level_classes.append(node.name)

    return _FileData(
        raw_module_imports=raw_module_imports,
        from_module=from_module,
        from_name=from_name,
        top_level_fns=top_level_fns,
        top_level_classes=top_level_classes,
        has_all=has_all,
        complexity=complexity,
    )


class _InvariantFileCache:
    """Persistent per-file AST analysis cache keyed by content hash.

    Stored as msgpack at ``.muse/code_invariants_cache.msgpack``.  Follows
    the same atomic-write and graceful-degradation pattern as
    :class:`~muse.core.symbol_cache.SymbolCache`.

    On a warm cache (unchanged snapshot), ``muse code invariants`` skips
    all ``ast.parse`` calls — reducing parse time from O(N×R) to O(1).
    """

    def __init__(self, muse_dir: pathlib.Path | None, data: FileDataMap) -> None:
        self._muse_dir = muse_dir
        self._data = data
        self._dirty = False

    @classmethod
    def load(cls, repo_root: pathlib.Path) -> _InvariantFileCache:
        """Load from ``repo_root/.muse/code_invariants_cache.msgpack``.

        Returns an empty cache on any failure — never raises.
        """
        muse_dir = repo_root / ".muse"
        if not muse_dir.is_dir():
            return cls(None, {})
        cache_file = muse_dir / _FILE_CACHE_NAME
        if not cache_file.is_file():
            return cls(muse_dir, {})
        try:
            doc = read_msgpack_file(cache_file)
            if not isinstance(doc, dict) or doc.get("version") != _FILE_CACHE_VERSION:
                logger.debug("⚠️ invariant file cache version mismatch — starting fresh")
                return cls(muse_dir, {})
            raw_entries = doc.get("entries", {})
            if not isinstance(raw_entries, dict):
                return cls(muse_dir, {})
            data: FileDataMap = {}
            for content_hash, v in raw_entries.items():
                if not isinstance(content_hash, str) or not isinstance(v, dict):
                    continue
                raw_mi = v.get("raw_module_imports")
                raw_fm = v.get("from_module")
                raw_fn = v.get("from_name")
                raw_tf = v.get("top_level_fns")
                raw_tc = v.get("top_level_classes")
                raw_cx = v.get("complexity")
                data[content_hash] = _FileData(
                    raw_module_imports=[str(x) for x in raw_mi] if isinstance(raw_mi, list) else [],
                    from_module=[str(x) for x in raw_fm] if isinstance(raw_fm, list) else [],
                    from_name=[str(x) for x in raw_fn] if isinstance(raw_fn, list) else [],
                    top_level_fns=[str(x) for x in raw_tf] if isinstance(raw_tf, list) else [],
                    top_level_classes=[str(x) for x in raw_tc] if isinstance(raw_tc, list) else [],
                    has_all=bool(v.get("has_all", False)),
                    complexity={
                        str(k): val
                        for k, val in raw_cx.items()
                        if isinstance(val, int)
                    } if isinstance(raw_cx, dict) else {},
                )
            logger.debug("✅ invariant file cache loaded (%d entries)", len(data))
            return cls(muse_dir, data)
        except Exception as exc:
            logger.debug("⚠️ invariant file cache unreadable (%s) — starting fresh", exc)
            return cls(muse_dir, {})

    @classmethod
    def empty(cls) -> _InvariantFileCache:
        """Return an in-memory-only cache (no persistence)."""
        return cls(None, {})

    def get(self, content_hash: str) -> _FileData | None:
        return self._data.get(content_hash)

    def put(self, content_hash: str, data: _FileData) -> None:
        self._data[content_hash] = data
        self._dirty = True

    def save(self) -> None:
        """Atomically persist if dirty.  Silently skips when there is no ``.muse`` dir."""
        if not self._dirty or self._muse_dir is None:
            return
        cache_file = self._muse_dir / _FILE_CACHE_NAME
        tmp = self._muse_dir / (_FILE_CACHE_NAME + ".tmp")
        doc = _InvariantCacheDoc(
            version=_FILE_CACHE_VERSION,
            entries=dict(self._data),
        )
        try:
            tmp.write_bytes(msgpack.packb(doc, use_bin_type=True))
            tmp.replace(cache_file)
            self._dirty = False
            logger.debug("✅ invariant file cache saved (%d entries)", len(self._data))
        except OSError as exc:
            logger.warning("⚠️ invariant file cache save failed: %s", exc)


def _build_file_data(
    manifest: Manifest,
    repo_root: pathlib.Path,
    cache: _InvariantFileCache,
) -> FileDataMap:
    """Return :class:`_FileData` for every Python file in *manifest*.

    Serves from *cache* on a hit; reads + parses the file on a miss, then
    stores the result back into the cache (which is later persisted).

    This is the entry point for both intra-run sharing (all rules receive
    the same dict) and cross-run caching (unchanged files are never re-parsed).
    """
    result: FileDataMap = {}
    for file_path, content_hash in manifest.items():
        if not file_path.endswith((".py", ".pyi")):
            continue
        cached = cache.get(content_hash)
        if cached is not None:
            result[file_path] = cached
            continue
        source = read_object(repo_root, content_hash)
        if source is None:
            continue
        info = _parse_file_info(source, file_path)
        cache.put(content_hash, info)
        result[file_path] = info
    return result


# ---------------------------------------------------------------------------
# Types
# ---------------------------------------------------------------------------


class _RuleRequired(TypedDict):
    name: str
    severity: InvariantSeverity
    scope: Literal["function", "file", "repo", "global"]
    rule_type: str


class CodeInvariantRule(_RuleRequired, total=False):
    """A single code invariant rule declaration.

    ``name``      Unique human-readable identifier.
    ``severity``  ``"info"``, ``"warning"``, or ``"error"``.
    ``scope``     Granularity: ``"function"``, ``"file"``, ``"repo"``.
    ``rule_type`` Built-in type: ``"max_complexity"``,
                  ``"no_circular_imports"``, ``"no_dead_exports"``,
                  ``"test_coverage_floor"``.
    ``params``    Rule-specific numeric / string parameters.
    """

    params: ParamsDict


class CodeViolation(TypedDict):
    """A code invariant violation with precise source location.

    ``rule_name``  Rule that fired.
    ``severity``   Inherited from the rule.
    ``address``    ``"file.py::symbol_name"`` or ``"file.py"`` for file-level.
    ``description`` Human-readable explanation.
    ``file``       Workspace-relative file path.
    ``symbol``     Symbol name (empty string for file-level violations).
    ``detail``     Additional context (e.g. complexity score, cycle path).
    """

    rule_name: str
    severity: InvariantSeverity
    address: str
    description: str
    file: str
    symbol: str
    detail: str


# ---------------------------------------------------------------------------
# Built-in default rules
# ---------------------------------------------------------------------------

_BUILTIN_DEFAULTS: list[CodeInvariantRule] = [
    CodeInvariantRule(
        name="complexity_gate",
        severity="warning",
        scope="function",
        rule_type="max_complexity",
        params={"threshold": 10},
    ),
    CodeInvariantRule(
        name="no_cycles",
        severity="error",
        scope="file",
        rule_type="no_circular_imports",
        params={},
    ),
    CodeInvariantRule(
        name="dead_exports",
        severity="warning",
        scope="file",
        rule_type="no_dead_exports",
        params={},
    ),
]


# ---------------------------------------------------------------------------
# Rule implementations
# ---------------------------------------------------------------------------


def _estimate_complexity(source: bytes, file_path: str) -> ComplexityMap:
    """Return {symbol_address: complexity_score} for a Python source file.

    Uses a simple branch-count heuristic — same logic as :func:`_parse_file_info`.
    Prefer passing ``file_data`` to :func:`check_max_complexity` instead of
    calling this directly, so the parse cost is shared across all rules.
    """
    return _parse_file_info(source, file_path)["complexity"]


def check_max_complexity(
    manifest: Manifest,
    repo_root: pathlib.Path,
    rule_name: str,
    severity: InvariantSeverity,
    *,
    threshold: int = 10,
    file_data: FileDataMap | None = None,
) -> list[CodeViolation]:
    """Detect functions whose estimated cyclomatic complexity exceeds *threshold*.

    When *file_data* is supplied (built by :func:`_build_file_data`), no I/O
    or parsing is performed — complexity scores come directly from the cache.
    """
    violations: list[CodeViolation] = []

    if file_data is not None:
        # Fast path: scores already computed, no I/O needed.
        for file_path, fd in sorted(file_data.items()):
            for addr, score in sorted(fd["complexity"].items()):
                if score > threshold:
                    symbol = addr.split("::", 1)[-1] if "::" in addr else ""
                    violations.append(CodeViolation(
                        rule_name=rule_name,
                        severity=severity,
                        address=addr,
                        description=(
                            f"Complexity {score} exceeds threshold {threshold}. "
                            "Consider extracting helper functions."
                        ),
                        file=file_path,
                        symbol=symbol,
                        detail=f"score={score} threshold={threshold}",
                    ))
        return violations

    # Slow path: read and parse each file individually.
    for file_path, content_hash in manifest.items():
        if not file_path.endswith((".py", ".pyi")):
            continue
        source = read_object(repo_root, content_hash)
        if source is None:
            continue
        scores = _estimate_complexity(source, file_path)
        for addr, score in sorted(scores.items()):
            if score > threshold:
                symbol = addr.split("::", 1)[-1] if "::" in addr else ""
                violations.append(CodeViolation(
                    rule_name=rule_name,
                    severity=severity,
                    address=addr,
                    description=(
                        f"Complexity {score} exceeds threshold {threshold}. "
                        "Consider extracting helper functions."
                    ),
                    file=file_path,
                    symbol=symbol,
                    detail=f"score={score} threshold={threshold}",
                ))
    return violations


def _build_module_index(manifest: Manifest) -> Manifest:
    """Return a module-name → file-path index for intra-repo import resolution."""
    module_to_file: Manifest = {}
    for fp in manifest:
        if fp.endswith((".py", ".pyi")):
            mod = fp.removesuffix(".pyi").removesuffix(".py").replace("/", ".").replace("\\", ".")
            module_to_file[mod] = fp
            module_to_file.setdefault(mod.rsplit(".", 1)[-1], fp)
    return module_to_file


def _build_import_graph(
    manifest: Manifest,
    repo_root: pathlib.Path,
    *,
    file_data: FileDataMap | None = None,
) -> ImportGraph:
    """Build a directed import graph: {file → set of imported files}.

    Only tracks intra-repo imports (files that exist in the manifest).

    When *file_data* is supplied (built by :func:`_build_file_data`), no I/O
    or ``ast.parse`` calls are made — imports come directly from the cache.
    """
    module_to_file = _build_module_index(manifest)
    graph: ImportGraph = {fp: set() for fp in manifest if fp.endswith(".py")}

    if file_data is not None:
        # Fast path: resolve imports from pre-parsed _FileData — zero I/O.
        for file_path, fd in file_data.items():
            if not file_path.endswith(".py"):
                continue
            for mod_name in fd["raw_module_imports"]:
                t = module_to_file.get(mod_name)
                if t and t != file_path:
                    graph[file_path].add(t)
            for module, name in zip(fd["from_module"], fd["from_name"]):
                t = module_to_file.get(module)
                if t and t != file_path:
                    graph[file_path].add(t)
                t2 = module_to_file.get(f"{module}.{name}")
                if t2 and t2 != file_path:
                    graph[file_path].add(t2)
        return graph

    # Slow path: read and parse each file individually.
    for file_path, content_hash in manifest.items():
        if not file_path.endswith(".py"):
            continue
        source = read_object(repo_root, content_hash)
        if source is None:
            continue
        if len(source) > MAX_AST_BYTES:
            continue
        if len(source) > MAX_AST_BYTES:
            continue
        try:
            tree = ast.parse(source, filename=file_path)
        except SyntaxError:
            continue

        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    target = module_to_file.get(alias.name)
                    if target and target != file_path:
                        graph[file_path].add(target)
            elif isinstance(node, ast.ImportFrom) and node.module:
                # Direct module match: `from foo.bar import baz` where foo.bar is a file.
                target = module_to_file.get(node.module)
                if target and target != file_path:
                    graph[file_path].add(target)
                # Submodule match: `from pkg import submod` where pkg.submod is a file.
                # This covers both `from src import b` (→ src.b → src/b.py) and
                # `from src.cli import app` (→ src.cli.app → src/cli/app.py).
                for alias in node.names:
                    sub = f"{node.module}.{alias.name}"
                    sub_target = module_to_file.get(sub)
                    if sub_target and sub_target != file_path:
                        graph[file_path].add(sub_target)

    return graph


def _find_cycles(graph: ImportGraph) -> list[list[str]]:
    """DFS cycle detection; returns list of cycles as file-path lists."""
    WHITE, GRAY, BLACK = 0, 1, 2
    color = {n: WHITE for n in graph}
    stack: list[str] = []
    cycles: list[list[str]] = []

    def dfs(node: str) -> None:
        color[node] = GRAY
        stack.append(node)
        for neighbor in sorted(graph.get(node, set())):
            if color.get(neighbor, BLACK) == WHITE:
                dfs(neighbor)
            elif color.get(neighbor, BLACK) == GRAY:
                # Found a cycle — extract from stack.
                idx = stack.index(neighbor)
                cycle = stack[idx:]
                # Deduplicate: only add if not already seen.
                cycle_key = frozenset(cycle)
                if not any(frozenset(c) == cycle_key for c in cycles):
                    cycles.append(list(cycle))
        stack.pop()
        color[node] = BLACK

    for node in sorted(graph):
        if color[node] == WHITE:
            dfs(node)

    return cycles


def check_no_circular_imports(
    manifest: Manifest,
    repo_root: pathlib.Path,
    rule_name: str,
    severity: InvariantSeverity,
    *,
    import_graph: ImportGraph | None = None,
) -> list[CodeViolation]:
    """Detect import cycles among Python files in the snapshot."""
    graph = import_graph if import_graph is not None else _build_import_graph(manifest, repo_root)
    cycles = _find_cycles(graph)
    violations: list[CodeViolation] = []
    for cycle in cycles:
        root_file = cycle[0]
        cycle_str = " → ".join([*cycle, cycle[0]])
        violations.append(CodeViolation(
            rule_name=rule_name,
            severity=severity,
            address=root_file,
            description=f"Circular import cycle detected: {cycle_str}",
            file=root_file,
            symbol="",
            detail=cycle_str,
        ))
    return violations


def check_no_dead_exports(
    manifest: Manifest,
    repo_root: pathlib.Path,
    rule_name: str,
    severity: InvariantSeverity,
    *,
    file_data: FileDataMap | None = None,
) -> list[CodeViolation]:
    """Detect top-level functions/classes never imported by any other file.

    Exempt: test files, ``__init__.py``, files with ``__all__`` declarations
    (which signal deliberate public API), and ``main`` functions.

    When *file_data* is supplied (built by :func:`_build_file_data`), no I/O
    or ``ast.parse`` calls are made.
    """
    violations: list[CodeViolation] = []

    if file_data is not None:
        # Fast path: derive imported names and top-level symbols from _FileData.
        imported_names: set[str] = set()
        for fd in file_data.values():
            imported_names.update(fd["from_name"])
            for mod in fd["raw_module_imports"]:
                imported_names.add(mod.split(".")[-1])

        for file_path, fd in file_data.items():
            if not file_path.endswith(".py"):
                continue
            base = pathlib.PurePosixPath(file_path).name
            if base.startswith("test_") or base == "__init__.py":
                continue
            if fd["has_all"]:
                continue
            for fn_name in fd["top_level_fns"]:
                if fn_name not in imported_names:
                    addr = f"{file_path}::{fn_name}"
                    violations.append(CodeViolation(
                        rule_name=rule_name, severity=severity, address=addr,
                        description=(
                            f"'{fn_name}' is never imported by any other file. "
                            "Consider removing, making private (prefix _), or adding to __all__."
                        ),
                        file=file_path, symbol=fn_name, detail="no importers found",
                    ))
            for cls_name in fd["top_level_classes"]:
                if cls_name not in imported_names:
                    addr = f"{file_path}::{cls_name}"
                    violations.append(CodeViolation(
                        rule_name=rule_name, severity=severity, address=addr,
                        description=(
                            f"Class '{cls_name}' is never imported by any other file. "
                            "Consider removing or making private."
                        ),
                        file=file_path, symbol=cls_name, detail="no importers found",
                    ))
        return violations

    # Slow path: read and parse each file individually.
    slow_imported: set[str] = set()
    for file_path, content_hash in manifest.items():
        if not file_path.endswith(".py"):
            continue
        source = read_object(repo_root, content_hash)
        if source is None:
            continue
        if len(source) > MAX_AST_BYTES:
            continue
        try:
            tree = ast.parse(source, filename=file_path)
        except SyntaxError:
            continue
        for node in ast.walk(tree):
            if isinstance(node, ast.Import):
                for alias in node.names:
                    slow_imported.add(alias.asname or alias.name.split(".")[-1])
            elif isinstance(node, ast.ImportFrom):
                for alias in node.names:
                    slow_imported.add(alias.asname or alias.name)

    for file_path, content_hash in manifest.items():
        if not file_path.endswith(".py"):
            continue
        base = pathlib.PurePosixPath(file_path).name
        if base.startswith("test_") or base == "__init__.py":
            continue
        source = read_object(repo_root, content_hash)
        if source is None:
            continue
        if len(source) > MAX_AST_BYTES:
            continue
        try:
            tree = ast.parse(source, filename=file_path)
        except SyntaxError:
            continue
        has_all = any(
            isinstance(n, ast.Assign)
            and any(isinstance(t, ast.Name) and t.id == "__all__" for t in n.targets)
            for n in ast.walk(tree)
        )
        if has_all:
            continue
        for node in ast.iter_child_nodes(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                if node.name.startswith("_") or node.name == "main":
                    continue
                if node.name not in slow_imported:
                    addr = f"{file_path}::{node.name}"
                    violations.append(CodeViolation(
                        rule_name=rule_name, severity=severity, address=addr,
                        description=(
                            f"'{node.name}' is never imported by any other file. "
                            "Consider removing, making private (prefix _), or adding to __all__."
                        ),
                        file=file_path, symbol=node.name, detail="no importers found",
                    ))
            elif isinstance(node, ast.ClassDef):
                if node.name.startswith("_"):
                    continue
                if node.name not in slow_imported:
                    addr = f"{file_path}::{node.name}"
                    violations.append(CodeViolation(
                        rule_name=rule_name, severity=severity, address=addr,
                        description=(
                            f"Class '{node.name}' is never imported by any other file. "
                            "Consider removing or making private."
                        ),
                        file=file_path, symbol=node.name, detail="no importers found",
                    ))

    return violations


def check_test_coverage_floor(
    manifest: Manifest,
    repo_root: pathlib.Path,
    rule_name: str,
    severity: InvariantSeverity,
    *,
    min_ratio: float = 0.30,
) -> list[CodeViolation]:
    """Require that at least *min_ratio* of functions have a test counterpart.

    A function ``foo`` is considered "tested" if any test file contains a
    function named ``test_foo`` or a class method containing ``foo`` in its
    name.  This is a naming-convention heuristic, not true coverage.
    """
    test_fn_names: set[str] = set()
    all_fn_names: set[str] = set()

    for file_path, content_hash in manifest.items():
        if not file_path.endswith(".py"):
            continue
        source = read_object(repo_root, content_hash)
        if source is None:
            continue
        if len(source) > MAX_AST_BYTES:
            continue
        try:
            tree = ast.parse(source, filename=file_path)
        except SyntaxError:
            continue

        base = pathlib.PurePosixPath(file_path).name
        is_test = base.startswith("test_") or base.endswith("_test.py")

        for node in ast.walk(tree):
            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                if is_test and node.name.startswith("test_"):
                    test_fn_names.add(node.name.removeprefix("test_"))
                elif not is_test and not node.name.startswith("_"):
                    all_fn_names.add(node.name)

    if not all_fn_names:
        return []

    covered = all_fn_names & test_fn_names
    ratio = len(covered) / len(all_fn_names)

    if ratio < min_ratio:
        pct_actual = round(ratio * 100, 1)
        pct_required = round(min_ratio * 100, 1)
        return [CodeViolation(
            rule_name=rule_name,
            severity=severity,
            address="repo",
            description=(
                f"Test coverage floor not met: {pct_actual}% of functions have test counterparts "
                f"(required {pct_required}%). Untested: "
                + ", ".join(sorted(all_fn_names - covered)[:10])
                + ("…" if len(all_fn_names - covered) > 10 else "")
            ),
            file="",
            symbol="",
            detail=f"ratio={ratio:.3f} required={min_ratio:.3f}",
        )]
    return []


# ---------------------------------------------------------------------------
# Rule dispatch
# ---------------------------------------------------------------------------


def _dispatch_rule(
    rule: CodeInvariantRule,
    manifest: Manifest,
    repo_root: pathlib.Path,
    *,
    file_data: FileDataMap | None = None,
    import_graph: ImportGraph | None = None,
) -> list[CodeViolation]:
    """Dispatch a single rule to its implementation function.

    When *file_data* and *import_graph* are supplied, all rule implementations
    use the fast (cached) code path — no I/O or ``ast.parse`` calls are made.
    """
    params = rule.get("params", {})
    rule_name = rule["name"]
    severity = rule["severity"]
    rt = rule["rule_type"]

    if rt == "max_complexity":
        threshold = int(params.get("threshold", 10))
        return check_max_complexity(
            manifest, repo_root, rule_name, severity,
            threshold=threshold, file_data=file_data,
        )

    if rt == "no_circular_imports":
        return check_no_circular_imports(
            manifest, repo_root, rule_name, severity,
            import_graph=import_graph,
        )

    if rt == "no_dead_exports":
        return check_no_dead_exports(
            manifest, repo_root, rule_name, severity,
            file_data=file_data,
        )

    if rt == "test_coverage_floor":
        min_ratio = float(params.get("min_ratio", 0.30))
        return check_test_coverage_floor(manifest, repo_root, rule_name, severity, min_ratio=min_ratio)

    if rt == "forbidden_dependency":
        return check_forbidden_dependency(
            manifest, repo_root, rule_name, severity,
            source_pattern=str(params.get("source_pattern", "")),
            forbidden_pattern=str(params.get("forbidden_pattern", "")),
            import_graph=import_graph,
        )

    if rt == "layer_boundary":
        return check_layer_boundary(
            manifest, repo_root, rule_name, severity,
            lower=str(params.get("lower", "")),
            upper=str(params.get("upper", "")),
            import_graph=import_graph,
        )

    logger.warning("Unknown code invariant rule_type: %r — skipping", rt)
    return []


# ---------------------------------------------------------------------------
# Public entry points
# ---------------------------------------------------------------------------


def load_invariant_rules(
    rules_file: pathlib.Path | None = None,
) -> list[CodeInvariantRule]:
    """Load code invariant rules from TOML, falling back to built-in defaults.

    When *rules_file* is ``None`` the default path (``.muse/code_invariants.toml``)
    is used; if that file is absent or empty, the built-in defaults are returned.

    When *rules_file* is explicitly provided, the file is used as-is.  An empty
    file means "no rules" — the built-in defaults are NOT applied.  This
    allows callers to pass ``--rules empty.toml`` and receive an empty rule set.

    Args:
        rules_file: Path to the TOML file.  ``None`` → use default path.

    Returns:
        List of :class:`CodeInvariantRule` dicts.
    """
    path = rules_file or pathlib.Path(_DEFAULT_RULES_FILE)
    raw = load_rules_toml(path)
    if not raw:
        # Only fall back to defaults when using the implicit default path.
        # An explicit file that is empty means "no rules".
        if rules_file is None:
            return list(_BUILTIN_DEFAULTS)
        return []

    rules: list[CodeInvariantRule] = []
    for r in raw:
        name = str(r.get("name", "unnamed"))
        raw_sev = str(r.get("severity", "warning"))
        _SevMap = dict[str, InvariantSeverity]
        _sev_map: _SevMap = {"info": "info", "warning": "warning", "error": "error"}
        severity: InvariantSeverity = _sev_map.get(raw_sev, "warning")
        scope_raw = str(r.get("scope", "function"))
        _ScopeMap = dict[str, Literal["function", "file", "repo", "global"]]
        _scope_map: _ScopeMap = {
            "function": "function", "file": "file", "repo": "repo", "global": "global",
        }
        scope: Literal["function", "file", "repo", "global"] = _scope_map.get(scope_raw, "function")
        rule_type = str(r.get("rule_type", ""))
        raw_params = r.get("params", {})
        params: ParamsDict = (
            {k: v for k, v in raw_params.items()}
            if isinstance(raw_params, dict)
            else {}
        )
        rule = CodeInvariantRule(
            name=name, severity=severity, scope=scope, rule_type=rule_type, params=params
        )
        rules.append(rule)
    return rules


def run_invariants(
    repo_root: pathlib.Path,
    commit_id: str,
    rules: list[CodeInvariantRule],
) -> BaseReport:
    """Evaluate all rules against the snapshot of *commit_id*.

    Args:
        repo_root:  Repository root.
        commit_id:  Commit to check.
        rules:      Rules to evaluate (from :func:`load_invariant_rules`).

    Returns:
        A :class:`~muse.core.invariants.BaseReport` with all violations.
    """
    manifest = get_commit_snapshot_manifest(repo_root, commit_id)
    if manifest is None:
        logger.warning("Could not load snapshot for commit %s", commit_id)
        return make_report(commit_id, "code", [], 0)

    mutable_manifest = dict(manifest)

    # Pre-flight: parse every Python file exactly once and cache the results.
    # All rules that need AST data share this pre-parsed context — no file is
    # read or parsed more than once per run_invariants call.
    inv_cache = _InvariantFileCache.load(repo_root)
    file_data = _build_file_data(mutable_manifest, repo_root, inv_cache)
    # Build the import graph once; shared by no_circular_imports,
    # forbidden_dependency, and layer_boundary rules.
    import_graph = _build_import_graph(mutable_manifest, repo_root, file_data=file_data)
    # Persist any newly parsed entries for future runs.
    inv_cache.save()

    all_violations: list[BaseViolation] = []
    for rule in rules:
        try:
            code_violations = _dispatch_rule(
                rule, mutable_manifest, repo_root,
                file_data=file_data,
                import_graph=import_graph,
            )
            for cv in code_violations:
                all_violations.append(BaseViolation(
                    rule_name=cv["rule_name"],
                    severity=cv["severity"],
                    address=cv["address"],
                    description=cv["description"],
                ))
        except Exception:
            logger.exception("Error evaluating rule %r on commit %s", rule["name"], commit_id)

    return make_report(commit_id, "code", all_violations, len(rules))


class CodeChecker:
    """Satisfies :class:`~muse.core.invariants.InvariantChecker` for the code domain."""

    def check(
        self,
        repo_root: pathlib.Path,
        commit_id: str,
        *,
        rules_file: pathlib.Path | None = None,
    ) -> BaseReport:
        """Run code invariant checks against *commit_id*."""
        rules = load_invariant_rules(rules_file)
        return run_invariants(repo_root, commit_id, rules)

def check_forbidden_dependency(
    manifest: Manifest,
    repo_root: pathlib.Path,
    rule_name: str,
    severity: InvariantSeverity,
    *,
    source_pattern: str = "",
    forbidden_pattern: str = "",
    import_graph: ImportGraph | None = None,
) -> list[CodeViolation]:
    """Detect files matching *source_pattern* that import from *forbidden_pattern*.

    Both patterns are simple substring matches against workspace-relative file
    paths.  An empty pattern matches nothing, so a misconfigured rule with
    missing patterns produces no violations rather than matching everything.
    """
    if not source_pattern or not forbidden_pattern:
        logger.warning(
            "Rule %r: forbidden_dependency requires source_pattern and "
            "forbidden_pattern params — skipping",
            rule_name,
        )
        return []

    graph = import_graph if import_graph is not None else _build_import_graph(manifest, repo_root)
    violations: list[CodeViolation] = []
    for file_path, deps in sorted(graph.items()):
        if source_pattern not in file_path:
            continue
        for dep in sorted(deps):
            if forbidden_pattern in dep:
                violations.append(CodeViolation(
                    rule_name=rule_name,
                    severity=severity,
                    address=file_path,
                    description=(
                        f"'{file_path}' imports '{dep}' which matches "
                        f"forbidden pattern '{forbidden_pattern}'"
                    ),
                    file=file_path,
                    symbol="",
                    detail=f"dep={dep}",
                ))
    return violations

def check_layer_boundary(
    manifest: Manifest,
    repo_root: pathlib.Path,
    rule_name: str,
    severity: InvariantSeverity,
    *,
    lower: str = "",
    upper: str = "",
    import_graph: ImportGraph | None = None,
) -> list[CodeViolation]:
    """Detect lower-layer files that import from upper-layer files.

    The convention: *lower* (e.g. ``muse/core/``) must not import from
    *upper* (e.g. ``muse/cli/``).  Both are simple substring matches against
    workspace-relative file paths.
    """
    if not lower or not upper:
        logger.warning(
            "Rule %r: layer_boundary requires lower and upper params — skipping",
            rule_name,
        )
        return []

    graph = import_graph if import_graph is not None else _build_import_graph(manifest, repo_root)
    violations: list[CodeViolation] = []
    for file_path, deps in sorted(graph.items()):
        if lower not in file_path:
            continue
        for dep in sorted(deps):
            if upper in dep:
                violations.append(CodeViolation(
                    rule_name=rule_name,
                    severity=severity,
                    address=file_path,
                    description=(
                        f"Lower-layer '{file_path}' imports upper-layer '{dep}' "
                        f"('{lower}' must not import from '{upper}')"
                    ),
                    file=file_path,
                    symbol="",
                    detail=f"lower={lower} upper={upper} dep={dep}",
                ))
    return violations