echo6-docs/engine/lib/lint.py
Matt Johnson 44f0257376 docs: migrate Authentik (SSO keystone) to edge2 CT 105
- Authentik -> edge2 CT 105 (Postgres pg_dump/restore; SECRET_KEY carried verbatim; zero-downtime until ~2s cutover)
- Multi-block Caddy cutover: auth.echo6.co + notes.echo6.co outpost/forward_auth -> 100.64.0.36:9000
- runbook: add reboot tailscale-before-docker gotcha; clarify dnsmasq must NOT be repointed (points at Caddy host)
- source left stopped + intact on Contabo as cold rollback

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-18 05:49:07 +00:00

697 lines
22 KiB
Python

#!/usr/bin/env python3
"""
lint.py — Deterministic Vault Lint (no LLM, 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 + {meta}
3. Dead wikilinks — [[target]] resolves against note basenames + entity vocab
4. Orphans — notes with zero incoming wikilinks (INFO, capped at 40)
Severity:
ERROR — dead wikilinks
WARN — schema violations, unknown tags
INFO — orphan notes
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 os
import re
import sys
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:
"""
Minimal YAML parser: handles only the scalar/list constructs in config.yaml.
Supports: key: value, key: [a, b], - item under a key, # comments.
Does NOT handle nested dicts beyond top-level — enough for config.yaml.
"""
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("#"):
# Flush pending list if indent drops
if current_list is not None and not line.startswith(" "):
result[current_key] = current_list
current_list = None
current_key = None
continue
# List item under current key
if stripped.startswith("- ") and current_list is not None:
current_list.append(stripped[2:].strip().strip('"').strip("'"))
continue
# End of list block
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("]"):
# Inline list: [a, b, c]
inner = value[1:-1]
result[key] = [v.strip().strip('"').strip("'") for v in inner.split(",") if v.strip()]
elif value == "":
# May start a list block
current_key = key
current_list = []
elif value.startswith("#"):
result[key] = ""
else:
# Scalar — strip inline comment
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:
"""Load config.yaml using minimal YAML parser."""
config_path = engine_dir / "config.yaml"
text = config_path.read_text(encoding="utf-8")
return _parse_simple_yaml(text)
def load_vocab(engine_dir: Path) -> dict:
"""Load vocab.json."""
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]:
"""
Parse YAML frontmatter block from a markdown file.
Returns (frontmatter_dict_or_None, body_text).
body_text is the full file text if no frontmatter.
"""
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:
"""
Parse simple flat YAML frontmatter. Handles:
key: scalar
key: [a, b, c]
key:
- a
- b
tags: [a, b]
tags:
- a
"""
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
# List item
if stripped.startswith("- ") and current_list is not None:
current_list.append(stripped[2:].strip().strip('"').strip("'"))
continue
# New key — flush pending list
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
# ---------------------------------------------------------------------------
VALID_TYPES = {"reference", "runbook", "project", "note", "index", "session"}
def check_frontmatter(fm: dict | None, config: dict) -> list[dict]:
"""
Returns list of findings dicts: {severity, check, message}
WARN severity for all schema violations.
"""
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
# ---------------------------------------------------------------------------
ALLOWED_EXTRA_TAGS = {"meta"}
def check_tags(fm: dict | None, allowed_tags: set[str]) -> list[dict]:
"""Returns WARN findings for tags outside the allowed vocabulary."""
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 or allowed extras)",
}
)
return findings
# ---------------------------------------------------------------------------
# Check 3: Dead wikilinks
# ---------------------------------------------------------------------------
# Matches [[target]], [[target|alias]], [[target#heading]]
# But NOT bash [[ ... ]] test syntax.
# Strategy: require the inner text to look like a note name:
# - no spaces at start/end of target
# - may contain word chars, hyphens, dots, spaces, #, | — but NOT operators like
# -n, -f, ==, !, &&, ||, etc.
# We detect shell syntax by checking for space-separated words starting with - or
# containing shell operators.
_WIKILINK_RE = re.compile(
r"""
\[\[ # opening [[
([^\[\]\n]+?) # capture: link target (non-greedy, no newlines)
(?:\|[^\[\]\n]*)? # optional |alias
\]\] # closing ]]
""",
re.VERBOSE,
)
_FENCED_CODE_RE = re.compile(r"```.*?```", re.DOTALL)
_INLINE_CODE_RE = re.compile(r"`[^`\n]+`")
def _looks_like_shell(inner: str) -> bool:
"""Return True if the wikilink inner text looks like bash test syntax."""
# Shell: -n "$VAR", ! -f "$FILE", "$CODEC" == "value", -z, etc.
s = inner.strip()
# Contains bash operators or variable expansions
if re.search(r'\$[{(A-Za-z_]', s):
return True
# Starts with - (flag) or ! (negation), or contains == / != / && / ||
if re.match(r'^\s*[!-]', s):
return True
if re.search(r'==|!=|&&|\|\|', s):
return True
# Multiple space-separated tokens that look like args
tokens = s.split()
if len(tokens) > 1 and tokens[0].startswith("-"):
return True
return False
def _is_note_like(inner: str) -> bool:
"""
Return True if inner text looks like a note name:
word chars, hyphens, dots, spaces, but no shell operators.
"""
if _looks_like_shell(inner):
return False
# Must match a reasonable note-name pattern
# Allow: letters, digits, hyphens, underscores, dots, spaces
return bool(re.match(r'^[\w\s.\-/]+$', inner.strip()))
def _strip_code_blocks(text: str) -> str:
"""Remove fenced code blocks and inline code to avoid false wikilink matches."""
text = _FENCED_CODE_RE.sub("", text)
text = _INLINE_CODE_RE.sub("", text)
return text
def _extract_wikilinks(body: str) -> list[str]:
"""Extract note targets from wikilinks in body text, ignoring code blocks and shell syntax."""
clean = _strip_code_blocks(body)
targets = []
for m in _WIKILINK_RE.finditer(clean):
raw = m.group(1)
# Strip heading: [[target#heading|alias]] -> target
# Strip alias already handled by regex (group 1 = before |)
target = raw.split("|")[0].split("#")[0].strip()
if target and _is_note_like(target):
targets.append(target)
return targets
def _note_basename(path: Path) -> str:
"""Return the basename (without extension) of a vault note, lowercased."""
return path.stem.lower()
def _normalize_name(name: str) -> str:
"""Normalize: lowercase, spaces↔hyphens."""
return name.lower().replace(" ", "-").replace("_", "-")
def build_note_index(vault_dir: Path) -> dict[str, Path]:
"""
Build a mapping of normalized name → Path for all vault notes.
One entry per file, keyed by normalized basename.
"""
index: dict[str, Path] = {}
for p in vault_dir.rglob("*.md"):
key = _normalize_name(p.stem)
index[key] = p
return index
def build_entity_names(vocab: dict) -> set[str]:
"""
Build a set of normalized entity names + aliases from vocab.json.
These count as valid wikilink targets (they're in the knowledge graph).
"""
names: set[str] = set()
for entity in vocab.get("entities", []):
names.add(_normalize_name(entity["name"]))
names.add(_normalize_name(entity["display"]))
for alias in entity.get("aliases", []):
names.add(_normalize_name(alias))
for cat in vocab.get("topic_categories", []):
names.add(_normalize_name(cat))
return names
def check_dead_links(
path: Path,
body: str,
note_index: dict[str, Path],
entity_names: set[str],
) -> list[dict]:
"""
Returns ERROR findings for wikilinks that resolve to neither a vault note
nor a known entity name.
"""
findings = []
for target in _extract_wikilinks(body):
norm = _normalize_name(target)
if norm in note_index or norm in entity_names:
continue
findings.append(
{
"severity": "ERROR",
"check": "dead-link",
"message": f"dead wikilink [[{target}]]",
}
)
return findings
# ---------------------------------------------------------------------------
# Check 4: Orphans (incoming wikilink count)
# ---------------------------------------------------------------------------
ORPHAN_CAP = 40
def build_backlink_counts(
vault_dir: Path,
note_index: dict[str, Path],
entity_names: set[str],
) -> dict[str, int]:
"""
Count how many other notes link TO each note.
Returns {normalized_stem: count}.
"""
counts: dict[str, int] = {k: 0 for k in note_index}
for p in vault_dir.rglob("*.md"):
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]:
"""
Return INFO findings for notes with zero incoming wikilinks.
Capped at ORPHAN_CAP entries.
"""
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]
# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------
class LintResult(NamedTuple):
path: Path
findings: list[dict]
def format_report(
results: list[LintResult],
vault_dir: Path,
total_docs: int,
elapsed_s: float,
) -> str:
"""Format a human-readable + markdown report."""
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"]
# Count specific warn subcategories
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 = [
f"# Vault Lint Report",
f"",
f"Generated: {ts} | Docs scanned: {total_docs} | Elapsed: {elapsed_s:.1f}s",
f"",
f"## Summary",
f"",
f"| Severity | Count |",
f"|----------|-------|",
f"| ERROR (dead links) | {len(errors)} |",
f"| WARN (schema) | {len(warns)} |",
f"| INFO (orphans) | {len(infos)} |",
f"",
f"### WARN breakdown",
f"- Missing frontmatter block: {fm_missing}",
f"- Invalid/missing frontmatter fields: {fm_invalid}",
f"- Unknown tags: {tag_warns}",
f"",
]
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:
rel = r.path.relative_to(vault_dir)
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("")
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Public API (importable by sweep.sh or other scripts)
# ---------------------------------------------------------------------------
def run_lint(vault_dir: Path, engine_dir: Path) -> tuple[list[LintResult], dict]:
"""
Run all lint checks. Returns (results, stats_dict).
Importable entry point.
"""
import time
config = load_config(engine_dir)
vocab = load_vocab(engine_dir)
# Build allowed tag set
topic_cats = set(vocab.get("topic_categories", []))
topic_cats |= ALLOWED_EXTRA_TAGS
note_index = build_note_index(vault_dir)
entity_names = build_entity_names(vocab)
# Build backlink counts (single pass over all docs)
backlink_counts = build_backlink_counts(vault_dir, note_index, entity_names)
all_paths = sorted(vault_dir.rglob("*.md"))
results: list[LintResult] = []
t0 = time.monotonic()
for path in all_paths:
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, entity_names))
results.append(LintResult(path=path, findings=findings))
# Orphan check (uses backlink counts already computed)
orphan_findings = find_orphans(vault_dir, note_index, backlink_counts)
# Attach orphan findings as a synthetic "vault" result
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,
}
return results, stats
# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------
def main() -> None:
import time
strict = "--strict" in sys.argv
# Locate engine/vault dirs relative to this file
this_file = Path(__file__).resolve()
engine_dir = this_file.parent.parent # engine/lib/lint.py → engine/
config = load_config(engine_dir)
vault_str = config.get("vault_dir", "")
if not vault_str:
# Fallback: sibling of engine_dir named "vault"
vault_dir = engine_dir.parent / "vault"
else:
vault_dir = Path(vault_str)
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)
report_text = format_report(
results, vault_dir, stats["total_docs"], stats["elapsed_s"]
)
# Write report
report_path = engine_dir / "lint-report.md"
report_path.write_text(report_text, encoding="utf-8")
# Print to stdout
print(report_text)
print(f"--- Wrote: {report_path} ---")
# Summary line
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()