#!/usr/bin/env python3 """Kit drift scan — canonical vs every mirror + the independently-evolving entabeni kit. Emits drift.json beside this script; run `python3 _build.py` after to regenerate kit-drift.html. Read-only everywhere. Mirrors: drift = files differing from canonical (rsync -an) + local-only scar ids (L34 harvest-first signal). Entabeni: NOT a mirror — tracked as its own kit: EL count, recent changes, structure. (L117: freshness from content where possible.)""" import os as _os; _os.chdir(_os.path.dirname(_os.path.abspath(__file__))) import json, os, re, subprocess, time SRC = "/Users/wolf/Projects/Kit/think-like-fable" OUT = os.path.join(os.path.dirname(os.path.abspath(__file__)), "drift.json") run = lambda cmd, timeout=30: subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout).stdout ids_of = lambda path: set(re.findall(r'^\s*- id:\s*([^\s#]+)', open(path).read(), re.M)) if os.path.isfile(path) else set() canon_ids = ids_of(f"{SRC}/kernel/lessons.yaml") mirrors = [] import glob copies = sorted(glob.glob("/Users/wolf/Projects/*/think-like-fable")) for c in copies: if not os.path.isdir(c) or os.path.realpath(c) == os.path.realpath(SRC): continue name = c.split("/Projects/")[-1].split("/")[0] if "/Projects/" in c else "terminus" itemized = run(f'rsync -an --delete --itemize-changes --exclude=GAME_COPY_NOTE.md --exclude=__pycache__/ --exclude="*.pyc" "{SRC}/" "{c}/"') changed = [ln.split(None, 1)[1] for ln in itemized.splitlines() if ln[:1] in "<>ch*" and len(ln.split(None, 1)) > 1 and not ln.startswith(".d..t")] local_scars = sorted(ids_of(f"{c}/kernel/lessons.yaml") - canon_ids) prov = len([s for s in local_scars if s.startswith("PROV-")]) mirrors.append({ "name": name, "path": c, "files_drifting": changed[:40], "drift_count": len(changed), "local_scars": local_scars, "prov": prov, "status": "LOCAL-SCARS" if local_scars else ("BEHIND" if changed else "CLEAN"), }) # entabeni — independent kit on macB, watched not resynced # --- Instances: each derived orchestration-kit instance (registry _instances.json), watched for # NEW lessons to promote UP into the god kit (Wolf 2026-07-18: "watch each instance ... update # lessons in both from instances"). Lesson-delta = scar ids not seen at the last scan, and scar ids # the instance holds that the god kit lacks (promotion candidates: EL -> L). State persists so a # lesson surfaces exactly once as NEW, and stays flagged as UNPROMOTED until it lands in canonical. INST_FILE = os.path.join(HERE_DIR, "_instances.json") if (HERE_DIR := os.path.dirname(os.path.abspath(__file__))) else "_instances.json" STATE_FILE = os.path.join(os.path.dirname(INST_FILE), "_instance-state.json") try: registry = json.load(open(INST_FILE)) except Exception: registry = [{"name": "think-like-entabeni", "ssh": "extra", "path": "~/Desktop/EntabeniRepos/think-like-entabeni", "scar_prefix": "EL", "desc": "the Entabeni engagement (first derived instance)"}] try: prev_state = json.load(open(STATE_FILE)) except Exception: prev_state = {} # god-kit scar 'stems' (kebab title after the id) — an instance lesson is 'already in canon' if a # canonical scar shares its stem, even under a different L-number. canon_stems = {i.split("-", 1)[1] for i in canon_ids if "-" in i} # DISPOSITION LEDGER (2026-07-18): each instance EL that is already covered/promoted in canon, or is # engagement-specific (client-only, not general doctrine), is dispositioned here — keyed by full EL id — # so the drift scan stops flagging every EL as an open promotion candidate (the exact-kebab-stem match # below over-counts: a covered lesson under a differently-worded L still failed the stem test). A NEW EL # absent from the ledger still surfaces as unpromoted (a real triage signal). LEDGER_FILE = os.path.join(os.path.dirname(INST_FILE), "_promotion-ledger.json") try: LEDGER = json.load(open(LEDGER_FILE)) except Exception: LEDGER = {} instances = [] new_state = {} for inst in registry: name = inst["name"]; ssh = inst.get("ssh", ""); path = inst["path"]; pref = inst.get("scar_prefix", "EL") rec = {"name": name, "ssh": ssh, "path": path, "scar_prefix": pref, "desc": inst.get("desc", ""), "reachable": False} try: if ssh: out = run(f"ssh -o ConnectTimeout=8 {ssh} " f"'d={path}; grep -oE \"^\\s*- id: {pref}[0-9A-Za-z-]*\" $d/kernel/lessons.yaml | " f"sed \"s/.*id: //\"; echo ===CHANGED===; " f"find $d -type f -newermt \"24 hours ago\" ! -name .DS_Store | sed \"s|$d/||\" | head -20'", timeout=25) else: # local instance (host: "local", no ssh) — read the path directly; quote for spaces. # A missing kernel is a hard error, not "reachable with 0 scars": the ===CHANGED=== # echo below succeeds even when the path is gone, which once painted a moved # instance as healthy-and-empty (terminus TL=0 while its real ledger held 21). d = os.path.expanduser(path) if not os.path.isfile(os.path.join(d, "kernel", "lessons.yaml")): raise FileNotFoundError(f"no kernel/lessons.yaml under {d}") out = run(f"d=\"{d}\"; grep -oE \"^\\s*- id: {pref}[0-9A-Za-z-]*\" \"$d/kernel/lessons.yaml\" | " f"sed \"s/.*id: //\"; echo ===CHANGED===; " f"find \"$d\" -type f -newermt \"24 hours ago\" ! -name .DS_Store | sed \"s|$d/||\" | head -20", timeout=25) if out.strip(): blocks = out.split("===CHANGED===") ids = [x.strip() for x in blocks[0].splitlines() if x.strip()] changed = [x.strip() for x in (blocks[1].splitlines() if len(blocks) > 1 else []) if x.strip()] rec["reachable"] = True rec["scar_count"] = len(ids) rec["scar_ids"] = ids prev_ids = set(prev_state.get(name, {}).get("scar_ids", [])) rec["new_since_last"] = [i for i in ids if i not in prev_ids] if prev_ids else [] # promotion candidates: instance scars whose kebab stem is NOT in canonical AND that have not # been dispositioned in the promotion ledger (covered/promoted/engagement-specific are excluded) rec["unpromoted"] = [i for i in ids if "-" in i and i.split("-", 1)[1] not in canon_stems and i not in LEDGER] rec["dispositioned"] = sum(1 for i in ids if i in LEDGER) rec["changed_24h"] = changed new_state[name] = {"scar_ids": ids} else: rec["error"] = "unreachable or empty" new_state[name] = prev_state.get(name, {}) except Exception as e: rec["error"] = str(e)[:120] new_state[name] = prev_state.get(name, {}) instances.append(rec) json.dump(new_state, open(STATE_FILE, "w"), indent=1) # keep the single-instance 'entabeni' key for the existing drift page, sourced from the registry scan enta = next((r for r in instances if r["name"] == "think-like-entabeni"), instances[0] if instances else {}) enta = {**enta, "host": "macB", "el_count": enta.get("scar_count"), "latest_el": enta.get("scar_ids", [])[-3:], "changed_24h": enta.get("changed_24h", [])} data = {"scanned": time.strftime("%Y-%m-%d %H:%M:%S"), "canon_scars": len([i for i in canon_ids if i.startswith("L")]), "mirrors": mirrors, "entabeni": enta, "instances": instances, "promotion_queue": sum(len(r.get("unpromoted", [])) for r in instances), "el_disposition": {b: sum(1 for v in LEDGER.values() if v.get("bucket") == b) for b in ("covered", "promoted", "engagement-specific")}, "summary": {"clean": sum(1 for m in mirrors if m["status"] == "CLEAN"), "behind": sum(1 for m in mirrors if m["status"] == "BEHIND"), "local_scars": sum(1 for m in mirrors if m["status"] == "LOCAL-SCARS")}} json.dump(data, open(OUT, "w"), indent=1) s = data["summary"] print(f"drift.json: {len(mirrors)} mirrors — {s['clean']} clean, {s['behind']} behind, " f"{s['local_scars']} with LOCAL SCARS · entabeni: " f"{'EL=' + str(enta.get('el_count')) if enta.get('reachable') else 'UNREACHABLE'}")