#!/usr/bin/env python3 """vault_lint.py — mechanically assert ECHO vault invariants. READ-ONLY. Catches the recurring drift bugs prose rules can't enforce: folder<->status mismatch, duplicate slugs, wikilinks leaking into frontmatter, duplicate "## Agent Log" headings, stale active projects, aging inbox captures, impossible dates, missing frontmatter, broken source_notes, scope drift, and paths that no route in routing.json permits. Cross-platform: pure Python via echo.py. Exit status: 0 = clean, 1 = violations found, 2 = vault unreachable, 3 = vault not bootstrapped (marker missing). Config (env overrides): ECHO_BASE, ECHO_KEY (via echo.py) ECHO_TODAY (default machine date) — pass the conversation's currentDate so stale/aging math uses the same clock the agent writes with. STALE_DAYS (default 30) INBOX_DAYS (default 14) SCOPE_STALE_SESSIONS (default 3) """ from __future__ import annotations import datetime as dt import json import os import re import sys from pathlib import Path SCRIPT_DIR = Path(__file__).resolve().parent sys.path.insert(0, str(SCRIPT_DIR)) import echo # noqa: E402 import echo_index as idx_mod # noqa: E402 import echo_links as links # noqa: E402 TODAY = dt.date.fromisoformat(os.environ.get("ECHO_TODAY") or dt.date.today().isoformat()) STALE_DAYS = int(os.environ.get("STALE_DAYS", "30")) INBOX_DAYS = int(os.environ.get("INBOX_DAYS", "14")) SCOPE_STALE_SESSIONS = int(os.environ.get("SCOPE_STALE_SESSIONS", "3")) LIFECYCLES = ["active", "incubating", "on-hold", "archived"] SKIP = {"README.md", "project-template.md", "decision-template.md"} REQUIRED_FM = ("type", "created") violations: list[tuple[str, str]] = [] def flag(check: str, message: str) -> None: violations.append((check, message)) def get(path: str) -> str | None: status, body = echo.request("GET", echo.vault_url(path)) if status == 404: return None echo.check(status, body, f"get {path}") return body.decode(errors="replace") def list_dir(path: str) -> tuple[list[str], list[str]]: if path and not path.endswith("/"): path += "/" body = get(path) if body is None: return [], [] try: payload = json.loads(body) except json.JSONDecodeError: return [], [] entries = list(payload.get("files", [])) + list(payload.get("folders", [])) files = [entry for entry in entries if not entry.endswith("/")] folders = [entry[:-1] for entry in entries if entry.endswith("/")] return files, folders def walk(prefix: str = ""): files, folders = list_dir(prefix) for filename in files: yield prefix + filename for folder in folders: yield from walk(f"{prefix}{folder}/") def split_frontmatter(text: str) -> tuple[str, str]: lines = text.splitlines() if not lines or lines[0].strip() != "---": return "", text for index in range(1, len(lines)): if lines[index].strip() == "---": return "\n".join(lines[1:index]), "\n".join(lines[index + 1:]) return "", text def parse_frontmatter(text: str) -> tuple[str, dict[str, object]]: raw, _ = split_frontmatter(text) fields: dict[str, object] = {} for line in raw.splitlines(): match = re.match(r"^([A-Za-z_][\w-]*):\s*(.*)$", line) if not match: continue value: object = match.group(2).strip().strip('"').strip("'") if isinstance(value, str) and value.startswith("[") and value.endswith("]"): value = [part.strip().strip('"').strip("'") for part in value[1:-1].split(",") if part.strip()] fields[match.group(1)] = value return raw, fields def parse_date(value: object) -> dt.date | None: match = re.match(r"(\d{4}-\d{2}-\d{2})", str(value or "")) if not match: return None try: return dt.date.fromisoformat(match.group(1)) except ValueError: return None def as_list(value: object) -> list[object]: if value in (None, ""): return [] return value if isinstance(value, list) else [value] def route_matchers(): routing = json.loads((SCRIPT_DIR / "routing.json").read_text(encoding="utf-8")) routes = [(item["id"], re.compile(item["pattern"])) for item in routing.get("routes", [])] retired = [(re.compile(item["pattern"]), item.get("replacement", "")) for item in routing.get("retired", [])] return routes, retired def main() -> int: try: if get("_agent/echo-vault.md") is None: print("vault-lint: marker missing — vault not bootstrapped (run bootstrap.py).", file=sys.stderr) return 3 except Exception as exc: print(f"vault-lint: vault unreachable ({exc}).", file=sys.stderr) return 2 try: routes, retired = route_matchers() except Exception as exc: routes, retired = [], [] flag("routing-manifest", f"could not load routing.json ({exc}) — path checks skipped") all_files = list(walk()) # Path membership + retired-path detection for path in all_files: if routes and not any(rx.match(path) for _, rx in routes): replacement = next((repl for rx, repl in retired if rx.match(path)), None) if replacement is not None: flag("retired-path", f"{path}: retired location — should be {replacement}") else: flag("unknown-path", f"{path}: matches no route in routing.json") # Per-note frontmatter checks template_re = re.compile(r"(^|/)(templates/|.*-template\.md$)") for path in all_files: base = path.rsplit("/", 1)[-1] if base in SKIP or template_re.search(path) or not path.endswith(".md"): continue text = get(path) or "" raw, fm = parse_frontmatter(text) if "[[" in raw: flag("frontmatter-wikilink", f"{path}: '[[...]]' inside frontmatter") missing = [k for k in REQUIRED_FM if not str(fm.get(k, "")).strip()] if fm and missing: flag("missing-frontmatter", f"{path}: missing {', '.join(missing)}") created, updated = parse_date(fm.get("created")), parse_date(fm.get("updated")) if created and updated and updated < created: flag("date-order", f"{path}: updated {updated} is before created {created}") if updated and updated > TODAY: flag("future-date", f"{path}: updated {updated} is in the future (today {TODAY})") for source_note in as_list(fm.get("source_notes")): if "[[" in str(source_note): flag("source-notes-wikilink", f"{path}: source_notes contains a wikilink '{source_note}'") # Projects: folder<->status, stale active, duplicate slugs slug_homes: dict[str, list[str]] = {} for lifecycle in LIFECYCLES: files, _ = list_dir(f"projects/{lifecycle}") for filename in files: if filename in SKIP or not filename.endswith(".md"): continue slug = filename[:-3] slug_homes.setdefault(slug, []).append(lifecycle) text = get(f"projects/{lifecycle}/{filename}") or "" _, fm = parse_frontmatter(text) status = str(fm.get("status", "")).strip() if status and status != lifecycle: flag("folder/status", f"projects/{lifecycle}/{filename}: status='{status}' but folder='{lifecycle}'") if lifecycle == "active": updated = parse_date(fm.get("updated")) if updated and (TODAY - updated).days > STALE_DAYS: flag("stale-active", f"projects/active/{filename}: updated {updated} ({(TODAY - updated).days}d ago) — consider on-hold/") for slug, homes in slug_homes.items(): if len(homes) > 1: flag("duplicate-slug", f"'{slug}' exists in {', '.join(homes)}") # Daily notes: duplicate "## Agent Log" headings for path in all_files: if re.match(r"^journal/daily/.*\.md$", path): text = get(path) or "" count = len(re.findall(r"(?m)^## Agent Log\s*$", text)) if count > 1: flag("duplicate-agent-log", f"{path}: {count} '## Agent Log' headings") # Inbox: captures aging past INBOX_DAYS inbox = get("inbox/captures/inbox.md") or "" for line in inbox.splitlines(): match = re.match(r"^\s*-\s*(\d{4}-\d{2}-\d{2})\b", line) date = parse_date(match.group(1)) if match else None if date and (TODAY - date).days > INBOX_DAYS: flag("aging-inbox", f"inbox capture {date} ({(TODAY - date).days}d): {line.strip()[:80]}") # Scope freshness (drift detector) context = get("_agent/context/current-context.md") if context is not None: _, fm = parse_frontmatter(context) scope_updated = parse_date(fm.get("scope_updated")) if scope_updated is None: flag("scope-no-timestamp", "_agent/context/current-context.md: no scope_updated frontmatter — scope drift cannot be detected; add it (bootstrap.py) and switch scope via `echo.py scope set`") else: since = [ path for path in all_files if (m := re.match(r"^_agent/sessions/(\d{4}-\d{2}-\d{2})", path)) and (d := parse_date(m.group(1))) and d > scope_updated ] if len(since) >= SCOPE_STALE_SESSIONS: flag("scope-stale", f"scope set {scope_updated}; {len(since)} session(s) logged since without a switch — confirm it still reflects current work (or run `echo.py scope set`)") # ---- Graph health: broken wikilinks, orphans, index drift ---------------- md_files = [p for p in all_files if p.endswith(".md")] md_set = set(md_files) basemap: dict[str, str] = {} for p in md_files: basemap.setdefault(idx_mod.slugify(p.rsplit("/", 1)[-1][:-3]), p) skip_link = re.compile(r"(^|/)(templates/|.*-template\.md$)") def resolve_link(target: str): target = target.strip() if not target: return None if "/" in target: cand = target if target.endswith(".md") else target + ".md" return cand if cand in md_set else None return basemap.get(idx_mod.slugify(target)) text_cache: dict[str, str] = {} inbound: set[str] = set() for path in md_files: if path.rsplit("/", 1)[-1] in SKIP or skip_link.search(path): continue text = get(path) or "" text_cache[path] = text _, body = split_frontmatter(text) for t in links.all_wikilinks(body): tp = resolve_link(t) if tp is None: flag("broken-wikilink", f"{path}: [[{t}]] resolves to no note") else: inbound.add(tp) def is_indexable(path: str) -> bool: # READMEs and templates are signposts, not entity notes (sweep skips them too). base = path.rsplit("/", 1)[-1] return (idx_mod.kind_for_path(path) is not None and base not in SKIP and not template_re.search(path)) # orphan: an entity note with no inbound links and an empty ## Related for path in md_files: if not is_indexable(path) or path in inbound: continue text = text_cache.get(path, get(path) or "") if not links.related_targets(text): flag("orphan", f"{path}: no inbound links and empty ## Related ({idx_mod.kind_for_path(path)}) — link it for recall") # index drift: entries pointing at gone files, or indexable notes not yet indexed try: index = idx_mod.load() idx_paths = {e.get("path") for e in index.get("entities", {}).values()} for slug, e in index.get("entities", {}).items(): if e.get("path") not in md_set: flag("index-stale", f"index entry '{slug}' -> {e.get('path')} no longer exists (run sweep.py)") for path in md_files: if is_indexable(path) and path not in idx_paths: flag("index-missing", f"{path}: indexable note absent from entities.json (run sweep.py)") except Exception as exc: flag("index-error", f"could not check entity index ({exc})") if not violations: print("vault-lint: clean — all invariants hold.") return 0 print(f"vault-lint: {len(violations)} violation(s) found\n") labels = { "folder/status": "Folder <-> status mismatch", "duplicate-slug": "Duplicate slug across lifecycle folders", "frontmatter-wikilink": "Wikilink in frontmatter (breaks reading view)", "duplicate-agent-log": "Duplicate '## Agent Log' heading", "stale-active": f"Stale active project (updated > {STALE_DAYS}d)", "aging-inbox": f"Inbox capture aging (> {INBOX_DAYS}d)", "unknown-path": "Path matches no route in routing.json", "retired-path": "Write to a retired/dead path", "missing-frontmatter": "Missing required frontmatter field", "date-order": "updated earlier than created", "future-date": "updated date is in the future", "source-notes-wikilink": "Wikilink in source_notes (must be plain paths)", "routing-manifest": "routing.json problem", "scope-no-timestamp": "current-context has no scope_updated (drift undetectable)", "scope-stale": f"Scope may have drifted (>= {SCOPE_STALE_SESSIONS} sessions since last switch)", "broken-wikilink": "Wikilink resolves to no note (dead link)", "orphan": "Orphan entity note (no inbound links, empty Related)", "index-stale": "Entity-index entry points at a missing file", "index-missing": "Indexable note absent from the entity index", "index-error": "Entity-index check failed", } grouped: dict[str, list[str]] = {} for check, message in violations: grouped.setdefault(check, []).append(message) for check, messages in grouped.items(): print(f"## {labels.get(check, check)}") for message in messages: print(f" - {message}") print() return 1 if __name__ == "__main__": raise SystemExit(main())