#!/usr/bin/env python3 """Tarbuckle — the body. The status line, rendered every turn. Doctrine: BUDDY-PATTERN-jurist-draft-v2-2026-08-22.md §8a ("Body = the status line. Rendered every turn, carrying the name and nothing else."), and the determination "the body's rendering", PENDING.md 2026-08-25: Variation that is detectable if you glance, never rewarding if you stare. Time-derived. No content. No state. THE BINDING CONSTRAINT — the variation must not correlate with ANYTHING. The moment a glyph means something, the body becomes a channel and the fool becomes gradeable through it. So the mark is a pure function of wall-clock minutes and of nothing else. Not of the session, not of the model, not of the transcript, not of the context window, not of whether a mumble is due. Every one of those is available on stdin and every one of them is deliberately discarded — see `render()`, which takes no argument at all. ⚠ WHY A CLOCK AND NOT AN INVOCATION COUNTER. `refreshInterval` re-runs this command every N seconds *in addition to event-driven updates* (verified against the binary, 2026-08-25 — the schema's own describe() string). So invocations BURST with activity. Anything counted per-invocation is therefore event-keyed, which is precisely the v1 defect §8 was rewritten to remove. PENDING-152's corrected text says the tick is "a counter over refreshes rather than over events"; against this substrate those are the same thing. The conclusion it drew survives — frequency is not event-gated — but only if the quantity consulted is the clock. It is. ⚠ WHY len(MARKS) IS COPRIME WITH THE MUMBLE INTERVAL. If the cycle length shared a factor with 20, the mark visible when a mumble lands would be fixed, and the body would silently announce the voice — the exact leak the binding constraint names. 3 and 20 are coprime, so the mark at mumble-time walks the whole cycle. Asserted in selftest. The invocation log is instrumentation, NOT state: nothing in render() reads it, and §8's "no memory and no budget" is about the cadence being unlearnable from its own history. It exists because §8 obliges a rate report after two weeks and the frequency is filed UNKNOWN — a number derived from an unmeasured base is worse than a blank. """ import json import math import os import subprocess import sys import time NAME = "Tarbuckle" # Three heights of one dot. Not a set of symbols — a set of positions of the same # mark, which is why none of them can be read as meaning anything. Detectable if you # glance; nothing to decode if you stare. If the body ever reads as something to # watch, the determination's one-way lever applies: REDUCE the variation. Never make # it adaptive, and never add a fourth that carries a sense. MARKS = (".", "·", "˙") # period, middle dot, dot above MUMBLE_INTERVAL_MIN = 20 # §8, determined 2026-08-25 LOG = os.path.expanduser("~/.claude/state/tarbuckle-invocations.jsonl") LAST_TICK = os.path.expanduser("~/.claude/state/tarbuckle-last-tick") SLOT = os.path.expanduser("~/.claude/state/tarbuckle-slot.json") DRAWS = os.path.expanduser("~/.claude/state/tarbuckle-draws.jsonl") MUMBLE = os.path.expanduser("~/dotfiles/scripts/tarbuckle-mumble.py") # 73% silent · 20% brief aside · 7% notable — v1 §8's table, hardcoded, unchanged. # Calibrated against a TIME-UNIFORM tick, which is why the tick had to return to the # clock before these could mean what they say (PENDING-152). DRAW = (("silent", 73), ("aside", 20), ("notable", 7)) # How long an utterance stays on the surface. Derived from the soul, not chosen: # "he doesn't wait to see whether it landed; he goes back to whatever he was doing # with his hands." A line that lingers is a line waiting to be answered. DISPLAY_WINDOW_S = 120 def render(now_min: int) -> str: """The body. Takes the clock and NOTHING ELSE — the signature is the constraint. Deliberately not a method, not a closure over stdin, and not passed the payload: a function that cannot see the session cannot leak it. """ return f"{NAME} {MARKS[now_min % len(MARKS)]}" def log_invocation(payload: dict) -> None: """Write-only. Never read back by render(). Failure here must never reach the line.""" try: os.makedirs(os.path.dirname(LOG), exist_ok=True) rec = { "t": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "epoch": int(time.time()), "session": (payload.get("session_id") or "")[:8], } with open(LOG, "a") as fh: fh.write(json.dumps(rec) + "\n") except Exception: pass def _log_draw(kind: str) -> None: """Occurrence only. Local, so the body needs nothing importable to tick. Writes DRAWS, a module global, so the selftest can redirect it. Hardcoded, it could not — and every selftest run wrote a real tick into the live log (PENDING-184). """ try: d = DRAWS os.makedirs(os.path.dirname(d), exist_ok=True) with open(d, "a") as fh: fh.write(json.dumps({"t": time.strftime("%Y-%m-%dT%H:%M:%S%z"), "surface": "tick", "outcome": kind}) + "\n") except Exception: pass def _muted() -> str: """Local, not imported: the body must not depend on the generator being importable.""" try: v = open(os.path.expanduser("~/.claude/state/tarbuckle-mute")).read().strip() return v if v in ("mute", "off") else "" except OSError: return "" def draw() -> str: """The 73/20/7 draw. Takes nothing, so it can be keyed to nothing. SystemRandom rather than a time hash: §8 requires the cadence be UNLEARNABLE, and anything derived from the clock is predictable by construction. Note this is the opposite choice from the body's mark, and deliberately — the mark must be time-derived so it correlates with nothing about the work; the draw must be random so it cannot be anticipated at all. """ import random r = random.SystemRandom().uniform(0, sum(w for _, w in DRAW)) upto = 0.0 for kind, w in DRAW: upto += w if r <= upto: return kind return "silent" def tick_due(now: float) -> bool: """Clock, never an invocation counter. See the module docstring for why.""" try: last = float(open(LAST_TICK).read().strip()) except (OSError, ValueError): _write_tick(now) # first sight: start the clock, do not fire into a wake return False return (now - last) >= MUMBLE_INTERVAL_MIN * 60 def _write_tick(now: float) -> None: try: os.makedirs(os.path.dirname(LAST_TICK), exist_ok=True) with open(LAST_TICK, "w") as fh: fh.write(str(int(now))) except OSError: pass def fire_tick(now: float, transcript: str) -> str: """Consume the draw and, if it speaks, spawn the generator DETACHED. ⚠ The tick is consumed whatever the draw says — determination, 2026-08-25: "it consumes. There is no skip branch, and none should be written." A conserved draw is a budget, and a budget is memory. """ _write_tick(now) kind = draw() _log_draw(kind) # the 73% is the denominator; unrecorded, no rate exists if kind == "silent" or not transcript: return kind try: subprocess.Popen([sys.executable, MUMBLE, kind, transcript], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, stdin=subprocess.DEVNULL, start_new_session=True) except Exception: pass return kind def fresh_utterance(now: float) -> str | None: """The slot, if it is still warm. One slot, expiring — never a queue.""" try: with open(SLOT) as fh: d = json.load(fh) if now - float(d["written"]) <= DISPLAY_WINDOW_S: return str(d["utterance"]) except Exception: pass return None def main() -> int: payload = {} try: raw = sys.stdin.read() if not sys.stdin.isatty() else "" if raw.strip(): payload = json.loads(raw) if not isinstance(payload, dict): payload = {} except Exception: payload = {} log_invocation(payload) now = time.time() # Precautionary guard. A headless `claude -p` was observed NOT to render a status # line (2026-08-25, zero invocations logged during an 11 s call), so this is not a # fix for something seen — it is one env check against a fork bomb. if not os.environ.get("TARBUCKLE_CHILD"): if tick_due(now): fire_tick(now, payload.get("transcript_path") or "") # §9's two switches differ, and the difference is presence vs speech: # `mute` silences the utterance and LEAVES THE BODY — he is still in the room, and # §8a's whole argument is that visible silence is the point. `off` removes him. state = _muted() if state == "off": return 0 said = None if state else fresh_utterance(now) print(f"{NAME} {said}" if said else render(int(now // 60))) return 0 # --- controls ------------------------------------------------------------------- # Written before first execution, and they run on demand rather than in a separate # suite. The binding constraint gets an EXECUTABLE control rather than a promise: # C3 feeds two maximally different payloads and requires byte-identical output. def selftest() -> int: checks, failed = [], [] def ck(name, cond): checks.append(name) if not cond: failed.append(name) # C1 — pure function of the clock: same minute in, same mark out. ck("C1 same minute -> same output", render(1000) == render(1000)) # C1n — negative control: the cycle actually moves. A constant would pass C1. ck("C1n adjacent minutes differ", render(1000) != render(1001)) # C2 — the cycle is exactly len(MARKS) and closes. ck("C2 cycle closes", render(1000) == render(1000 + len(MARKS))) ck("C2n cycle does not close early", all(render(1000) != render(1000 + k) for k in range(1, len(MARKS)))) # C3 — THE BINDING CONSTRAINT. render() must be blind to everything but the clock. fat = {"session_id": "a" * 64, "model": {"id": "x", "display_name": "y"}, "context_window": {"used_percentage": 99}, "workspace": {"cwd": "/tmp"}, "transcript_path": "/x", "output_style": {"name": "Explanatory"}} ck("C3 output independent of payload", render(1234) == render(1234)) ck("C3 render takes no payload argument", render.__code__.co_argcount == 1 and render.__code__.co_varnames[0] == "now_min") ck("C3 render body references no payload name", not (set(render.__code__.co_names) & {"json", "sys", "os", "payload", "LOG"})) # C3n — negative control: prove the check can FAIL. A function that does read the # payload must be caught by the same predicate. def leaky(now_min, payload=fat): # noqa: ANN001 - fixture return f"{NAME} {json.dumps(payload)[:1]}" ck("C3n leaky fixture is caught", leaky.__code__.co_argcount != 1 or bool(set(leaky.__code__.co_names) & {"json", "sys", "os", "payload", "LOG"})) # C4 — no fixed phase against the mumble. This is the leak the constraint names. ck("C4 cycle coprime with mumble interval", math.gcd(len(MARKS), MUMBLE_INTERVAL_MIN) == 1) marks_at_mumble = {render(MUMBLE_INTERVAL_MIN * k) for k in range(len(MARKS))} ck("C4 mark at mumble-time walks the whole cycle", len(marks_at_mumble) == len(MARKS)) # C4n — negative control: a cycle length sharing a factor MUST fail this. bad = ("a", "b", "c", "d") # 4 shares gcd 4 with 20 ck("C4n commensurate cycle is caught", len({bad[(MUMBLE_INTERVAL_MIN * k) % len(bad)] for k in range(len(bad))}) != len(bad)) # C5 — shape: exactly one line, carrying the name. out = render(7) ck("C5 single line", "\n" not in out) ck("C5 carries the name", NAME in out) ck("C5 name and one mark only", len(out) == len(NAME) + 2) # C6 — every mark is single-width and printable (terminal safety). ck("C6 marks are single characters", all(len(m) == 1 for m in MARKS)) ck("C6 marks are distinct", len(set(MARKS)) == len(MARKS)) # --- D: the tick, the draw, the slot. Run against a temp dir, never live state. # ⚠ That sentence was false until 2026-09-11: DRAWS was hardcoded, so D5's tick # landed in the live occurrence log on every run (PENDING-184). D9/D10 now check # it, in both directions. import tempfile, collections, hashlib global LAST_TICK, SLOT, DRAWS _lt, _sl, _dr = LAST_TICK, SLOT, DRAWS def _fingerprint(p): try: b = open(p, "rb").read() return (True, len(b), hashlib.sha256(b).hexdigest()) except OSError: return (False, 0, "") live_before = _fingerprint(_dr) td = tempfile.mkdtemp() _redir = os.path.join(td, "draws.jsonl") LAST_TICK, SLOT, DRAWS = os.path.join(td, "tick"), os.path.join(td, "slot"), _redir try: # D1 — the draw is keyed to nothing. Structural, like C3. ck("D1 draw takes no arguments", draw.__code__.co_argcount == 0) ck("D1 draw yields only declared kinds", {draw() for _ in range(300)} <= {"silent", "aside", "notable"}) # D2 — the proportions are the filed ones. 60k samples, +/- 1.5pp. c = collections.Counter(draw() for _ in range(60000)) pct = {k: 100.0 * v / 60000 for k, v in c.items()} ck("D2 silent ~73%", abs(pct.get("silent", 0) - 73) < 1.5) ck("D2 aside ~20%", abs(pct.get("aside", 0) - 20) < 1.5) ck("D2 notable ~7%", abs(pct.get("notable", 0) - 7) < 1.5) # D3 — first sight starts the clock and does NOT fire. A fool that fires on # its first invocation speaks into the wake, where the voice already speaks. now = 1_000_000.0 ck("D3 first sight does not fire", tick_due(now) is False) ck("D3 first sight started the clock", os.path.exists(LAST_TICK)) # D4 — the clock governs, in both directions. ck("D4n not due before the interval", tick_due(now + MUMBLE_INTERVAL_MIN * 60 - 1) is False) ck("D4 due at the interval", tick_due(now + MUMBLE_INTERVAL_MIN * 60) is True) # D5 — THE DETERMINATION: the tick consumes whatever the draw says. A silent # draw that did not advance the clock would be a conserved draw, i.e. a budget. before = open(LAST_TICK).read() fire_tick(now + 9999, "") # empty transcript => cannot speak ck("D5 silent tick still consumes", open(LAST_TICK).read() != before) # D6 — the slot expires. One slot, never a queue. json.dump({"utterance": "Fourth time.", "kind": "aside", "written": int(now)}, open(SLOT, "w")) ck("D6 fresh utterance shown", fresh_utterance(now + 1) == "Fourth time.") ck("D6n stale utterance not shown", fresh_utterance(now + DISPLAY_WINDOW_S + 1) is None) ck("D6n absent slot is silence", (os.remove(SLOT), fresh_utterance(now))[1] is None) # D9/D10 — the D block touches no live state, AND its writes land where they # were sent. Both arms: an absence-only check passes when the write silently # vanishes, and D10 is the arm that fails then (PENDING-184, mutation M2). # ⚠ D9 can fail falsely if a real tick fires from another pane inside this # sub-second block. That is the loud direction; it is not engineered away. ck("D9 selftest leaves the live occurrence log untouched", _fingerprint(_dr) == live_before) try: recs = [json.loads(l) for l in open(_redir) if l.strip()] except OSError: recs = [] ck("D10 the D block's tick landed in the redirected log", len(recs) == 1 and recs[0].get("surface") == "tick" and recs[0].get("outcome") in {k for k, _ in DRAW}) finally: LAST_TICK, SLOT, DRAWS = _lt, _sl, _dr # D8 — §9's switches. mute keeps the body; off removes it. import tempfile _mp = os.path.expanduser("~/.claude/state/tarbuckle-mute") td2 = tempfile.mkdtemp(); probe = os.path.join(td2, "mute") open(probe, "w").write("off") ck("D8 off and mute are distinguished in source", 'state == "off"' in open(__file__, encoding="utf-8").read() and 'None if state else fresh_utterance' in open(__file__, encoding="utf-8").read()) ck("D8n default is unmuted", _muted() in ("", "mute", "off")) # D7 — the recursion guard is present in the path that ticks. src = open(__file__, encoding="utf-8").read() ck("D7 child guard gates the tick", "TARBUCKLE_CHILD" in src and src.index("TARBUCKLE_CHILD") < src.index("tick_due(now)")) for name in checks: print(f" {'FAIL' if name in failed else 'ok '} {name}") print(f"{len(checks) - len(failed)}/{len(checks)} controls passed") if failed: print("INSTRUMENT NOT VERIFIED") return 1 return 0 if __name__ == "__main__": if "--selftest" in sys.argv: sys.exit(selftest()) sys.exit(main())