David F GliddenandClaude Opus 5
7948c0929b
`Stop` fires for `claude -p`, so the wrap hook re-entered inside the fool's own generator child. Measured over the fortnight to 2026-09-09: 3 of 10 wrap runs landed in a generation subprocess, and 18 s of the 80 s of blocking `Stop` — 30% of occasions, 23% of the time — went on occasions that were the end of a generation nobody was waiting on. The gate is one predicate against TARBUCKLE_CHILD, which the four seams already set on the spawn and which `tarbuckle-body.py:203` already reads as a fork-bomb guard. ⚠ The variable now carries two meanings that do not imply each other — DO NOT TICK in the body, DO NOT SPEAK OR WORK here — and both sites now say so, because the next reader would otherwise remove the coupling as arbitrary. The heartbeat write stays ABOVE the gate, deliberately: it answers "did the hook fire", and gating it would narrow the one artifact that separates a hook that never fires from a hook that fires and does nothing. W6/W6n are BEHAVIOURAL, not source strings — the source-string form of this kind of check is what passed vacuously for its whole life in the seam (PENDING-180). Verified by mutation, not by the green run: deleting the gate fails W6 and W6n; hoisting it above the heartbeat fails the heartbeat arm; nothing else moves. 28/28 controls. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01T1i5VRfHjD79hfaXjWsBXA
376 lines
18 KiB
Python
Executable File
376 lines
18 KiB
Python
Executable File
#!/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, log_rejection, REJECTS)
|
|
|
|
FIRED = os.path.expanduser("~/.claude/state/tarbuckle-wrap-fired")
|
|
# ⚠ ONE TIMESTAMP, OVERWRITTEN — not appended. The silent-net objection against this hook
|
|
# was that it writes nothing on an ordinary turn, so there is no evidence it is invoked at
|
|
# all. A heartbeat answers that; an append-only log of every turn would be noise and would
|
|
# become the ledger §9 forbids. Overwrite is the proportionate form: it proves liveness and
|
|
# remembers nothing.
|
|
HEARTBEAT = os.path.expanduser("~/.claude/state/tarbuckle-wrap-lastrun")
|
|
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
|
|
content = (rec.get("message") or {}).get("content")
|
|
# (a) the steward TYPES /wrap-up -> a user record whose content is a plain string
|
|
if rec.get("type") == "user" and isinstance(content, str):
|
|
if any(m in content for m in WRAP_MARKERS):
|
|
return True
|
|
# (b) the steward says "wrap" in prose and the EXECUTOR invokes the skill.
|
|
#
|
|
# ⚠ THIS BRANCH IS THE WHOLE BUG, FOUND BY THE WRAP IT WAS BUILT FOR. The first
|
|
# version had only (a), and on the day it shipped the steward wrote "then wrap"
|
|
# and the executor called the Skill tool: zero user-typed records, 29 assistant
|
|
# invocations, detector correctly returns False, fool silent. The detector was
|
|
# not broken — what it was built to detect is not how a wrap actually arrives.
|
|
#
|
|
# ⚠ AND THE FIX THAT MADE IT CORRECT IS WHAT BLINDED IT. Restricting to user
|
|
# records was the right answer to the self-reference bug (the executor's own
|
|
# tool_use inputs matched the literal). The same restriction excludes the real
|
|
# path. Recorded because "the correct fix caused the next failure" is not a
|
|
# shape the controls can see; it is PENDING-160's subject exactly.
|
|
#
|
|
# Structural, not textual: a `tool_use` block whose NAME is Skill and whose
|
|
# input names the wrap-up skill. A Bash command that merely echoes the string
|
|
# has name="Bash" and does not match.
|
|
if rec.get("type") == "assistant" and isinstance(content, list):
|
|
for blk in content:
|
|
if not isinstance(blk, dict) or blk.get("type") != "tool_use":
|
|
continue
|
|
if blk.get("name") != "Skill":
|
|
continue
|
|
if "wrap-up" in str(blk.get("input") or {}):
|
|
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.
|
|
- ⚠ THIS IS MECHANICALLY ENFORCED, NOT ADVISORY: any FOUR consecutive words you
|
|
share with a sample line are detected and the whole line is discarded. You are
|
|
then silent. Nothing is rewritten for you and nothing is retried.
|
|
- 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:
|
|
"""Routed through the canonical writer, which cannot record an accepted line."""
|
|
log_rejection(why, line, "wrap")
|
|
|
|
|
|
def main() -> int:
|
|
if muted():
|
|
return 0
|
|
try:
|
|
payload = json.loads(sys.stdin.read() or "{}") or {}
|
|
except Exception:
|
|
return 0
|
|
try:
|
|
os.makedirs(os.path.dirname(HEARTBEAT), exist_ok=True)
|
|
with open(HEARTBEAT, "w") as fh:
|
|
fh.write(time.strftime("%Y-%m-%dT%H:%M:%S%z"))
|
|
except OSError:
|
|
pass
|
|
|
|
# ⚠ `Stop` FIRES FOR `claude -p` TOO, so this hook re-enters inside the fool's own
|
|
# generator child. Measured over the fortnight to 2026-09-09: 3 of 10 wrap runs landed
|
|
# in a generation subprocess, and 18 s of the 80 s of blocking `Stop` went on occasions
|
|
# that were the end of a generation nobody was waiting on. The child already carries the
|
|
# variable — it is set on the spawn below, and by the three sibling seams.
|
|
# ⚠ THE HEARTBEAT STAYS ABOVE THIS GATE, deliberately: it answers "did the hook fire",
|
|
# and gating it would quietly narrow the one artifact that can tell a hook that never
|
|
# fires from a hook that fires and does nothing.
|
|
# ⚠ THE VARIABLE NOW MEANS TWO THINGS AND NEITHER IMPLIES THE OTHER. In
|
|
# `tarbuckle-body.py` it means DO NOT TICK — a fork-bomb guard on the status line.
|
|
# Here it means DO NOT SPEAK OR WORK — a generation's `Stop` is not a turn, and he has
|
|
# no business wrapping a session that is one of his own sentences. Whoever deletes one
|
|
# meaning must check the other: they are separate guards that share a name.
|
|
if os.environ.get("TARBUCKLE_CHILD"):
|
|
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:
|
|
# W6 rebinds these three for the duration of one control and restores them; Python
|
|
# requires the declaration ahead of first use, which W3's calls would otherwise be.
|
|
global wrap_invoked, muted, HEARTBEAT
|
|
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)
|
|
# W3b — THE REAL PATH: steward says "wrap" in prose, executor invokes the skill.
|
|
open(t, "w").write(json.dumps({"type": "assistant", "message": {"content": [
|
|
{"type": "tool_use", "name": "Skill", "input": {"skill": "wrap-up"}}]}}) + "\n")
|
|
ck("W3b executor Skill invocation IS a wrap", wrap_invoked(t) is True)
|
|
# and it must still exclude the executor merely writing about it
|
|
open(t, "w").write(json.dumps({"type": "assistant", "message": {"content": [
|
|
{"type": "tool_use", "name": "Bash",
|
|
"input": {"command": "echo wrap-up " + WRAP_MARKERS[0]}}]}}) + "\n")
|
|
ck("W3bn a Bash echo of the marker is NOT a wrap", wrap_invoked(t) is False)
|
|
open(t, "w").write(json.dumps({"type": "assistant", "message": {"content": [
|
|
{"type": "tool_use", "name": "Skill", "input": {"skill": "wake-up"}}]}}) + "\n")
|
|
ck("W3bn a different skill is NOT a wrap", wrap_invoked(t) is False)
|
|
ck("W4 heartbeat is overwritten, never appended",
|
|
'open(HEARTBEAT, "w")' in src)
|
|
|
|
# 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)
|
|
|
|
# W6 — `Stop` fires for `claude -p`, so this hook re-enters inside the fool's own
|
|
# generator. ⚠ BEHAVIOURAL, NOT A SOURCE STRING. The source-string form of exactly
|
|
# this kind of check is what passed vacuously for its whole life two files away
|
|
# (PENDING-180), and no string in this file can see whether the gate is reached
|
|
# BEFORE the work — which is the entire claim.
|
|
import io as _io
|
|
_wi, _mu, _hb = wrap_invoked, muted, HEARTBEAT
|
|
_env, _stdin, seen = os.environ.get("TARBUCKLE_CHILD"), sys.stdin, []
|
|
_hbt = os.path.join(td, "hb")
|
|
|
|
def _spy(path):
|
|
seen.append(path)
|
|
return False
|
|
|
|
_payload = json.dumps({"session_id": "w6", "transcript_path": t})
|
|
try:
|
|
# muted() is neutralised for the duration: a control whose outcome depends on a
|
|
# switch it does not name is not a control, and mute is a different question.
|
|
wrap_invoked, muted, HEARTBEAT = _spy, (lambda: False), _hbt
|
|
os.environ["TARBUCKLE_CHILD"] = "1"
|
|
sys.stdin = _io.StringIO(_payload)
|
|
ck("W6 a generation's Stop returns without reading the transcript",
|
|
main() == 0 and not seen)
|
|
ck("W6 the heartbeat is written above the gate, so liveness still reports",
|
|
os.path.exists(_hbt))
|
|
os.environ.pop("TARBUCKLE_CHILD", None)
|
|
sys.stdin = _io.StringIO(_payload)
|
|
main()
|
|
ck("W6n the predicate can fail: an ordinary Stop does read the transcript",
|
|
len(seen) == 1)
|
|
finally:
|
|
wrap_invoked, muted, HEARTBEAT = _wi, _mu, _hb
|
|
sys.stdin = _stdin
|
|
if _env is None:
|
|
os.environ.pop("TARBUCKLE_CHILD", None)
|
|
else:
|
|
os.environ["TARBUCKLE_CHILD"] = _env
|
|
|
|
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())
|