dream-cycle.py

#!/usr/bin/env python3
"""Dream cycle - nightly vault consolidation.

Idempotent nightly maintenance (ARCHITECTURE.md, Layer 2):
  1. Backlink repair (Iron Law): notes mentioning an entity page must add a
     backlink to that page's "Referenced in" section.
  2. NOW.md cleanup: prune essentials.md, close resolved threads, append
     recent.md changelog line.
  3. Memory decay: apply TTL decay to the memory store.
  4. Dedupe report: flag near-duplicate entity pages (report only, never
     deletes).

Usage:
    python3 bin/dream-cycle.py [--vault PATH] [--quick|--full] [--dry-run]

--quick = steps 2,3 (daily default). --full = everything (weekly).
--dry-run prints what would change and writes nothing.
"""
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) -> list[str]:
    """Extract [[wikilink]] targets from note text."""
    return re.findall(r"\[\[([^\]|#]+)", text)


def slugify(name: str) -> str:
    return re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")


def main():
    ap = argparse.ArgumentParser(description="Nightly vault consolidation")
    ap.add_argument("--vault", help="vault path (or BRAINSTACK_VAULT env)")
    ap.add_argument("--quick", action="store_true", help="only NOW.md + decay")
    ap.add_argument("--full", action="store_true", help="everything, incl. backlinks + dedupe")
    ap.add_argument("--dry-run", action="store_true", help="print plan, write nothing")
    args = ap.parse_args()

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

    mode = "full" if args.full else ("quick" if args.quick else "quick")
    actions = []  # (action, detail) for the report

    # --- Entity pages (People/, Organizations/, Projects/) ----------------
    entity_pages = {}
    for root in ("People", "Organizations", "Projects"):
        d = vault / root
        if d.is_dir():
            for p in d.rglob("*.md"):
                entity_pages[p.stem] = p

    # --- 1. Backlink repair (Iron Law) ------------------------------------
    if mode == "full":
        for note in list(vault.rglob("*.md")):
            rel = str(note.relative_to(vault))
            if rel.startswith((".brainstack", "Archive", "02-MOCs")):
                continue
            try:
                text = note.read_text(encoding="utf-8", errors="replace")
            except OSError:
                continue
            for target in parse_links(text):
                page = entity_pages.get(target)
                if not page or page == note:
                    continue
                if "Referenced in" not in page.read_text(encoding="utf-8", errors="replace"):
                    if args.dry_run:
                        actions.append(("backlink", f"{page.name} <- {rel} (missing)"))
                        continue
                    with page.open("a", encoding="utf-8") as f:
                        f.write(f"\n- **{datetime.now():%Y-%m-%d}** | "
                                f"Referenced in [[{note.stem}]] - auto (dream cycle)\n")
                    actions.append(("backlink", f"{page.name} <- {rel}"))

    # --- 2. NOW.md cleanup ------------------------------------------------
    now_dir = vault / "01-Wiki" / "now"
    essentials = now_dir / "essentials.md"
    if essentials.exists():
        if args.dry_run:
            actions.append(("now", "essentials.md: prune stale items"))
        else:
            # Keep only items with a date >= today or marked sticky
            today = datetime.now().strftime("%Y-%m-%d")
            kept, removed = [], 0
            for line in essentials.read_text(encoding="utf-8").splitlines():
                m = re.search(r"(\d{4}-\d{2}-\d{2})", line)
                if line.strip().startswith("#") or not line.strip():
                    kept.append(line)
                elif m and m.group(1) < today and "sticky" not in line:
                    removed += 1
                else:
                    kept.append(line)
            essentials.write_text("\n".join(kept) + "\n", encoding="utf-8")
            actions.append(("now", f"essentials.md: pruned {removed} stale lines"))

    recent = now_dir / "recent.md"
    if recent.exists() and not args.dry_run:
        line = f"- **{datetime.now():%Y-%m-%d %H:%M}** | dream cycle ran (mode={mode})\n"
        content = recent.read_text(encoding="utf-8")
        recent.write_text(line + content, encoding="utf-8")
        actions.append(("now", "recent.md: appended dream cycle entry"))

    # --- 3. Memory decay ---------------------------------------------------
    store_file = vault / ".brainstack" / "memory-store.json"
    if store_file.exists() and not args.dry_run:
        store = json.loads(store_file.read_text())
        changed = 0
        for e in store:
            if e["type"] in ("preference", "decision"):
                for days, delta in ((30, -0.1), (90, -0.2)) if e["type"] == "preference" else ((180, -0.05),):
                    if e["confidence"] > 0.0:
                        e["confidence"] = round(max(0.0, e["confidence"] + delta), 2)
                        changed += 1
        store_file.write_text(json.dumps(store, indent=2))
        actions.append(("decay", f"applied decay to {changed} entries"))

    # --- 4. Dedupe report (full mode) -------------------------------------
    if mode == "full":
        seen, dupes = {}, []
        for root in ("People", "Organizations"):
            d = vault / root
            if not d.is_dir():
                continue
            for p in d.rglob("*.md"):
                key = slugify(p.stem)
                if key in seen:
                    dupes.append(f"{seen[key]} ~ {p.relative_to(vault)}")
                else:
                    seen[key] = str(p.relative_to(vault))
        for d in dupes:
            actions.append(("dupe", d))

    # --- Report ------------------------------------------------------------
    print(f"# Dream cycle ({mode}) - {datetime.now():%Y-%m-%d %H:%M}")
    if args.dry_run:
        print(f"DRY RUN - {len(actions)} actions planned:")
    for kind, detail in actions:
        print(f"  [{kind}] {detail}")
    if not actions:
        print("  nothing to do - vault is clean.")


if __name__ == "__main__":
    main()