#!/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()