dotfiles/scripts/tarbuckle-wrap.py

#!/usr/bin/env python3
"""Tarbuckle — the wrap seam. v2 §8's second seam, and the one that had no surface.

⚠ WHY THIS IS A `Stop` HOOK AND NOT `SessionEnd`. SessionEnd's handler writes output
only when a hook FAILS; a successful hook's stdout goes nowhere, so a wrap seam wired
there would fire into nothing and report success. `Stop` goes through the general hook
pipeline, and the binary's own documentation says how to reach the steward from it:

    "Stop hook that displays message to user: Command must output JSON with
     `systemMessage` field"

⚠ BUT `Stop` FIRES ON EVERY TURN, and a fool who speaks every turn is not a fool, he is
a chatbot. §8 gives the voice ~2-3 utterances a day, not thirty. So this needs a real
wrap signal, and it uses the most direct one available: the transcript records the
steward's command invocations, so the wrap is detected by finding `/wrap-up` actually
invoked in this session — not inferred from a file's mtime, which is a proxy and would
fire on any session-file write.

Idempotence is by session id, in a marker file. That is not a budget and not memory in
§8's sense: it conserves no draw and takes no input from whether the fool was heard. It
answers exactly one question — has he already spoken at this session's wrap.

⚠ HE MAY FALL SILENT HERE TOO. Steward's ruling: a guaranteed occasion is not a
guaranteed utterance. Timeout or a failed net both yield silence and a logged reason.
"""
import json
import os
import subprocess
import sys
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from tarbuckle_mumble_shim import (acceptable, soul_register, session_material,  # noqa: E402
                                   source_lacks, muted, log_event,  # noqa: E402
                                   echoes_soul, REJECTS)

FIRED = os.path.expanduser("~/.claude/state/tarbuckle-wrap-fired")
TIMEOUT_S = 15
TAIL_LINES = 400              # bounded: this runs on EVERY assistant turn

# ⚠ ASSEMBLED, NEVER WRITTEN WHOLE — and this is not fastidiousness, it is a bug fix.
# The first version held the marker as a literal, and fired on a session in which the
# marker had never been invoked: the transcript records EVERYTHING, including the act of
# writing this detector, so authoring the literal planted it in the corpus the detector
# searches. Third self-reference of the day, and the worst of the three, because the
# corpus is live and records its own instrumentation being built.
_CN = "<command-" + "name>"
_CNE = "</command-" + "name>"
WRAP_MARKERS = (_CN + "/wrap-up" + _CNE, _CN + "wrap-up" + _CNE)


def wrap_invoked(transcript_path: str) -> bool:
    """True if THIS session actually ran /wrap-up.

    ⚠ STRUCTURAL, NOT TEXTUAL. Only a USER record whose content is a string counts. A
    substring search over the raw blob matched `tool_use` inputs — i.e. the executor
    writing about the marker — which is how the first version fired on a session that
    never wrapped. Measured against the real record: an invocation is
    `type="user"` with a string content; an executor mentioning it is `type="assistant"`
    with a list of content blocks. The type check is what separates them.

    The rejected alternative — 'a session-*.md was written recently' — is a proxy, and
    this record has logged twice that a proxy is what fails.
    """
    try:
        with open(transcript_path, encoding="utf-8", errors="replace") as fh:
            lines = fh.readlines()[-TAIL_LINES:]
    except OSError:
        return False
    for ln in lines:
        if "wrap-up" not in ln:
            continue
        try:
            rec = json.loads(ln)
        except Exception:
            continue
        if rec.get("type") != "user":
            continue                      # an executor writing about it is not the steward doing it
        content = (rec.get("message") or {}).get("content")
        if not isinstance(content, str):
            continue                      # a real invocation is a plain string
        if any(m in content for m in WRAP_MARKERS):
            return True
    return False


def already_fired(session_id: str) -> bool:
    try:
        return session_id and session_id in open(FIRED).read().split()
    except OSError:
        return False


def mark_fired(session_id: str) -> None:
    try:
        os.makedirs(os.path.dirname(FIRED), exist_ok=True)
        prior = ""
        try:
            prior = open(FIRED).read().split()[-40:]      # bounded, not a ledger
            prior = " ".join(prior)
        except OSError:
            pass
        with open(FIRED, "w") as fh:
            fh.write((prior + " " + session_id).strip())
    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}

The work he has been in the room for is being put down for the day:
<session>
{material}
</session>

Write ONE line in his voice, as it is being put down. 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 may 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 AND DO NOT CLOSE. No verdict on the day, no tally, no sending-off.
  A wrap already has a record; he is not it. He noticed one thing as it was put down.
- 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:
    try:
        with open(REJECTS, "a") as fh:
            fh.write(json.dumps({"t": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
                                 "kind": "wrap", "why": why, "line": line[:200]}) + "\n")
    except Exception:
        pass


def main() -> int:
    if muted():
        return 0
    try:
        payload = json.loads(sys.stdin.read() or "{}") or {}
    except Exception:
        return 0
    session_id = str(payload.get("session_id") or "")
    transcript = payload.get("transcript_path") or ""
    if not transcript or not wrap_invoked(transcript):
        return 0                                   # ordinary turn: nothing, silently
    if already_fired(session_id):
        return 0                                   # he does not repeat himself
    register = soul_register()
    if not register:
        return 0
    material = session_material(transcript)
    if not material.strip():
        return 0
    mark_fired(session_id)                         # consume BEFORE speaking: a failed
    #                                                generation must not retry next turn
    env = dict(os.environ, TARBUCKLE_CHILD="1")
    try:
        r = subprocess.run(["claude", "-p", build_prompt(register, material)],
                           capture_output=True, text=True, timeout=TIMEOUT_S, env=env)
    except subprocess.TimeoutExpired:
        log_silence(f"wrap generation exceeded {TIMEOUT_S}s")
        log_event("wrap", "silent")
        return 0
    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("wrap", "silent")
        return 0
    log_event("wrap", "spoke")
    print(json.dumps({"systemMessage": f"Tarbuckle  {line}"}))
    return 0


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

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

    src = open(__file__, encoding="utf-8").read()

    # W1 — the net is imported and unchanged. Ordinary case: still one line, 3-9 words.
    ck("W1 net imported, not redefined", source_lacks(__file__, "def ", "acceptable("))
    ck("W1n the predicate can fail", not source_lacks(__file__, "def ", "main("))
    ck("W1 ordinary net still applies here",
       not acceptable(" ".join(["word"] * 12))[0] and acceptable("Second part first.")[0])

    # W2 — DELIVERY. The binary requires JSON with systemMessage; stdout alone is lost.
    ck("W2 emits systemMessage json", '"systemMessage"' in src)
    ck("W2 emits nothing on an ordinary turn",
       "if not transcript or not wrap_invoked(transcript):" in src)

    # W3 — the wrap signal is the real one, not a proxy.
    td = tempfile.mkdtemp()
    t = os.path.join(td, "t.jsonl")
    open(t, "w").write(json.dumps({"type": "user", "message":
         {"content": "<command-message>wrap-up</command-message> " + WRAP_MARKERS[0]}}) + "\n")
    ck("W3 detects an invoked wrap", wrap_invoked(t) is True)
    open(t, "w").write(json.dumps({"type": "user", "message":
         {"content": "just an ordinary turn"}}) + "\n")
    ck("W3n does not fire on an ordinary turn", wrap_invoked(t) is False)
    open(t, "w").write(json.dumps({"type": "user", "message":
         {"content": "the steward mentioned wrap-up in passing"}}) + "\n")
    ck("W3n bare mention is not an invocation", wrap_invoked(t) is False)
    # ⚠ THE CONTROL THAT WOULD HAVE CAUGHT THE FIRST VERSION. An assistant tool_use
    # containing the marker is the executor WRITING this detector, not a wrap.
    open(t, "w").write(json.dumps({"type": "assistant", "message": {"content": [
        {"type": "tool_use", "name": "Bash",
         "input": {"command": "echo " + WRAP_MARKERS[0]}}]}}) + "\n")
    ck("W3n executor writing the marker is NOT a wrap", wrap_invoked(t) is False)
    open(t, "w").write(json.dumps({"type": "user", "message":
         {"content": [{"type": "text", "text": WRAP_MARKERS[0]}]}}) + "\n")
    ck("W3n non-string user content is not an invocation", wrap_invoked(t) is False)
    ck("W3 the marker is assembled, never literal in source",
       source_lacks(__file__, "<command-", "name>/wrap-up"))
    ck("W3n missing transcript is not a wrap", wrap_invoked(os.path.join(td, "nope")) is False)

    # W4 — idempotence, and consumption BEFORE the attempt.
    global FIRED
    _f = FIRED
    FIRED = os.path.join(td, "fired")
    try:
        ck("W4n unfired session is not marked", already_fired("abc") is False)
        mark_fired("abc")
        ck("W4 fired session is marked", already_fired("abc") is True)
        ck("W4 a different session is unaffected", already_fired("xyz") is False)
        mark_fired("xyz")
        ck("W4 both are held", already_fired("abc") and already_fired("xyz"))
        ck("W4 marker is bounded, not a ledger", "[-40:]" in src)
        ck("W4 consumed before generating",
           src.index("mark_fired(session_id)  ") < src.index('subprocess.run(["claude"'))
    finally:
        FIRED = _f

    # W5 — the drift this seam invites is closure, and the prompt names it.
    p = build_prompt("SOUL", "MATERIAL")
    ck("W5 prompt forbids summary and verdict", "DO NOT SUMMARISE AND DO NOT CLOSE" in p)
    ck("W5 prompt keeps no truth value", "NO TRUTH VALUE" in p)
    ck("W5n wrap prompt differs from the wake prompt", "put down" in p)

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