echo6-docs/engine/lib/lint.py

#!/usr/bin/env python3
"""
lint.py — Deterministic Vault Lint (stdlib only)

Checks performed (all deterministic):
  1. Frontmatter schema  — required keys, valid type, tags is a list
  2. Tag vocabulary      — tags must be in topic_categories
  3. Dead wikilinks      — [[target]] must resolve against vault note basenames
  4. Orphans             — notes with zero incoming wikilinks (INFO, capped at 40)

Gaps & suggestions (written to ## Gaps & suggestions in lint-report.md):
  - Docs with NO tags
  - True orphans (no inbound link AND shares no tag with any other doc)
  - Dead wikilinks (must be 0 after v4 sweep)
  - Tags not in topic_categories
  - "Earned-a-doc?" candidates: terms mentioned in >= 4 docs with no dedicated doc (top 15)

Usage:
  python3 engine/lib/lint.py           # report mode, always exit 0
  python3 engine/lib/lint.py --strict  # exit 1 if any ERROR findings

Writes: engine/lint-report.md
"""

from __future__ import annotations

import json
import re
import sys
from collections import Counter, defaultdict
from datetime import datetime, timezone
from pathlib import Path
from typing import NamedTuple


# ---------------------------------------------------------------------------
# Config / vocab loading (stdlib-only minimal YAML parser)
# ---------------------------------------------------------------------------

def _parse_simple_yaml(text: str) -> dict:
    result: dict = {}
    current_key = None
    current_list: list | None = None

    for raw_line in text.splitlines():
        line = raw_line.rstrip()
        stripped = line.lstrip()
        if not stripped or stripped.startswith("#"):
            if current_list is not None and not line.startswith(" "):
                result[current_key] = current_list
                current_list = None
                current_key = None
            continue
        if stripped.startswith("- ") and current_list is not None:
            current_list.append(stripped[2:].strip().strip('"').strip("'"))
            continue
        if current_list is not None and not stripped.startswith("-"):
            result[current_key] = current_list
            current_list = None
            current_key = None
        if ":" in stripped:
            key, _, value = stripped.partition(":")
            key = key.strip()
            value = value.strip()
            if value.startswith("[") and value.endswith("]"):
                inner = value[1:-1]
                result[key] = [v.strip().strip('"').strip("'") for v in inner.split(",") if v.strip()]
            elif value == "":
                current_key = key
                current_list = []
            elif value.startswith("#"):
                result[key] = ""
            else:
                value = value.split(" #")[0].strip().strip('"').strip("'")
                result[key] = value
    if current_list is not None and current_key:
        result[current_key] = current_list
    return result


def load_config(engine_dir: Path) -> dict:
    config_path = engine_dir / "config.yaml"
    return _parse_simple_yaml(config_path.read_text(encoding="utf-8"))


def load_vocab(engine_dir: Path) -> dict:
    vocab_path = engine_dir / "vocab.json"
    return json.loads(vocab_path.read_text(encoding="utf-8"))


# ---------------------------------------------------------------------------
# Frontmatter parsing
# ---------------------------------------------------------------------------

_FM_FENCE = re.compile(r"^---\s*$")


def parse_frontmatter(path: Path) -> tuple[dict | None, str]:
    text = path.read_text(encoding="utf-8", errors="replace")
    lines = text.splitlines(keepends=True)

    if not lines or not _FM_FENCE.match(lines[0].rstrip()):
        return None, text

    end_idx = None
    for i in range(1, len(lines)):
        if _FM_FENCE.match(lines[i].rstrip()):
            end_idx = i
            break

    if end_idx is None:
        return None, text

    fm_text = "".join(lines[1:end_idx])
    body = "".join(lines[end_idx + 1:])
    fm = _parse_yaml_frontmatter(fm_text)
    return fm, body


def _parse_yaml_frontmatter(text: str) -> dict:
    result: dict = {}
    current_key: str | None = None
    current_list: list | None = None

    for raw_line in text.splitlines():
        line = raw_line.rstrip()
        stripped = line.lstrip()

        if not stripped:
            continue
        if stripped.startswith("- ") and current_list is not None:
            current_list.append(stripped[2:].strip().strip('"').strip("'"))
            continue
        if current_list is not None:
            result[current_key] = current_list
            current_list = None
            current_key = None

        if ":" in stripped:
            key, _, value = stripped.partition(":")
            key = key.strip()
            value = value.strip()

            if value.startswith("[") and value.endswith("]"):
                inner = value[1:-1]
                result[key] = [
                    v.strip().strip('"').strip("'")
                    for v in inner.split(",")
                    if v.strip()
                ]
            elif value == "":
                current_key = key
                current_list = []
            else:
                result[key] = value.strip('"').strip("'")

    if current_list is not None and current_key:
        result[current_key] = current_list

    return result


# ---------------------------------------------------------------------------
# Check 1: Frontmatter schema
# ---------------------------------------------------------------------------

# v4: entity/concept types removed; valid types are the real doc types
VALID_TYPES = {"reference", "runbook", "project", "note", "index", "session"}


def check_frontmatter(fm: dict | None, config: dict) -> list[dict]:
    findings = []
    required_keys = ["title", "type", "tags", "updated"]

    if fm is None:
        findings.append({
            "severity": "WARN",
            "check": "frontmatter",
            "message": "missing frontmatter block entirely",
        })
        return findings

    for key in required_keys:
        if key not in fm or fm[key] is None or fm[key] == "":
            findings.append({
                "severity": "WARN",
                "check": "frontmatter",
                "message": f"missing required key: {key!r}",
            })

    if "type" in fm and fm["type"]:
        if fm["type"] not in VALID_TYPES:
            findings.append({
                "severity": "WARN",
                "check": "frontmatter",
                "message": f"invalid type {fm['type']!r} — must be one of {sorted(VALID_TYPES)}",
            })

    if "tags" in fm and fm["tags"] is not None:
        if not isinstance(fm["tags"], list):
            findings.append({
                "severity": "WARN",
                "check": "frontmatter",
                "message": f"tags must be a list, got {type(fm['tags']).__name__}: {fm['tags']!r}",
            })

    return findings


# ---------------------------------------------------------------------------
# Check 2: Tag vocabulary
# ---------------------------------------------------------------------------

def check_tags(fm: dict | None, allowed_tags: set[str]) -> list[dict]:
    if fm is None:
        return []
    tags = fm.get("tags")
    if not tags or not isinstance(tags, list):
        return []
    findings = []
    for tag in tags:
        if tag not in allowed_tags:
            findings.append({
                "severity": "WARN",
                "check": "tag-vocab",
                "message": f"unknown tag {tag!r} (not in topic_categories)",
            })
    return findings


# ---------------------------------------------------------------------------
# Check 3: Dead wikilinks
# ---------------------------------------------------------------------------

_WIKILINK_RE = re.compile(
    r"""
    \[\[
    ([^\[\]\n]+?)
    (?:\|[^\[\]\n]*)?
    \]\]
    """,
    re.VERBOSE,
)

_FENCED_CODE_RE = re.compile(r"```.*?```", re.DOTALL)
_INLINE_CODE_RE = re.compile(r"`[^`\n]+`")


def _looks_like_shell(inner: str) -> bool:
    s = inner.strip()
    if re.search(r'\$[{(A-Za-z_]', s):
        return True
    if re.match(r'^\s*[!-]', s):
        return True
    if re.search(r'==|!=|&&|\|\|', s):
        return True
    tokens = s.split()
    if len(tokens) > 1 and tokens[0].startswith("-"):
        return True
    return False


def _is_note_like(inner: str) -> bool:
    if _looks_like_shell(inner):
        return False
    return bool(re.match(r'^[\w\s.\-/]+$', inner.strip()))


def _strip_code_blocks(text: str) -> str:
    text = _FENCED_CODE_RE.sub("", text)
    text = _INLINE_CODE_RE.sub("", text)
    return text


def _extract_wikilinks(body: str) -> list[str]:
    clean = _strip_code_blocks(body)
    targets = []
    for m in _WIKILINK_RE.finditer(clean):
        raw = m.group(1)
        target = raw.split("|")[0].split("#")[0].strip()
        if target and _is_note_like(target):
            targets.append(target)
    return targets


def _normalize_name(name: str) -> str:
    return name.lower().replace(" ", "-").replace("_", "-")


def build_note_index(vault_dir: Path) -> dict[str, Path]:
    index: dict[str, Path] = {}
    for p in vault_dir.rglob("*.md"):
        if "archive" in p.parts:
            continue
        key = _normalize_name(p.stem)
        index[key] = p
    return index


def check_dead_links(
    path: Path,
    body: str,
    note_index: dict[str, Path],
) -> list[dict]:
    findings = []
    for target in _extract_wikilinks(body):
        norm = _normalize_name(target)
        if norm in note_index:
            continue
        findings.append({
            "severity": "ERROR",
            "check": "dead-link",
            "message": f"dead wikilink [[{target}]]",
        })
    return findings


# ---------------------------------------------------------------------------
# Check 4: Orphans
# ---------------------------------------------------------------------------

ORPHAN_CAP = 40


def build_backlink_counts(
    vault_dir: Path,
    note_index: dict[str, Path],
) -> dict[str, int]:
    counts: dict[str, int] = {k: 0 for k in note_index}
    for p in vault_dir.rglob("*.md"):
        if "archive" in p.parts:
            continue
        try:
            _, body = parse_frontmatter(p)
        except Exception:
            continue
        for target in _extract_wikilinks(body):
            norm = _normalize_name(target)
            if norm in counts:
                counts[norm] += 1
    return counts


def find_orphans(
    vault_dir: Path,
    note_index: dict[str, Path],
    backlink_counts: dict[str, int],
) -> list[dict]:
    findings = []
    for key, path in sorted(note_index.items()):
        if backlink_counts.get(key, 0) == 0:
            rel = path.relative_to(vault_dir)
            findings.append({
                "severity": "INFO",
                "check": "orphan",
                "message": f"no incoming links: {rel}",
                "_path": path,
            })
    return findings[:ORPHAN_CAP]


# ---------------------------------------------------------------------------
# Gaps & suggestions
# ---------------------------------------------------------------------------

IMPORTANT_CANDIDATE_TERMS = [
    "aida-nebra", "meshtastic", "headscale", "tailscale", "headplane",
    "qdrant", "aurora", "open-webui", "gemini", "syncthing", "livesync",
    "peertube", "mailcow", "forgejo", "jellyfin", "immich", "nextcloud",
    "vaultwarden", "meshmonitor", "lora", "mt-isr", "sigil", "navi",
    "searxng", "authentik", "docker", "proxmox", "caddy", "dnsmasq",
    "acme-sh", "qdrant", "bge-m3", "qwen", "ollama", "meshtasticd",
    "matrix-synapse", "element", "mautrix", "synapse",
]


def build_earned_a_doc_candidates(
    vault_dir: Path,
    note_index: dict[str, Path],
) -> list[tuple[str, int]]:
    """
    Find terms mentioned in >= 4 docs that have no dedicated vault doc.
    Returns [(term, mention_count)] sorted by count desc, top 15.
    """
    existing_stems = set(note_index.keys())
    doc_mentions: dict[str, set] = defaultdict(set)

    for p in vault_dir.rglob("*.md"):
        if "archive" in p.parts:
            continue
        try:
            text = p.read_text(encoding="utf-8", errors="replace")
        except Exception:
            continue

        # Strip frontmatter
        fm_match = re.match(r"^---.*?---\n", text, re.DOTALL)
        body = text[fm_match.end():] if fm_match else text
        # Strip code blocks
        body = re.sub(r"```.*?```", "", body, flags=re.DOTALL)
        body = re.sub(r"`[^`]+`", "", body)
        body_lower = body.lower()

        path_str = str(p)

        # Check candidate terms
        for term in IMPORTANT_CANDIDATE_TERMS:
            norm = _normalize_name(term)
            if norm in existing_stems:
                continue  # already has a doc
            # Search for the term (and space variant) in body
            variants = [term, term.replace("-", " ")]
            for v in variants:
                if v in body_lower:
                    doc_mentions[term].add(path_str)
                    break

        # Also scan for hyphenated identifiers in wikilinks (already-linked terms
        # that point to non-existent docs would show up as dead links above;
        # here we look for plain-text mentions of kebab terms)
        for m in re.finditer(r"\b([a-z][a-z0-9]{2,}-[a-z0-9][a-z0-9\-]{2,})\b", body_lower):
            t = m.group(1)
            if t not in existing_stems and len(t) >= 6:
                # Filter out version strings, IPs, etc.
                if not re.match(r"^\d", t) and "--" not in t:
                    doc_mentions[t].add(path_str)

    # Filter to >= 4 mentions, sort by count desc, top 15
    candidates = [
        (term, len(docs))
        for term, docs in doc_mentions.items()
        if len(docs) >= 4
    ]
    candidates.sort(key=lambda x: -x[1])
    return candidates[:15]


def build_tag_coverage(
    vault_dir: Path,
    note_index: dict[str, Path],
) -> tuple[list[str], dict[str, list[str]]]:
    """
    Returns (no_tag_paths, tag_to_paths) for shared-tag orphan check.
    no_tag_paths: relative paths of docs with empty tags.
    tag_to_paths: {tag: [doc_stem, ...]}
    """
    no_tag_paths = []
    tag_to_paths: dict[str, list[str]] = defaultdict(list)

    for p in vault_dir.rglob("*.md"):
        if "archive" in p.parts:
            continue
        try:
            fm, _ = parse_frontmatter(p)
        except Exception:
            fm = None
        tags = []
        if fm:
            t = fm.get("tags")
            if isinstance(t, list):
                tags = t
        stem = _normalize_name(p.stem)
        if not tags:
            no_tag_paths.append(str(p.relative_to(vault_dir)))
        for tag in tags:
            tag_to_paths[tag].append(stem)

    return no_tag_paths, tag_to_paths


def build_gaps_section(
    vault_dir: Path,
    note_index: dict[str, Path],
    backlink_counts: dict[str, int],
    all_results: list,
    allowed_tags: set[str],
) -> str:
    """Build the ## Gaps & suggestions section."""
    lines = ["## Gaps & suggestions", ""]

    # 1. Docs with NO tags
    no_tag_paths, tag_to_paths = build_tag_coverage(vault_dir, note_index)
    lines.append("### Docs with no tags")
    lines.append("")
    if no_tag_paths:
        for p in sorted(no_tag_paths):
            lines.append(f"- `{p}`")
    else:
        lines.append("_None — all docs have at least one tag._")
    lines.append("")

    # 2. True orphans (no inbound link AND no shared tag)
    lines.append("### True orphans (no inbound link, no shared tag)")
    lines.append("")
    true_orphans = []
    for key, path in sorted(note_index.items()):
        if backlink_counts.get(key, 0) > 0:
            continue
        # Check if any of its tags appear in other docs
        try:
            fm, _ = parse_frontmatter(path)
        except Exception:
            fm = None
        tags = []
        if fm:
            t = fm.get("tags")
            if isinstance(t, list):
                tags = t
        # Does any other doc share a tag?
        shares_tag = False
        for tag in tags:
            others = tag_to_paths.get(tag, [])
            # Others = docs with this tag; if any != self, shares_tag = True
            self_key = _normalize_name(path.stem)
            if any(s != self_key for s in others):
                shares_tag = True
                break
        if not shares_tag:
            rel = path.relative_to(vault_dir)
            true_orphans.append(str(rel))
    if true_orphans:
        for p in true_orphans[:20]:
            lines.append(f"- `{p}`")
    else:
        lines.append("_None._")
    lines.append("")

    # 3. Dead wikilinks (already in ERROR section, just summarize)
    dead_count = sum(
        1 for r in all_results
        for f in r.findings
        if f["severity"] == "ERROR" and f["check"] == "dead-link"
    )
    lines.append("### Dead wikilinks")
    lines.append("")
    if dead_count == 0:
        lines.append("_None — zero dead wikilinks. All [[links]] resolve._")
    else:
        lines.append(f"**{dead_count} dead wikilinks** — see ERROR section above for details.")
    lines.append("")

    # 4. Tags not in topic_categories
    lines.append("### Unknown tags (not in topic_categories)")
    lines.append("")
    unknown_tags: list[str] = []
    for r in all_results:
        for f in r.findings:
            if f["severity"] == "WARN" and f["check"] == "tag-vocab":
                unknown_tags.append(f["message"])
    if unknown_tags:
        for msg in unknown_tags:
            lines.append(f"- {msg}")
    else:
        lines.append("_None — all tags are in the controlled vocabulary._")
    lines.append("")

    # 5. "Earned-a-doc?" candidates
    lines.append("### Earned-a-doc? candidates (terms in ≥4 docs, no dedicated doc)")
    lines.append("")
    lines.append("Terms mentioned frequently across the vault but with no dedicated doc.")
    lines.append("Matt decides whether to create a real doc — when he does, future sweeps will link to it.")
    lines.append("")
    candidates = build_earned_a_doc_candidates(vault_dir, note_index)
    if candidates:
        lines.append("| Term | Docs mentioning it |")
        lines.append("|------|--------------------|")
        for term, count in candidates:
            lines.append(f"| `{term}` | {count} |")
    else:
        lines.append("_No candidates found (all frequent terms already have dedicated docs)._")
    lines.append("")

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------

class LintResult(NamedTuple):
    path: Path
    findings: list[dict]


def format_report(
    results: list[LintResult],
    vault_dir: Path,
    total_docs: int,
    elapsed_s: float,
    gaps_section: str,
) -> str:
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")

    errors = [f for r in results for f in r.findings if f["severity"] == "ERROR"]
    warns = [f for r in results for f in r.findings if f["severity"] == "WARN"]
    infos = [f for r in results for f in r.findings if f["severity"] == "INFO"]

    fm_missing = sum(
        1 for r in results for f in r.findings
        if f["severity"] == "WARN" and f["check"] == "frontmatter"
        and "missing frontmatter block" in f["message"]
    )
    fm_invalid = sum(
        1 for r in results for f in r.findings
        if f["severity"] == "WARN" and f["check"] == "frontmatter"
        and "missing frontmatter block" not in f["message"]
    )
    tag_warns = sum(
        1 for r in results for f in r.findings
        if f["severity"] == "WARN" and f["check"] == "tag-vocab"
    )

    lines = [
        "# Vault Lint Report",
        "",
        f"Generated: {ts}  |  Docs scanned: {total_docs}  |  Elapsed: {elapsed_s:.1f}s",
        "",
        "## Summary",
        "",
        "| Severity | Count |",
        "|----------|-------|",
        f"| ERROR (dead links) | {len(errors)} |",
        f"| WARN (schema) | {len(warns)} |",
        f"| INFO (orphans) | {len(infos)} |",
        "",
        "### WARN breakdown",
        f"- Missing frontmatter block: {fm_missing}",
        f"- Invalid/missing frontmatter fields: {fm_invalid}",
        f"- Unknown tags: {tag_warns}",
        "",
    ]

    if errors:
        lines.append("## ERROR — Dead Wikilinks")
        lines.append("")
        for r in results:
            errs = [f for f in r.findings if f["severity"] == "ERROR"]
            if errs:
                rel = r.path.relative_to(vault_dir)
                for f in errs:
                    lines.append(f"- `{rel}` — {f['message']}")
        lines.append("")
    else:
        lines.append("## ERROR — Dead Wikilinks")
        lines.append("")
        lines.append("_None. All wikilinks resolve._")
        lines.append("")

    if warns:
        lines.append("## WARN — Schema & Tag Violations")
        lines.append("")
        for r in results:
            ws = [f for f in r.findings if f["severity"] == "WARN"]
            if ws:
                try:
                    rel = r.path.relative_to(vault_dir)
                except ValueError:
                    rel = r.path
                for f in ws:
                    lines.append(f"- `{rel}` — {f['message']}")
        lines.append("")

    if infos:
        lines.append(f"## INFO — Orphan Notes (no incoming links, capped at {ORPHAN_CAP})")
        lines.append("")
        for f in infos:
            lines.append(f"- {f['message']}")
        lines.append("")

    # Append gaps & suggestions section
    lines.append(gaps_section)

    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def run_lint(vault_dir: Path, engine_dir: Path) -> tuple[list[LintResult], dict]:
    import time

    config = load_config(engine_dir)
    vocab = load_vocab(engine_dir)

    topic_cats = set(vocab.get("topic_categories", []))
    # No extra allowed tags in v4 (removed 'meta' catch-all)

    note_index = build_note_index(vault_dir)
    backlink_counts = build_backlink_counts(vault_dir, note_index)

    all_paths = sorted(vault_dir.rglob("*.md"))
    results: list[LintResult] = []

    t0 = time.monotonic()

    for path in all_paths:
        if "archive" in path.parts:
            continue
        findings: list[dict] = []
        try:
            fm, body = parse_frontmatter(path)
        except Exception as exc:
            findings.append({
                "severity": "WARN",
                "check": "parse-error",
                "message": f"could not parse: {exc}",
            })
            results.append(LintResult(path=path, findings=findings))
            continue

        findings.extend(check_frontmatter(fm, config))
        findings.extend(check_tags(fm, topic_cats))
        findings.extend(check_dead_links(path, body, note_index))
        results.append(LintResult(path=path, findings=findings))

    # Orphan check
    orphan_findings = find_orphans(vault_dir, note_index, backlink_counts)
    if orphan_findings:
        results.append(LintResult(path=vault_dir / "_orphans_", findings=orphan_findings))

    elapsed = time.monotonic() - t0

    stats = {
        "total_docs": len(all_paths),
        "errors": sum(1 for r in results for f in r.findings if f["severity"] == "ERROR"),
        "warns": sum(1 for r in results for f in r.findings if f["severity"] == "WARN"),
        "infos": sum(1 for r in results for f in r.findings if f["severity"] == "INFO"),
        "elapsed_s": elapsed,
        "note_index": note_index,
        "backlink_counts": backlink_counts,
        "topic_cats": topic_cats,
    }
    return results, stats


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def main() -> None:
    import time

    strict = "--strict" in sys.argv

    this_file = Path(__file__).resolve()
    engine_dir = this_file.parent.parent
    config = load_config(engine_dir)

    vault_str = config.get("vault_dir", "")
    vault_dir = Path(vault_str) if vault_str else engine_dir.parent / "vault"

    if not vault_dir.exists():
        print(f"ERROR: vault_dir not found: {vault_dir}", file=sys.stderr)
        sys.exit(1)

    results, stats = run_lint(vault_dir, engine_dir)

    gaps_section = build_gaps_section(
        vault_dir,
        stats["note_index"],
        stats["backlink_counts"],
        results,
        stats["topic_cats"],
    )

    report_text = format_report(
        results, vault_dir, stats["total_docs"], stats["elapsed_s"],
        gaps_section,
    )

    report_path = engine_dir / "lint-report.md"
    report_path.write_text(report_text, encoding="utf-8")

    print(report_text)
    print(f"--- Wrote: {report_path} ---")
    print(
        f"\nSummary: {stats['total_docs']} docs | "
        f"{stats['errors']} ERRORs | {stats['warns']} WARNs | {stats['infos']} INFOs"
    )

    if strict and stats["errors"] > 0:
        sys.exit(1)
    sys.exit(0)


if __name__ == "__main__":
    main()