`knowledge-layer/scripts/kl_scan.py`

132 lines
#!/usr/bin/env python3
"""
kl_scan.py — incremental compile work-list for a Knowledge Layer.

The whole point of the Karpathy/gBrain pattern is that you DON'T recompile the
whole wiki every time. This script content-hashes everything under `sources/`,
diffs against `.kl/manifest.json`, and prints exactly what changed so the
compile agent only touches NEW / CHANGED sources (and knows which entries a
REMOVED source used to support).

Usage:
    kl_scan.py [<kl-root>]              # show the work-list (read-only)
    kl_scan.py [<kl-root>] --json       # same, machine-readable
    kl_scan.py [<kl-root>] --commit     # mark current sources as compiled

Run WITHOUT --commit before a compile pass (to get the work-list) and WITH
--commit only after you have finished writing the wiki entries for those
sources — otherwise the manifest will claim work is done that isn't.

Exit codes: 0 = nothing to do, 10 = there is pending work, 2 = error.
"""
import argparse
import hashlib
import json
import sys
from datetime import datetime, timezone
from pathlib import Path

TEXT_SUFFIXES = {".md", ".markdown", ".txt", ".rst", ".org", ".json",
                 ".yaml", ".yml", ".csv", ".html", ".htm", ".vtt", ".srt"}


def sha256(p: Path) -> str:
    h = hashlib.sha256()
    with p.open("rb") as fh:
        for chunk in iter(lambda: fh.read(65536), b""):
            h.update(chunk)
    return h.hexdigest()


def scan_sources(root: Path) -> dict:
    src = root / "sources"
    out = {}
    if not src.is_dir():
        return out
    for p in sorted(src.rglob("*")):
        if not p.is_file() or p.name.startswith("."):
            continue
        if p.suffix.lower() not in TEXT_SUFFIXES:
            # still track binaries by hash so their removal is noticed
            pass
        rel = p.relative_to(root).as_posix()
        st = p.stat()
        out[rel] = {"hash": sha256(p), "bytes": st.st_size}
    return out


def load_manifest(root: Path) -> dict:
    m = root / ".kl" / "manifest.json"
    if m.exists():
        try:
            return json.loads(m.read_text())
        except json.JSONDecodeError:
            pass
    return {"version": 1, "compiled_at": None, "sources": {}}


def main() -> int:
    ap = argparse.ArgumentParser(description="Incremental compile work-list.")
    ap.add_argument("root", nargs="?", default="knowledge-layer")
    ap.add_argument("--json", action="store_true", help="machine-readable output")
    ap.add_argument("--commit", action="store_true",
                    help="record current sources as compiled")
    args = ap.parse_args()

    root = Path(args.root).resolve()
    if not (root / ".kl").is_dir():
        print(f"ERROR: {root} is not a Knowledge Layer (run kl_init.py first).",
              file=sys.stderr)
        return 2

    manifest = load_manifest(root)
    prev = manifest.get("sources", {})
    now = scan_sources(root)

    new = [k for k in now if k not in prev]
    changed = [k for k in now if k in prev and now[k]["hash"] != prev[k]["hash"]]
    removed = [k for k in prev if k not in now]

    if args.commit:
        manifest["sources"] = now
        manifest["compiled_at"] = datetime.now(timezone.utc).isoformat()
        (root / ".kl" / "manifest.json").write_text(
            json.dumps(manifest, indent=2) + "\n")
        print(f"committed: {len(now)} sources marked compiled at "
              f"{manifest['compiled_at']}")
        return 0

    pending = len(new) + len(changed) + len(removed)
    if args.json:
        print(json.dumps({"new": new, "changed": changed, "removed": removed,
                          "total_sources": len(now),
                          "last_compiled": manifest.get("compiled_at")}, indent=2))
    else:
        print(f"Knowledge Layer: {root}")
        print(f"last compiled: {manifest.get('compiled_at') or 'never'}")
        print(f"sources: {len(now)} tracked\n")
        if not pending:
            print("Up to date — no NEW/CHANGED/REMOVED sources.")
        else:
            if new:
                print(f"NEW ({len(new)}) — compile into typed entries:")
                for k in new:
                    print(f"  + {k}")
            if changed:
                print(f"CHANGED ({len(changed)}) — re-read & merge into existing "
                      f"entries (surface contradictions, don't silently overwrite):")
                for k in changed:
                    print(f"  ~ {k}")
            if removed:
                print(f"REMOVED ({len(removed)}) — review entries that cited these; "
                      f"drop unsupported claims or mark them stale:")
                for k in removed:
                    print(f"  - {k}")
            print("\nAfter you finish writing entries, run with --commit.")

    return 10 if pending else 0


if __name__ == "__main__":
    sys.exit(main())