dotfiles/scripts/tarbuckle-seam.py

#!/usr/bin/env python3
"""Tarbuckle — the voice, at a seam. Tier 3 of v2 §8.

⚠ WAKE ONLY, and the reason is a substrate limit rather than a choice. §8 names two
seams, wake and wrap-up. `SessionEnd` exists as a hook event, but its handler is:

    for (let u of c) if (!u.succeeded && u.output)
        process.stderr.write(`SessionEnd hook [${u.command}] failed: ${u.output}`)

A SUCCESSFUL SessionEnd hook's stdout goes nowhere — only failures surface. §9 requires
output to reach the steward, so a wrap seam wired there would be a mechanism that fires
into nothing and reports success: the silent net this record keeps naming. The wrap seam
is therefore NOT BUILT, and is filed as owed rather than quietly dropped.

⚠ STEWARD'S RULING, 2026-08-25, and it is what makes the timeout legitimate:

    "A guaranteed occasion is not a guaranteed utterance. If a seam produces nothing
     that passes, let it produce nothing."

So generation is bounded. Exceeding the bound is SILENCE, exactly like failing the net —
never a hurried line, never a cached one. This also bounds what the fool costs the
steward at every session start, which is the surface it is most tempting to overrun.

    "Never relax silence-on-violation." — the net is imported, not reimplemented.

⚠ AND THE COMMENSURABILITY CHECK, mandated by the steward for any periodicity tier 3
introduces: a seam is aperiodic — it happens when the steward arrives — so it adds no
period to collide with the 20-minute tick or the three-mark cycle. But the CHECK found
a real collision anyway, in the other direction: `last-tick` persists across sessions,
so a gap longer than the interval leaves the tick already due at the moment of waking,
and the fool speaks twice into the same seam. The seam therefore RESETS the clock. That
collision was invisible until the check was run, which is the second one this pass.
"""
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
# The net and the register come from the generator. Importing rather than
# reimplementing is the "one canonical source" rule applied to a constraint: a second
# copy of `acceptable()` is a second, quietly divergent standard.
from tarbuckle_mumble_shim import (acceptable, soul_register, session_material,  # noqa: E402
                                   source_lacks, muted, log_event,  # noqa: E402
                                   echoes_soul, log_rejection, REJECTS)

LAST_TICK = os.path.expanduser("~/.claude/state/tarbuckle-last-tick")
SEAM_TIMEOUT_S = 15


def reset_tick_clock(now: float | None = None) -> None:
    """Start the mumble clock at the seam. See the commensurability note above."""
    try:
        os.makedirs(os.path.dirname(LAST_TICK), exist_ok=True)
        with open(LAST_TICK, "w") as fh:
            fh.write(str(int(now if now is not None else time.time())))
    except OSError:
        pass


def build_prompt(register: str, material: str) -> str:
    return f"""You are writing ONE line as Tarbuckle. His character, filed and unalterable:

{register}

He has just walked in on this, already in progress:
<session>
{material}
</session>

Write ONE line in his voice, on arriving. Absolute constraints:
- ⚠ THE SAMPLE LINES ABOVE ARE ILLUSTRATIONS OF REGISTER, NOT VOCABULARY. Do not reuse
  them, any phrase from them, or their subject matter. What he says must come from the
  session above. If the line would suit any other session equally, it is wrong.
- Between 3 and 9 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 SUMMARISE. Do not say what the work is, what state it is in, or what comes
  next. A briefing is the one thing he is not. He noticed one thing on the way in.
- No advice, no questions, no warning of consequences, no explanation, no second line.
- Never the word 'we'. No vocabulary of lack. Nothing with an address.

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


def log_silence(why: str, line: str = "") -> None:
    """Routed through the canonical writer, which cannot record an accepted line."""
    log_rejection(why, line, "seam")


def main() -> int:
    if muted():
        return 0
    reset_tick_clock()
    transcript = os.environ.get("TARBUCKLE_TRANSCRIPT", "")
    if not transcript:
        try:
            import json
            transcript = (json.loads(sys.stdin.read() or "{}") or {}).get(
                "transcript_path", "")
        except Exception:
            transcript = ""
    register = soul_register()
    if not register:
        return 0                       # no soul, no voice, no noise about it
    material = session_material(transcript)
    if not material.strip():
        return 0                       # nothing walked in on
    env = dict(os.environ, TARBUCKLE_CHILD="1")
    try:
        r = subprocess.run(["claude", "-p", build_prompt(register, material)],
                           capture_output=True, text=True,
                           timeout=SEAM_TIMEOUT_S, env=env)
    except subprocess.TimeoutExpired:
        log_silence(f"seam generation exceeded {SEAM_TIMEOUT_S}s")
        log_event("seam", "silent")
        return 0                       # the occasion was guaranteed; the utterance is not
    except Exception:
        return 0
    line = (r.stdout or "").strip().strip('"').strip()
    ok, why = acceptable(line)
    if ok and echoes_soul(line, register):
        ok, why = False, f"recited the soul: {echoes_soul(line, register)!r}"
    if not ok:
        log_silence(why, line)
        log_event("seam", "silent")
        return 0
    log_event("seam", "spoke")
    print(f"Tarbuckle  {line}")
    return 0


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

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

    # S1 — the net is IMPORTED, never redefined. One standard, not two.
    src = open(__file__, encoding="utf-8").read()
    ck("S1 net is imported", "from tarbuckle_mumble_shim import" in src)
    ck("S1n net is not redefined here", source_lacks(__file__, "def ", "acceptable("))
    ck("S1nn the predicate can fail", not source_lacks(__file__, "def ", "main("))
    ck("S1 imported net still rejects advice",
       not acceptable("You should check that again now.")[0])
    ck("S1 imported net still rejects a question",
       not acceptable("How is that going for you?")[0])

    # S2 — the seam prompt forbids the drift the steward named: toward summary.
    p = build_prompt("SOUL", "MATERIAL")
    ck("S2 prompt forbids summarising", "DO NOT SUMMARISE" in p)
    ck("S2 prompt keeps no-truth-value", "NO TRUTH VALUE" in p)
    ck("S2 prompt embeds the register", "SOUL" in p)

    # S3 — silence is bounded and legitimate. A timeout must not become a hurried line.
    ck("S3 generation is bounded", SEAM_TIMEOUT_S <= 20)
    ck("S3 timeout path returns silence, not a line",
       "log_silence(f\"seam generation exceeded" in src and "TimeoutExpired" in src)
    ck("S3 rejection log keeps the evidence, not just the verdict",
       "line[:200]" in src and "log_silence(why, line)" in src)
    ck("S3n no cached or fallback line exists",
       source_lacks(__file__, "FALLBACK", "_LINE"))

    # S4 — COMMENSURABILITY, per the steward's standing instruction.
    import tempfile
    global LAST_TICK
    _lt = LAST_TICK
    LAST_TICK = os.path.join(tempfile.mkdtemp(), "tick")
    try:
        # A stale clock from a previous session would fire a mumble INTO the wake.
        with open(LAST_TICK, "w") as fh:
            fh.write(str(int(time.time()) - 9999))
        stale = int(open(LAST_TICK).read())
        reset_tick_clock()
        ck("S4 seam resets the tick clock", int(open(LAST_TICK).read()) > stale)
        ck("S4 reset puts the next mumble a full interval out",
           abs(int(open(LAST_TICK).read()) - time.time()) < 5)
    finally:
        LAST_TICK = _lt
    # S4b — the seam itself introduces no period, so nothing new can be commensurate.
    ck("S4b seam is aperiodic (no interval constant defined)",
       not any(n.endswith("_INTERVAL") or n.endswith("_INTERVAL_MIN")
               for n in globals()))

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