dotfiles/scripts/tarbuckle-mumble.py

#!/usr/bin/env python3
"""Tarbuckle — the mumble generator. Runs DETACHED; never inside the status line.

Doctrine: v2 §8 (three tiers), §9 (one line, filed nowhere), and PENDING-152
AMENDMENT 6, which settles the three-way separation:

    Material — the live session.   Register — the soul, §7.   Occasion — the refresh tick.
    "The model supplies the words, in the soul's register, about the session."

⚠ THE UTTERANCE MUST HAVE NO TRUTH VALUE. The criterion is NOT "checkable in
principle" — it is whether an ADJUDICATION PATH EXISTS (AMENDMENT 6, superseding the
executor's own earlier test). "PENDING-4 has been open since April" has a path: open
the file. Once a path exists the trio walks it, the fool acquires a truth value, and it
is a checker again — which §2 says means the design has failed. "Scoring without
signal" has no path: it is a gesture at a shape, not a claim. That is the whole safety.

⚠ THE REGISTER IS NOT DUPLICATED HERE. The soul is read from its filed artifact at
run time. Pasting it into this file would create a parallel version of a governed
record, which is the context-rot failure CLAUDE.md names outright. If the soul cannot
be read, NOTHING IS GENERATED — honest degradation, not a fallback voice.

Latency measured 2026-08-25: ~11 s for a headless call. That is why this is detached
and why the status line never waits on it.
"""
import json
import os
import re
import subprocess
import sys
import time

SOUL = os.path.expanduser(
    "~/dotfiles/claude/governance/fool/seed/FOOL-SOUL-2026-08-25.md")
SLOT = os.path.expanduser("~/.claude/state/tarbuckle-slot.json")
REJECTS = os.path.expanduser("~/.claude/state/tarbuckle-rejects.jsonl")

MIN_WORDS, MAX_WORDS = 3, 9        # observed Thistleweld register: three to nine words

# From the soul's own "What he never does", mechanically enforced. The model is asked
# for the register; this is the net under it. A violation yields SILENCE, never a
# repaired line — rewriting the fool's words would make the executor its editor.
BANNED = (
    r"\bshould\b", r"\btry\b", r"\bmust\b",                    # advice
    r"\bgone\b", r"\bif only\b", r"\bused to be\b", r"\bmissing\b",  # vocabulary of lack
    r"\bwe\b",                                                  # never says 'we' about the work
    r"\?",                                                      # never asks
)


MUTE = os.path.expanduser("~/.claude/state/tarbuckle-mute")


def muted() -> str:
    """'' | 'mute' | 'off'. §9: "mute / off available at all times."

    ⚠ Read by EVERY surface, and it is the one piece of state the fool is permitted to
    be steered by — because §9 says muting is never a fault, and a mute the fool could
    ignore is not a mute. It conserves nothing and no utterance depends on it having
    been set before: it is a switch, not a memory.
    """
    try:
        v = open(MUTE).read().strip()
        return v if v in ("mute", "off") else ""
    except OSError:
        return ""


def source_lacks(path: str, *parts: str) -> bool:
    """True if the joined needle does NOT appear in `path`.

    ⚠ THE NEEDLE IS ASSEMBLED FROM PARTS, AND THAT IS THE ENTIRE POINT. A control that
    writes its needle as a literal PLANTS that literal in the very file it searches, so
    it can only ever fail. That bug was written twice in one session — the second time
    by the party who had just fixed the first, minutes earlier, while watching for it.
    Correcting it a second time by hand would have been the same one-off; this is the
    mechanism, so the shape cannot be written again by accident.
    """
    return "".join(parts) not in open(path, encoding="utf-8").read()


def soul_register() -> str | None:
    """The soul, verbatim, from the filed artifact. None if unreadable."""
    try:
        body = open(SOUL, encoding="utf-8").read()
    except OSError:
        return None
    m = re.search(r"```markdown\n(.*?)\n```", body, re.S)
    return m.group(1) if m else None


def session_material(transcript_path: str, budget: int = 6000) -> str:
    """The tail of the live session. Bounded, and tool output is dropped.

    Deliberately NOT the docket. AMENDMENT 6: the docket has a forum and the session
    does not, which is why the session is safe material and PENDING.md is not.
    """
    try:
        lines = open(transcript_path, encoding="utf-8", errors="replace").readlines()
    except OSError:
        return ""
    out = []
    for ln in reversed(lines[-400:]):
        try:
            rec = json.loads(ln)
        except Exception:
            continue
        if rec.get("type") not in ("user", "assistant"):
            continue
        msg = rec.get("message") or {}
        content = msg.get("content")
        text = ""
        if isinstance(content, str):
            text = content
        elif isinstance(content, list):
            text = " ".join(c.get("text", "") for c in content
                            if isinstance(c, dict) and c.get("type") == "text")
        text = text.strip()
        if not text:
            continue
        out.append(f"{rec['type']}: {text[:600]}")
        if sum(len(s) for s in out) > budget:
            break
    return "\n".join(reversed(out))


def build_prompt(kind: str, register: str, material: str) -> str:
    weight = ("Something at the shape of the work, not its detail."
              if kind == "notable" else
              "An ordinary passing remark. Small.")
    return f"""You are writing ONE line as Tarbuckle. His character, filed and unalterable:

{register}

Here is the tail of the session he is in the room for:
<session>
{material}
</session>

{weight}

Write ONE line in his voice. Absolute constraints:
- Between {MIN_WORDS} and {MAX_WORDS} words. One clause. Present tense. Flat, no lift.
- IT MUST HAVE NO TRUTH VALUE. Nobody must be able to open a file and check it, agree
  with it, or refute it. Put two things next to each other so a shape shows. Do not
  state a fact about the work, the record, the code, or the docket.
- No advice, no questions, no warning of consequences, no explanation, no second line.
- Never the word 'we'. No vocabulary of lack.
- Do not name files, items, numbers of open things, or anything with an address.

Output the line and nothing else. No quotes, no preamble."""


def acceptable(line: str, max_words: int = MAX_WORDS,
               one_line: bool = True) -> tuple[bool, str]:
    """The net. Widening is possible but must be PASSED EXPLICITLY at the call site.

    ⚠ Steward's ruling, 2026-08-25: "If the seam voice needs a wider net because seams
    warrant more than nine words, widen it explicitly and say so, but NEVER relax
    silence-on-violation." So the defaults are the filed ones, a caller that wants more
    room has to say so in its own source, and no caller can turn the net off — the
    clauses below are not parameterised, and deliberately.
    """
    if not line:
        return False, "empty"
    if one_line and "\n" in line.strip():
        return False, "not one line"
    n = len(line.split())
    if not (MIN_WORDS <= n <= max_words):
        return False, f"{n} words"
    for pat in BANNED:
        if re.search(pat, line, re.I):
            return False, f"banned {pat}"
    return True, ""


def main() -> int:
    if muted():
        return 1
    kind = sys.argv[1] if len(sys.argv) > 1 else "aside"
    transcript = sys.argv[2] if len(sys.argv) > 2 else ""
    register = soul_register()
    if not register:
        return 1                      # no soul, no voice. Deliberately no fallback.
    material = session_material(transcript)
    if not material.strip():
        return 1
    env = dict(os.environ, TARBUCKLE_CHILD="1")   # precautionary; see body script
    try:
        r = subprocess.run(["claude", "-p", build_prompt(kind, register, material)],
                           capture_output=True, text=True, timeout=120, env=env)
    except Exception:
        return 1
    line = (r.stdout or "").strip().strip('"').strip()
    ok, why = acceptable(line)
    if not ok:
        try:
            with open(REJECTS, "a") as fh:
                fh.write(json.dumps({"t": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
                                     "kind": kind, "why": why, "line": line[:200]}) + "\n")
        except Exception:
            pass
        return 1                      # silence. The draw was already consumed.
    try:
        os.makedirs(os.path.dirname(SLOT), exist_ok=True)
        with open(SLOT, "w") as fh:
            json.dump({"utterance": line, "kind": kind, "written": int(time.time())}, fh)
    except Exception:
        return 1
    return 0


def selftest() -> int:
    checks, failed = [], []

    def ck(name, cond):
        checks.append(name)
        if not cond:
            failed.append(name)

    # A1 — the register is READ, not duplicated. One canonical source.
    reg = soul_register()
    ck("A1 soul readable from its filed artifact", bool(reg))
    ck("A1 soul is the real thing", bool(reg) and "Tarbuckle" in reg and "SUCCESSION" in reg)
    # ⚠ The needle is BUILT rather than written, because the first version of this
    # control failed against itself: the literal phrase it searched for was placed in
    # the file BY the search. Same class as the hand-typed link canary whose only
    # finding was the pattern inside its own specification.
    ck("A1n this file does not contain a copy of the soul",
       source_lacks(__file__, "registrar rather than ", "a guardian"))
    ck("A1nn the predicate can fail",
       not source_lacks(__file__, "soul_", "register"))

    # A2 — the acceptability net. Positive AND negative controls on every clause.
    ck("A2 accepts an in-register line", acceptable("Fourth time. First one was better.")[0])
    ck("A2 accepts a short collision", acceptable("Two names, one thing.")[0])
    ck("A2n rejects too few words", not acceptable("Yes.")[0])
    ck("A2n rejects too many words",
       not acceptable(" ".join(["word"] * (MAX_WORDS + 1)))[0])
    ck("A2n rejects advice", not acceptable("You should check that again now.")[0])
    ck("A2n rejects a question", not acceptable("How is that going for you?")[0])
    ck("A2n rejects vocabulary of lack", not acceptable("The third one is missing now.")[0])
    ck("A2n rejects 'we'", not acceptable("We did the second part first.")[0])
    ck("A2n rejects two lines", not acceptable("First line here.\nSecond line here.")[0])
    # The widening is explicit, and it widens ONLY what it names.
    ck("A2 widening admits a longer line",
       acceptable(" ".join(["word"] * 40), max_words=120)[0])
    ck("A2n widening does NOT relax the clauses",
       not acceptable("You should " + " ".join(["word"] * 40), max_words=120)[0])
    ck("A2n widening does NOT relax 'we'",
       not acceptable("We " + " ".join(["word"] * 40), max_words=120)[0])

    # A3 — the prompt carries the no-truth-value constraint verbatim, not by intention.
    p = build_prompt("aside", "SOUL", "MATERIAL")
    ck("A3 prompt states no truth value", "NO TRUTH VALUE" in p)
    ck("A3 prompt forbids addresses", "anything with an address" in p)
    ck("A3 prompt embeds the register", "SOUL" in p)
    ck("A3 notable differs from aside",
       build_prompt("notable", "S", "M") != build_prompt("aside", "S", "M"))

    # A4 — material is the session, never the docket.
    # Structural, not textual: the module's file constants are the only things it
    # opens, so assert none of them addresses the docket.
    paths = (SOUL, SLOT, REJECTS)
    ck("A4 material paths are session/soul only",
       not any(("PENDING" in q or "REVIEWED" in q) for q in paths))
    # A5 — the mute switch, and that every surface must be able to see it.
    import tempfile
    global MUTE
    _m = MUTE
    MUTE = os.path.join(tempfile.mkdtemp(), "mute")
    try:
        ck("A5n unmuted by default", muted() == "")
        open(MUTE, "w").write("mute")
        ck("A5 mute is read", muted() == "mute")
        open(MUTE, "w").write("off")
        ck("A5 off is read", muted() == "off")
        open(MUTE, "w").write("nonsense")
        ck("A5n a junk value is not a mute", muted() == "")
    finally:
        MUTE = _m
    ck("A4n the predicate can fail",
       any(("PENDING" in q) for q in paths + ("/x/PENDING.md",)))

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