#!/usr/bin/env python3 """thread-query.py — the dynamo. Query the corpora with the question you have NOW, rather than with the calendar. /wake-up already reads a recency slice: MEMORY.md, the newest session files, the trackers. Measured 2026-08-24 that is 6,107 words of an 859,803-word memory corpus — 0.71%. Everything else is dark unless someone goes looking. Twice on 2026-08-24 someone did, by hand, and both greps produced the day's best findings: the Hearth definition from a December Compass document, and the humic layer from April. This turns that hand-motion into an instrument. It deliberately EXCLUDES the slice the wake already loads and deliberately FAVOURS age, because its whole job is to surface what recency will not. Returns pointers — path, date, and the literal matching lines. Never summaries: a summary of the steward's thinking, written by the executor, is the thing the vault harvest of 2026-08-24 existed to get away from. """ import argparse, datetime, glob, json, math, os, re, sys, tempfile, shutil HOME = os.path.expanduser("~") MEM = os.path.join(HOME, ".claude/projects/-Users-davidglidden/memory") VAULT = os.path.join(HOME, "Library/Mobile Documents/iCloud~md~obsidian/Documents", "David, root-and-branch") # Path COMPONENTS, never substrings. Earned 2026-08-23: the filter "99. Archive" # is a prefix of "99. Archives—Previous Iterations" and silently swallowed 78 live # notes from every figure of that day. VAULT_EXCLUDED_COMPONENTS = {".obsidian", ".trash", "99. chatgpt-conversations", "thinking-mirror"} # spec §1a: a derived tree; cite the original def is_archive_component(name): """The vault's archive rule, ADOPTED not reinvented. `vault-links.py:152` defines it: ARCHIVE_DIR = "99. Archive", matched as an exact path COMPONENT at any depth. That definition was itself earned on 2026-08-23, when a substring test also swallowed `99. Archives—Previous Iterations` and hid 78 live notes from every figure of the day. A first attempt here used startswith("99. archive"), which reintroduced exactly that bug — caught by the control carried over from that day. Deriving a second, divergent archive rule in a second script is the field-computed-twice failure; if the rule should change, it changes in vault-links.py and both consumers follow. OPEN, deliberately not decided here: nested folders literally named `99. Archives` (e.g. `06. Projects/Pattern, Presence, Practice/99. Archives/`) are NOT excluded by the vault's rule, so they are not excluded here either. Whether they should be is a question about the vault's archive definition, not about this tool. """ return name == "99. Archive" STOPWORDS = set("""a an and are as at be been but by can could did do does for from had has have he her his how i if in into is it its me my no not of on or our out över she so some such than that the their them then there these they this those to too under until up very was we were what when where which while who why will with would you your about after again all also am any because before being between both during each few more most other over same still just now new one two three thing things something anything nothing make makes made get gets got go goes going""".split()) def terms_from(text): """Content terms: lowercase, >=4 chars, not a stopword, deduped, order kept.""" seen, out = set(), [] for w in re.findall(r"[A-Za-z][A-Za-z'-]+", text.lower()): w = w.strip("'-") if len(w) < 4 or w in STOPWORDS or w in seen: continue seen.add(w) out.append(w) return out def file_date(path, text=None): """Prefer a date in the filename, then frontmatter created:, then mtime.""" m = re.search(r"(\d{4})-(\d{2})-(\d{2})", os.path.basename(path)) if m: try: return datetime.date(*map(int, m.groups())) except ValueError: pass if text: m = re.search(r"^(?:created|date):\s*[\"']?(\d{4})-(\d{2})-(\d{2})", text[:2000], re.M) if m: try: return datetime.date(*map(int, m.groups())) except ValueError: pass try: return datetime.date.fromtimestamp(os.path.getmtime(path)) except OSError: return datetime.date.today() def recency_slice(mem_dir, keep=3): """Exactly what /wake-up already loads. Excluded so we never hand back the slice.""" out = set() for n in ("MEMORY.md", "MEMORY-reference.md"): out.add(os.path.join(mem_dir, n)) sessions = sorted(glob.glob(os.path.join(mem_dir, "session-2*.md")), key=lambda p: os.path.basename(p), reverse=True) for p in sessions[:keep]: out.add(p) today = datetime.date.today().isoformat() for p in glob.glob(os.path.join(mem_dir, f"*{today}*.md")): out.add(p) return out def collect(mem_dir, vault_dir, keep=3): """(path, text) for every candidate file, recency slice and archive removed.""" skip = recency_slice(mem_dir, keep) files = [] if os.path.isdir(mem_dir): for p in glob.glob(os.path.join(mem_dir, "*.md")): if os.path.realpath(p) not in {os.path.realpath(s) for s in skip}: files.append(p) if os.path.isdir(vault_dir): for root, dirs, names in os.walk(vault_dir): dirs[:] = [d for d in dirs if d not in VAULT_EXCLUDED_COMPONENTS and not is_archive_component(d) and not d.startswith(".")] for n in names: if n.endswith(".md"): files.append(os.path.join(root, n)) out = [] for p in files: try: out.append((p, open(p, encoding="utf-8", errors="replace").read())) except OSError: continue return out def best_window(text, qterms, window=40): """Distinct query terms co-occurring inside the best ~40-line window. ⚠ EARNED 2026-08-24, live run. Scoring on distinct terms present ANYWHERE in a document made Carruthers' *Book of Memory* (1.6 MB) and Yates' *Art of Memory* (1 MB) the top two hits for every query — a million-word book contains "instruments", "turning" and "heap" somewhere by accident. Presence over an unbounded document measures LENGTH, not relevance. The selftest was 14/14 green at the time, because the fixture held only small uniform files: the same PASS-BUT-FALSELY shape vault-links.py hit the day before. Requiring co-occurrence in a window is what makes a match a match, and it is also what makes the excerpt worth reading. """ lines = text.lower().splitlines() if not lines: return 0, 0 best, best_at = 0, 0 for start in range(0, max(1, len(lines)), max(1, window // 2)): chunk = "\n".join(lines[start:start + window]) n = sum(1 for q in qterms if re.search(r"\b" + re.escape(q), chunk)) if n > best: best, best_at = n, start return best, best_at def score(text, path, qterms, today=None): """Co-occurring terms in the tightest window, weighted by age.""" n, at = best_window(text, qterms) if n < 2: # one term is noise; scattered terms are length return 0.0, [], None low = text.lower() matched = [q for q in qterms if re.search(r"\b" + re.escape(q), low)] d = file_date(path, text) age_days = max(0, ((today or datetime.date.today()) - d).days) # Age is a deliberate thumb on the scale: recency is what the wake already has. weight = 1.0 + math.log1p(age_days / 30.0) return n * weight, matched, d def excerpts(text, matched, limit=2, qterms=None): lines, out = text.splitlines(), [] if qterms: _, at = best_window(text, qterms) lines = lines[at:at + 40] for ln in lines: s = ln.strip() if len(s) < 20 or s.startswith(("---", "#", "|", "```")): continue if sum(1 for t in matched if t in s.lower()) >= 2: out.append(re.sub(r"\s+", " ", s)[:200]) if len(out) >= limit: break return out def run(thread, mem_dir=MEM, vault_dir=VAULT, top=5, keep=3, today=None): qterms = terms_from(thread) rows = [] for path, text in collect(mem_dir, vault_dir, keep): sc, matched, d = score(text, path, qterms, today) if sc > 0: rows.append({"path": path, "score": round(sc, 2), "date": d.isoformat(), "matched": matched, "excerpts": excerpts(text, matched, qterms=qterms)}) rows.sort(key=lambda r: -r["score"]) return {"thread_terms": qterms, "candidates": len(rows), "results": rows[:top]} # --------------------------------------------------------------------------- tests def selftest(): ok = fail = 0 def check(name, cond): nonlocal ok, fail if cond: ok += 1 print(f" PASS {name}") else: fail += 1 print(f" FAIL {name}") d = tempfile.mkdtemp() mem, vault = os.path.join(d, "mem"), os.path.join(d, "vault") arch = os.path.join(vault, "99. Archive") near = os.path.join(vault, "99. Archives—Previous Iterations") for x in (mem, vault, arch, near): os.makedirs(x, exist_ok=True) def w(p, s): open(p, "w", encoding="utf-8").write(s) w(os.path.join(mem, "MEMORY.md"), "compost heap dynamo putrefaction everywhere") w(os.path.join(mem, "session-2026-08-24-a.md"), "compost heap dynamo putrefaction newest") w(os.path.join(mem, "session-2026-08-23-b.md"), "compost heap dynamo putrefaction second") w(os.path.join(mem, "session-2026-08-22-c.md"), "compost heap dynamo putrefaction third") w(os.path.join(mem, "session-2026-01-01-old.md"), "The compost heap and the dynamo turning of putrefaction, an old and forgotten note here.") w(os.path.join(mem, "user-unrelated.md"), "bicycle maintenance and pastry") w(os.path.join(vault, "recent.md"), "created: 2026-08-20\ncompost dynamo putrefaction recent note") w(os.path.join(arch, "archived.md"), "compost dynamo putrefaction archived note") w(os.path.join(near, "live-but-near.md"), "created: 2025-03-01\ncompost dynamo putrefaction in a folder whose name merely starts alike") today = datetime.date(2026, 8, 24) q = "the compost heap and the dynamo, putrefaction" r = run(q, mem, vault, top=20, today=today) paths = [x["path"] for x in r["results"]] base = [os.path.basename(p) for p in paths] # positive control — the instrument can find what is there check("finds a planted old memory note", "session-2026-01-01-old.md" in base) check("finds a planted vault note", "recent.md" in base) # negative control — it does not find what is not there check("nonsense query returns nothing", run("zzzqqx wwwvvy", mem, vault, today=today)["results"] == []) check("unrelated file never returned", "user-unrelated.md" not in base) # the recency slice is excluded — the whole point check("MEMORY.md excluded", "MEMORY.md" not in base) check("3 newest session files excluded", not any(b.startswith(("session-2026-08-24", "session-2026-08-23", "session-2026-08-22")) for b in base)) # the earned path-component lesson check("'99. Archive' excluded by path component", "archived.md" not in base) check("'99. Archives—Previous Iterations' NOT excluded", "live-but-near.md" in base) # THE LENGTH-BIAS CONTROL — earned from the live run, not from the fixture. # A long document with the query terms scattered far apart must not outrank a short # focused one. Without this the two 1MB memory-treatises topped every query. w(os.path.join(mem, "user-huge-scattered.md"), "compost " + ("filler " * 4000) + " dynamo " + ("filler " * 4000) + " putrefaction") w(os.path.join(mem, "user-short-focused.md"), "created: 2026-01-05\nThe compost heap, the dynamo, and putrefaction together in one place.") r2 = run(q, mem, vault, top=20, today=today) b2 = [os.path.basename(x["path"]) for x in r2["results"]] check("scattered long document does not outrank a short focused one", "user-short-focused.md" in b2 and (("user-huge-scattered.md" not in b2) or b2.index("user-short-focused.md") < b2.index("user-huge-scattered.md"))) # derived trees are out of scope per Frontmatter Specification §1a tm = os.path.join(vault, "08. Notes", "CapableMind", "thinking-mirror") os.makedirs(tm, exist_ok=True) w(os.path.join(tm, "mirrored.md"), "compost heap dynamo putrefaction in a derived tree") check("thinking-mirror (derived tree) excluded", "mirrored.md" not in [os.path.basename(x["path"]) for x in run(q, mem, vault, top=20, today=today)["results"]]) # age is a real thumb on the scale old = next(x for x in r["results"] if os.path.basename(x["path"]) == "session-2026-01-01-old.md") new = next(x for x in r["results"] if os.path.basename(x["path"]) == "recent.md") check("older note outranks newer at equal term-match", old["score"] > new["score"]) # single-term matches are noise, not signal w(os.path.join(mem, "user-oneterm.md"), "compost only, nothing else at all in this note") check("single matched term is not enough", "user-oneterm.md" not in [os.path.basename(x["path"]) for x in run(q, mem, vault, top=20, today=today)["results"]]) # term extraction check("stopwords dropped", "the" not in terms_from("the compost and the heap")) check("short words dropped", "and" not in terms_from("compost and heap")) check("content terms kept", set(["compost", "heap"]) <= set(terms_from("the compost and heap"))) # excerpts are literal lines, not summaries check("excerpt is a literal substring of the source", all(e[:60] in re.sub(r"\s+", " ", open(x["path"], encoding="utf-8").read()) for x in r["results"] for e in x["excerpts"])) shutil.rmtree(d, ignore_errors=True) print(f"\nSELFTEST {'PASS' if not fail else 'FAIL'} — {ok} ok, {fail} failed") return 0 if not fail else 1 def main(): ap = argparse.ArgumentParser(description="Query the corpora with the current thread.") ap.add_argument("thread", nargs="?", help="the pulling thread, in words") ap.add_argument("--top", type=int, default=5) ap.add_argument("--keep", type=int, default=3, help="how many newest session files the wake reads") ap.add_argument("--json", action="store_true") ap.add_argument("--selftest", action="store_true") ap.add_argument("--log", metavar="PATH", nargs="?", const=os.path.join(HOME, "dotfiles/claude/governance/thread-query-log.jsonl"), help="append one JSON line per run so the trial grades from a record, " "not from anyone's memory of whether it helped") a = ap.parse_args() if a.selftest: sys.exit(selftest()) if not a.thread: ap.error("give the thread as an argument, or use --selftest") r = run(a.thread, top=a.top, keep=a.keep) if a.log: # Mechanical record. The daily-note failure of 2026-08-24 was a continuous # obligation with no cue; a trial graded from recollection has the same shape. try: os.makedirs(os.path.dirname(a.log), exist_ok=True) with open(a.log, "a", encoding="utf-8") as fh: fh.write(json.dumps({ "date": datetime.date.today().isoformat(), "thread": a.thread[:300], "terms": r["thread_terms"], "candidates": r["candidates"], "returned": [{"path": x["path"].replace(HOME, "~"), "date": x["date"], "matched": x["matched"]} for x in r["results"]], }, ensure_ascii=False) + "\n") except OSError: pass if a.json: print(json.dumps(r, indent=2)) return print(f"THREAD QUERY — {r['candidates']} candidates over {len(r['thread_terms'])} terms; " f"recency slice excluded\n") if not r["results"]: print(" nothing the recency slice would not already have shown.") print(" (That is a real result, not a failure — record it for the trial.)") return for x in r["results"]: rel = x["path"].replace(HOME, "~") print(f" [{x['date']}] {rel}") print(f" matched: {', '.join(x['matched'][:8])}") for e in x["excerpts"]: print(f" > {e}") print() if __name__ == "__main__": main()