374 lines
14 KiB
Python
374 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
vault-librarian.py — deterministic ($0) Librarian report for the R/J/F/I vault.
|
|
|
|
Emits every mechanical fact the Librarian judgment pass needs, so the model only
|
|
reads a small report instead of the whole vault. NO model calls here.
|
|
|
|
Checks:
|
|
1. Broken wikilinks (classified: introduced-guessable vs pre-existing/unresolvable)
|
|
2. Cross-tree topology diff (/vault vs /data; _unsorted exempt)
|
|
3. JD structural violations:
|
|
- a category folder that duplicates an existing ID number (the "09 should be 00.09" class)
|
|
- missing .00 management note per category
|
|
- missing folder-note (Folder Notes plugin)
|
|
- extend-notation `NN+ ...` leaves that never got a real ID
|
|
- notes nested deeper than the ID level
|
|
4. To-do harvest (- [ ] ...) with file + context
|
|
5. Category health (nearing 10-ID limit; duplicate titles across systems = single-home risk)
|
|
|
|
Usage:
|
|
python3 vault-librarian.py # JSON
|
|
python3 vault-librarian.py --pretty # human-readable markdown
|
|
"""
|
|
import os, re, sys, json
|
|
from collections import defaultdict
|
|
sys.path.insert(0, os.path.dirname(os.path.realpath(__file__)))
|
|
import vault_lib as vl
|
|
|
|
TODO_RE = re.compile(r'^\s*[-*]\s+\[ \]\s+(.*)$')
|
|
DONE_RE = re.compile(r'^\s*[-*]\s+\[[xX]\]\s+(.*)$')
|
|
|
|
|
|
def check_broken_links():
|
|
broken = vl.find_broken_links(vl.VAULT)
|
|
guessable, unresolvable = [], []
|
|
for b in broken:
|
|
(guessable if b["guess"] else unresolvable).append(b)
|
|
return {"guessable": guessable, "unresolvable": unresolvable,
|
|
"count": len(broken)}
|
|
|
|
|
|
def check_topology():
|
|
return vl.topology_diff()
|
|
|
|
|
|
def check_structure():
|
|
out = {"missing_dot00": [],
|
|
"missing_folder_note": [], "extend_notation": [], "too_deep": []}
|
|
s = vl.walk_structure(vl.VAULT, include_meta=True)
|
|
id_nums = set(s["ids"].keys())
|
|
# NOTE: the `09 Archive of pre-RJFI tree` subtree is a deliberate garbage dump
|
|
# (Ryan's ruling 2026-07-22) — it is EXCLUDED from all structural checks. Anything
|
|
# may accumulate there; we never flag or fix inside it.
|
|
for dp, dns, fns in os.walk(vl.VAULT):
|
|
dns[:] = [d for d in dns if d not in vl.SKIP_DIRS]
|
|
rel = os.path.relpath(dp, vl.VAULT)
|
|
if vl.is_archived(rel):
|
|
dns[:] = []
|
|
continue
|
|
# extend-notation leaves
|
|
for fn in fns:
|
|
if vl.EXTEND_RE.match(fn):
|
|
out["extend_notation"].append(os.path.relpath(os.path.join(dp, fn), vl.VAULT))
|
|
# (b) .00 category-management notes are NOT a vault-wide requirement (ruling 2026-07-22:
|
|
# category-level "about this category" info can live in the category's own SYS.AC note;
|
|
# .00 is only added ad-hoc when a category needs its own inbox/archive/store). Not checked.
|
|
# (c) missing folder-note per structural folder
|
|
for level in ("areas", "cats", "ids"):
|
|
for code, rel in s[level].items():
|
|
f = os.path.join(vl.VAULT, rel)
|
|
note = os.path.join(f, os.path.basename(f) + ".md")
|
|
if not os.path.exists(note):
|
|
out["missing_folder_note"].append(rel)
|
|
return out
|
|
|
|
|
|
def _todo_bucket(rel, text):
|
|
"""Classify a to-do as 'long_term' or 'pressing'.
|
|
long_term: lives in a spec/design/wishlist doc (big standing checklists we don't nag about).
|
|
pressing: journal entries, *Tasks*/*Todo* notes, or near-term signal (a name/date/question)."""
|
|
low = rel.lower()
|
|
fname = os.path.basename(rel).lower()
|
|
# spec/design/wishlist docs -> long term standing lists
|
|
long_markers = ("spec", "design", "roadmap", "wishlist", "agora", "about the vault",
|
|
"the r-j-f-i system")
|
|
if any(m in low for m in long_markers):
|
|
return "long_term"
|
|
# journal & explicit task notes -> pressing
|
|
if "journal" in low or "11.84" in low or "tasks for" in fname or "todo" in fname or "to-do" in fname:
|
|
return "pressing"
|
|
# near-term signal in the text
|
|
t = text.lower()
|
|
if t.startswith("respond to") or "awaiting" in t or "?" in text or re.search(r'\b20\d\d-\d\d-\d\d\b', text):
|
|
return "pressing"
|
|
return "long_term"
|
|
|
|
|
|
def check_todos():
|
|
pressing, long_term = [], []
|
|
for p in vl.all_notes(vl.VAULT, include_archive=False):
|
|
rel = os.path.relpath(p, vl.VAULT)
|
|
try:
|
|
lines = open(p, encoding="utf-8").read().splitlines()
|
|
except Exception:
|
|
continue
|
|
for i, line in enumerate(lines):
|
|
m = TODO_RE.match(line)
|
|
if m:
|
|
text = m.group(1).strip()
|
|
item = {"file": rel, "line": i + 1, "text": text[:140]}
|
|
(pressing if _todo_bucket(rel, text) == "pressing" else long_term).append(item)
|
|
return {"pressing": pressing, "long_term": long_term,
|
|
"count": len(pressing) + len(long_term)}
|
|
|
|
|
|
def check_category_health():
|
|
s = vl.walk_structure(vl.VAULT, include_meta=True)
|
|
near_limit = []
|
|
per_cat = defaultdict(int)
|
|
for idc in s["ids"]:
|
|
per_cat[idc[:idc.rindex(".")]] += 1
|
|
for cat, n in per_cat.items():
|
|
if n >= 9:
|
|
near_limit.append({"category": cat, "ids": n})
|
|
# duplicate ID-note titles across systems (single-home smell)
|
|
titlemap = defaultdict(list)
|
|
for code, name in s["ids"].items():
|
|
title = name.split(" ", 1)[1] if " " in name else name
|
|
titlemap[title].append(code)
|
|
dup_titles = {t: codes for t, codes in titlemap.items() if len(codes) > 1}
|
|
return {"near_id_limit": near_limit, "duplicate_titles": dup_titles}
|
|
|
|
|
|
def build_report():
|
|
return {
|
|
"counts": vl.build_manifest()["counts"],
|
|
"broken_links": check_broken_links(),
|
|
"topology": check_topology(),
|
|
"structure": check_structure(),
|
|
"todos": check_todos(),
|
|
"category_health": check_category_health(),
|
|
}
|
|
|
|
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
# TUI color helpers
|
|
# ─────────────────────────────────────────────────────────────────────────────
|
|
import shutil
|
|
|
|
_TTY = sys.stdout.isatty() and os.environ.get("NO_COLOR") is None
|
|
|
|
# ANSI color codes
|
|
_BOLD = "\033[1m"
|
|
_DIM = "\033[2m"
|
|
_RED = "\033[31m"
|
|
_GREEN = "\033[32m"
|
|
_YELLOW = "\033[33m"
|
|
_CYAN = "\033[36m"
|
|
_RESET = "\033[0m"
|
|
|
|
|
|
def c(text, code):
|
|
"""Apply color code if TTY; return plain text otherwise."""
|
|
if not _TTY:
|
|
return text
|
|
return f"{code}{text}{_RESET}"
|
|
|
|
|
|
def rule(title):
|
|
"""Draw a full-width section header with title."""
|
|
width = shutil.get_terminal_size((100, 20)).columns
|
|
# Format: "━ TITLE " padded to width with ━
|
|
prefix = "━ "
|
|
suffix = " "
|
|
available = width - len(prefix) - len(suffix) - len(title)
|
|
if available < 0:
|
|
available = 0
|
|
padding = "━" * available
|
|
return c(f"{prefix}{title}{suffix}{padding}", _CYAN)
|
|
|
|
|
|
def badge(count, good_if_zero=True):
|
|
"""Return a color-coded count badge."""
|
|
if count == 0:
|
|
return c("✓ none", _GREEN) if good_if_zero else c("0", _GREEN)
|
|
elif count <= 3:
|
|
return c(str(count), _YELLOW)
|
|
else:
|
|
return c(str(count), _RED)
|
|
|
|
|
|
def box_summary(counts):
|
|
"""Render a boxed summary header with counts."""
|
|
# Build content lines
|
|
items = [
|
|
(c("Systems", _DIM), c(str(counts["systems"]), _CYAN + _BOLD)),
|
|
(c("Areas", _DIM), c(str(counts["areas"]), _CYAN + _BOLD)),
|
|
(c("Categories", _DIM), c(str(counts["categories"]), _CYAN + _BOLD)),
|
|
(c("IDs", _DIM), c(str(counts["ids"]), _CYAN + _BOLD)),
|
|
]
|
|
# Calculate width accounting for ANSI codes (rough estimate)
|
|
box_width = shutil.get_terminal_size((100, 20)).columns
|
|
|
|
# Build box
|
|
if _TTY:
|
|
top = f"┌─{c('─' * (box_width - 4), _CYAN)}─┐"
|
|
else:
|
|
top = f"┌─{'─' * (box_width - 4)}─┐"
|
|
|
|
content = " " + " · ".join(f"{label} {value}" for label, value in items) + " "
|
|
mid = f"│ {content} │"
|
|
|
|
if _TTY:
|
|
bot = f"└─{c('─' * (box_width - 4), _CYAN)}─┘"
|
|
else:
|
|
bot = f"└─{'─' * (box_width - 4)}─┘"
|
|
|
|
return f"{top}\n{mid}\n{bot}"
|
|
|
|
|
|
def pretty(r):
|
|
L = []
|
|
c_counts = r["counts"]
|
|
|
|
# Summary banner
|
|
L.append(box_summary(c_counts))
|
|
L.append("")
|
|
|
|
# Broken links
|
|
bl = r["broken_links"]
|
|
if bl["count"] == 0:
|
|
L.append(c(rule("Broken wikilinks"), _GREEN))
|
|
L.append(f" {c('✓ none', _GREEN)}")
|
|
else:
|
|
L.append(c(rule(f"Broken wikilinks — {bl['count']} found"), _RED))
|
|
if bl["guessable"]:
|
|
guessable_len = len(bl["guessable"])
|
|
L.append(f" {c(f'Likely fixable ({guessable_len})', _GREEN)} — target renamed, single candidate:")
|
|
for b in bl["guessable"][:30]:
|
|
target = b['target']
|
|
guess = b['guess']
|
|
L.append(f" {c(b['file'], _DIM)}: {c(f'[[{target}]]', _BOLD)} {c('→', _DIM)} guess {c(f'[[{guess}]]', _BOLD)}")
|
|
if len(bl["guessable"]) > 30:
|
|
more = len(bl["guessable"]) - 30
|
|
L.append(f" {c(f'… and {more} more', _DIM)}")
|
|
if bl["unresolvable"]:
|
|
unresolvable_len = len(bl["unresolvable"])
|
|
L.append(f" {c(f'Unresolvable ({unresolvable_len})', _YELLOW)} — target never existed / dropped stub:")
|
|
for b in bl["unresolvable"][:30]:
|
|
target = b['target']
|
|
L.append(f" {c(b['file'], _DIM)}: {c(f'[[{target}]]', _BOLD)}")
|
|
if len(bl["unresolvable"]) > 30:
|
|
more = len(bl["unresolvable"]) - 30
|
|
L.append(f" {c(f'… and {more} more', _DIM)}")
|
|
L.append("")
|
|
|
|
# Topology
|
|
t = r["topology"]
|
|
has_mismatch = any(
|
|
t[level]["missing_in_data"] or t[level]["orphan_in_data"]
|
|
for level in ("systems", "areas", "cats", "ids")
|
|
)
|
|
|
|
if not has_mismatch:
|
|
L.append(c(rule("Cross-tree topology (/vault vs /data)"), _GREEN))
|
|
L.append(f" {c('✓ in sync', _GREEN)}")
|
|
else:
|
|
L.append(c(rule("Cross-tree topology — mismatches found"), _YELLOW))
|
|
for level in ("systems", "areas", "cats", "ids"):
|
|
md = t[level]["missing_in_data"]
|
|
od = t[level]["orphan_in_data"]
|
|
if md:
|
|
md_str = ", ".join(md[:20]) + (f" {c('…', _DIM)}" if len(md) > 20 else "")
|
|
L.append(f" {c(f'{level} missing /data ({len(md)})', _YELLOW)}: {md_str}")
|
|
if od:
|
|
od_str = ", ".join(od[:20]) + (f" {c('…', _DIM)}" if len(od) > 20 else "")
|
|
L.append(f" {c(f'{level} orphan in /data ({len(od)})', _YELLOW)}: {od_str}")
|
|
L.append("")
|
|
|
|
# Structure
|
|
st = r["structure"]
|
|
has_struct_issues = bool(st["extend_notation"] or st["missing_dot00"] or st["missing_folder_note"])
|
|
|
|
if has_struct_issues:
|
|
L.append(c(rule("JD structural issues"), _YELLOW))
|
|
if st["extend_notation"]:
|
|
extend_len = len(st["extend_notation"])
|
|
L.append(f" {c(f'extend-notation `NN+` needing real IDs ({extend_len})', _YELLOW)}:")
|
|
for e in st["extend_notation"][:20]:
|
|
L.append(f" {c(e, _DIM)}")
|
|
if len(st["extend_notation"]) > 20:
|
|
more = len(st["extend_notation"]) - 20
|
|
L.append(f" {c(f'… and {more} more', _DIM)}")
|
|
if st["missing_dot00"]:
|
|
codes = ", ".join(sorted(st["missing_dot00"])[:25])
|
|
if len(st["missing_dot00"]) > 25:
|
|
codes += f" {c('…', _DIM)}"
|
|
dot00_len = len(st["missing_dot00"])
|
|
L.append(f" {c(f'categories missing .00 note ({dot00_len})', _YELLOW)}: {codes}")
|
|
if st["missing_folder_note"]:
|
|
folder_len = len(st["missing_folder_note"])
|
|
L.append(f" {c(f'folders missing folder-note ({folder_len})', _YELLOW)}")
|
|
L.append("")
|
|
|
|
# To-dos
|
|
td = r["todos"]
|
|
total_todos = td['count']
|
|
L.append(c(rule(f"To-dos: {c(str(total_todos), _BOLD)} open"), _CYAN))
|
|
pressing_len = len(td["pressing"])
|
|
longterm_len = len(td["long_term"])
|
|
L.append(f" {c(f'Pressing: {pressing_len}', _BOLD)} / {c(f'Long-term: {longterm_len}', _DIM)}")
|
|
|
|
if td["pressing"]:
|
|
L.append(f" {c('Pressing:', _BOLD)}")
|
|
for x in td["pressing"][:40]:
|
|
file_info = c(f'{x["file"]}:{x["line"]}', _CYAN)
|
|
L.append(f" {file_info} {x['text']}")
|
|
if len(td["pressing"]) > 40:
|
|
more = len(td["pressing"]) - 40
|
|
L.append(f" {c(f'… and {more} more', _DIM)}")
|
|
|
|
if td["long_term"]:
|
|
L.append(f" {c('Long-term (standing checklists):', _DIM)}")
|
|
for x in td["long_term"][:15]:
|
|
file_info = c(f'{x["file"]}:{x["line"]}', _DIM)
|
|
text_info = c(x['text'], _DIM)
|
|
L.append(f" {file_info} {text_info}")
|
|
if len(td["long_term"]) > 15:
|
|
more = len(td["long_term"]) - 15
|
|
L.append(f" {c(f'… and {more} more', _DIM)}")
|
|
L.append("")
|
|
|
|
# Category health
|
|
ch = r["category_health"]
|
|
if ch["near_id_limit"] or ch["duplicate_titles"]:
|
|
L.append(c(rule("Category health"), _YELLOW))
|
|
for x in ch["near_id_limit"]:
|
|
category = c(x['category'], _BOLD)
|
|
ids_count = c(str(x['ids']), _YELLOW)
|
|
L.append(f" {category} — {ids_count} IDs, near 10 limit")
|
|
if ch["duplicate_titles"]:
|
|
for t, codes in list(ch["duplicate_titles"].items())[:10]:
|
|
codes_str = ", ".join(codes)
|
|
L.append(f" {c('⚠ duplicate title:', _YELLOW)} {t} {c(codes_str, _DIM)}")
|
|
L.append("")
|
|
|
|
# Footer status bar
|
|
issues_count = sum([
|
|
bl["count"],
|
|
sum(1 for level in ("systems", "areas", "cats", "ids")
|
|
if t[level]["missing_in_data"] or t[level]["orphan_in_data"]),
|
|
len(st["extend_notation"]) + len(st["missing_dot00"]) + len(st["missing_folder_note"]),
|
|
len(ch["near_id_limit"]) + len(ch["duplicate_titles"]),
|
|
])
|
|
|
|
if issues_count == 0:
|
|
L.append(c("● VAULT HEALTHY", _GREEN))
|
|
else:
|
|
L.append(c(f"● {issues_count} item(s) need attention", _YELLOW))
|
|
|
|
return "\n".join(L)
|
|
|
|
|
|
def main():
|
|
r = build_report()
|
|
if "--pretty" in sys.argv:
|
|
print(pretty(r))
|
|
else:
|
|
print(json.dumps(r, indent=2, ensure_ascii=False))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|