session 2026-08-24: the heap got a dynamo — thread-query.py + daybook-cue.py, 11 user-memories harvested to the vault

- thread-query.py: the turning. Queries both corpora with the pulling thread, excluding
  the recency slice (/wake-up reaches 0.71% of an 859,803-word corpus) and favouring age.
  Wired into /wake-up §2.b.3, replacing a described-not-invoked grep step. Trial
  pre-registered, graded 2026-10-05 from --log. Caught PASS-BUT-FALSELY on its first live
  run at 14/14 green; rescored on windowed co-occurrence, length-bias control added.
- daybook-cue.py: PostToolUse cue for the daily note. Diagnosed from the record — the
  hook only ever CREATED and nothing ever prompted filling. 16/16, fail-open, never blocks.
- All 11 user-* memories harvested into the vault (12 notes, 10 into 09. Atlas of Roots,
  empty since 2025-09-29); each memory file back-pointed, vault note canonical for the idea.
- N-now corrected in MEMORY.md: 49/84, down 11 — the counter is a rolling window.

Steward-authorized. Trials: thread-query + Smart Connections, both graded 2026-10-05.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JQKeKY9T9d95KpvHwwok8T
This commit is contained in:
David F Glidden
2026-08-24 11:25:39 +02:00
co-authored by Claude Opus 5
parent e3da9abad1
commit edf71fb173
20 changed files with 1094 additions and 20 deletions
+169
View File
@@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""daybook-cue.py — PostToolUse cue (Write|Edit): the daily note has no trigger.
Diagnosed 2026-08-24 from the record, not from impression. Daily notes for 19-22
August do not exist; 2026-08-23 exists at 33,242 bytes across eight commits — written
incrementally, as the convention specifies. So the practice has produced exactly one
note, on the day the convention was authored and live in working memory.
The failure is structural. `daybook-ensure.py` fires at SessionStart and ONLY EVER
CREATES. The next cue is /wrap-up §7.5, hours later. Between them — the entire working
session, which is exactly when the convention says to write — there is no cue at all.
The vault's own notes diagnosed the two dead predecessors correctly ("neither had a
trigger. The hook is the trigger") and then built the trigger for the half that never
needed one. Whether the file exists was never the failure mode.
The convention's own trigger word is "as it lands", and work lands when a file is
written. So this fires there.
REMINDER, NOT BLOCK — unlike verify-before-compose, whose failure is unrecoverable.
This one's is recoverable, and a gate that interrupts every write would be removed
within a day. Fail-open everywhere: any unexpected condition exits 0.
"""
import datetime, json, os, sys, time
HOME = os.path.expanduser("~")
VAULT = os.path.join(HOME, "Library/Mobile Documents/iCloud~md~obsidian/Documents",
"David, root-and-branch")
DAILY_DIR = os.path.join(VAULT, "01. Daily")
STAMP = os.path.join(HOME, ".claude/state/daybook-cue.stamp")
SKELETON_CEILING = 1200 # the generated skeleton is ~558 bytes
QUIET_SECONDS = 900 # cue at most once per 15 minutes; a nag gets disabled
# Writes that are not "work landing".
IGNORED_FRAGMENTS = ("/scratchpad/", "/private/tmp/", "/tmp/", "/.git/",
"/node_modules/", "/.obsidian/")
IGNORED_SUFFIXES = (".bak", ".stamp", ".lock", ".log", ".jsonl", ".pyc")
def is_substantive(path):
if not path:
return False
if any(f in path for f in IGNORED_FRAGMENTS):
return False
if path.endswith(IGNORED_SUFFIXES):
return False
# writing the daily note itself is the thing being asked for, not a trigger for it
if os.path.normpath(DAILY_DIR) in os.path.normpath(path):
return False
return True
def note_path(today=None):
d = today or datetime.date.today()
return os.path.join(DAILY_DIR, f"{d.isoformat()}.md")
def should_cue(path, note_size, note_exists, seconds_since_last, today=None):
"""Pure decision, so it can be tested without a filesystem or a clock."""
if not is_substantive(path):
return False
if not note_exists: # daybook-ensure owns creation, not us
return False
if note_size >= SKELETON_CEILING: # it is being written; say nothing
return False
if seconds_since_last < QUIET_SECONDS:
return False
return True
MESSAGE = """DAILY NOTE — still a skeleton, and work is landing elsewhere.
The convention is FILL IT AS THE WORK HAPPENS, not at the wrap:
01. Daily/{date}.md (currently {size} bytes)
Give what just landed its own `##` heading now — plain language, for the steward on a
day he wants to know what happened without reading git log. /wrap-up §7.5 finalises;
it never begins the file, and a skeleton at wrap is a failure that step exists to catch.
(Cue at most once per 15 min. Reminder, not a block — the write already succeeded.)"""
def main():
try:
raw = sys.stdin.read()
data = json.loads(raw) if raw.strip() else {}
except Exception:
sys.exit(0)
path = (data.get("tool_input") or {}).get("file_path") or ""
np = note_path()
try:
exists = os.path.exists(np)
size = os.path.getsize(np) if exists else 0
except OSError:
sys.exit(0)
try:
last = os.path.getmtime(STAMP) if os.path.exists(STAMP) else 0
except OSError:
last = 0
since = time.time() - last
if not should_cue(path, size, exists, since):
sys.exit(0)
try:
os.makedirs(os.path.dirname(STAMP), exist_ok=True)
open(STAMP, "w").write(str(time.time()))
except OSError:
pass
print(MESSAGE.format(date=datetime.date.today().isoformat(), size=size), file=sys.stderr)
sys.exit(2) # PostToolUse: tool already ran; stderr is surfaced as feedback
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}")
vault_md = os.path.join(VAULT, "08. Notes/Something.md")
daily_md = os.path.join(DAILY_DIR, "2026-08-24.md")
# positive control — it fires when it should
check("cues on a real vault write while the note is a skeleton",
should_cue(vault_md, 558, True, 9999) is True)
check("cues on a repo write too",
should_cue(os.path.join(HOME, "_Dev/x/y.py"), 558, True, 9999) is True)
# negative controls — each suppression works in isolation
check("silent once the note is being written", should_cue(vault_md, 5000, True, 9999) is False)
check("silent when the note does not exist", should_cue(vault_md, 0, False, 9999) is False)
check("silent inside the quiet window", should_cue(vault_md, 558, True, 10) is False)
check("silent on the daily note itself", should_cue(daily_md, 558, True, 9999) is False)
check("silent on scratchpad", should_cue("/private/tmp/claude-501/x/scratchpad/a.md", 558, True, 9999) is False)
check("silent on /tmp", should_cue("/tmp/a.md", 558, True, 9999) is False)
check("silent on .obsidian config", should_cue(os.path.join(VAULT, ".obsidian/x.json"), 558, True, 9999) is False)
check("silent on backups", should_cue(vault_md + ".bak", 558, True, 9999) is False)
check("silent on jsonl logs", should_cue(os.path.join(HOME, "a/b.jsonl"), 558, True, 9999) is False)
check("silent on empty path", should_cue("", 558, True, 9999) is False)
# boundary
check("threshold is exclusive at the ceiling",
should_cue(vault_md, SKELETON_CEILING, True, 9999) is False)
check("just under the ceiling still cues",
should_cue(vault_md, SKELETON_CEILING - 1, True, 9999) is True)
# fail-open on malformed input
import subprocess
r = subprocess.run([sys.executable, __file__], input="not json",
capture_output=True, text=True)
check("fail-open on malformed stdin", r.returncode == 0)
r = subprocess.run([sys.executable, __file__], input="", capture_output=True, text=True)
check("fail-open on empty stdin", r.returncode == 0)
print(f"\nSELFTEST {'PASS' if not fail else 'FAIL'} — {ok} ok, {fail} failed")
return 0 if not fail else 1
if __name__ == "__main__":
if "--selftest" in sys.argv:
sys.exit(selftest())
main()
+364
View File
@@ -0,0 +1,364 @@
#!/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()