vault-health.py

#!/usr/bin/env python3
"""Vault health - audit a brainstack vault and produce a scored report.

Checks (from the `health` skill): orphans, broken links, duplicate pages,
Iron Law violations, schema drift, index staleness, inbox backlog, oversized
notes. Report only - nothing is fixed or deleted.

Usage:
    python3 bin/vault-health.py [--vault PATH] [--json]

Config:
    BRAINSTACK_VAULT env var (default: ~/vault)
"""
import argparse
import json
import os
import re
import sys
from datetime import datetime
from pathlib import Path


def get_vault(args) -> Path:
    return Path(args.vault or os.environ.get("BRAINSTACK_VAULT", "~/vault")).expanduser()


def parse_links(text: str) -> set[str]:
    return set(re.findall(r"\[\[([^\]|#]+)", text))


def main():
    ap = argparse.ArgumentParser(description="Audit vault health")
    ap.add_argument("--vault", help="vault path (or BRAINSTACK_VAULT env)")
    ap.add_argument("--json", action="store_true", help="machine-readable output")
    args = ap.parse_args()

    vault = get_vault(args)
    if not vault.is_dir():
        sys.exit(f"vault not found: {vault}")

    notes = {p for p in vault.rglob("*.md") if ".brainstack" not in p.parts}
    note_paths = {p.stem: p for p in notes}
    report = {"timestamp": datetime.now().isoformat(), "issues": []}

    def issue(sev, check, detail):
        report["issues"].append({"severity": sev, "check": check, "detail": detail})

    # 1. Orphans (no incoming or outgoing links)
    orphans = 0
    for p in notes:
        if p.name in ("log.md", "Tree Index.md") or "now/" in str(p):
            continue
        text = p.read_text(encoding="utf-8", errors="replace")
        if not parse_links(text) and "Referenced in" not in text:
            orphans += 1
    if orphans:
        issue("P2", "orphans", f"{orphans} notes with no links in or out")

    # 2. Broken links
    broken = []
    for p in notes:
        text = p.read_text(encoding="utf-8", errors="replace")
        for target in parse_links(text):
            if target not in note_paths:
                broken.append(f"{p.name} -> [[{target}]]")
    if broken:
        issue("P1", "broken_links", f"{len(broken)} broken: {broken[:5]}...")

    # 3. Duplicates (normalized stem collisions in People/Organizations)
    seen, dupes = {}, []
    for root in ("People", "Organizations"):
        d = vault / root
        if not d.is_dir():
            continue
        for p in d.rglob("*.md"):
            key = re.sub(r"[^a-z0-9]+", "-", p.stem.lower()).strip("-")
            if key in seen:
                dupes.append(f"{seen[key]} ~ {p.relative_to(vault)}")
            else:
                seen[key] = str(p.relative_to(vault))
    if dupes:
        issue("P1", "duplicates", f"{len(dupes)}: {dupes[:5]}...")

    # 4. Schema drift (notes missing type frontmatter)
    drift = 0
    for p in notes:
        if p.name in ("log.md", "Tree Index.md") or "now/" in str(p):
            continue
        text = p.read_text(encoding="utf-8", errors="replace")
        if text.startswith("---") and "type:" not in text.split("---", 2)[1]:
            drift += 1
    if drift:
        issue("P2", "schema_drift", f"{drift} notes missing typed frontmatter")

    # 5. Index staleness
    index_file = vault / ".brainstack" / "index.json"
    if index_file.exists():
        idx_age_days = (datetime.now().timestamp() - index_file.stat().st_mtime) / 86400
        newest = max((p.stat().st_mtime for p in notes), default=0)
        if newest > index_file.stat().st_mtime:
            issue("P1", "index_stale", "semantic index older than newest note - reindex")
        elif idx_age_days > 7:
            issue("P2", "index_stale", f"index {idx_age_days:.0f} days old")
    else:
        issue("P1", "index_missing", "no semantic index - run bin/semantic-index.py")

    # 6. Inbox backlog
    inbox = vault / "00-Inbox"
    if inbox.is_dir():
        old = [p.name for p in inbox.iterdir()
               if p.is_file() and not p.name.startswith(".")
               and (datetime.now().timestamp() - p.stat().st_mtime) > 7 * 86400]
        if old:
            issue("P3", "inbox_backlog", f"{len(old)} items older than 7 days")

    # 7. Oversized notes (atomicity smell)
    heavy = []
    for p in notes:
        size = p.stat().st_size
        if size > 20_000:
            heavy.append(f"{p.name} ({size//1000}KB)")
    if heavy:
        issue("P3", "oversized", f"{len(heavy)} notes >20KB: {heavy[:5]}...")

    # Score: start 10, subtract by severity
    weights = {"P0": 2.0, "P1": 1.0, "P2": 0.5, "P3": 0.2}
    score = max(0.0, 10.0 - sum(weights[i["severity"]] for i in report["issues"]))
    report["score"] = round(score, 1)

    if args.json:
        print(json.dumps(report, indent=2, ensure_ascii=False))
        return

    print(f"# Vault Health - {report['timestamp'][:10]}")
    print(f"Score: {report['score']}/10 ({len(report['issues'])} issues)")
    for i in report["issues"]:
        print(f"  [{i['severity']}] {i['check']}: {i['detail']}")
    if not report["issues"]:
        print("  vault is healthy.")


if __name__ == "__main__":
    main()