dotfiles/scripts/tarbuckle-body.py

#!/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")
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 _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()
    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.
    import tempfile, collections
    global LAST_TICK, SLOT
    _lt, _sl = LAST_TICK, SLOT
    td = tempfile.mkdtemp()
    LAST_TICK, SLOT = os.path.join(td, "tick"), os.path.join(td, "slot")
    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)
    finally:
        LAST_TICK, SLOT = _lt, _sl

    # 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())