← back to Terminal Status
stop_verdict.py
536 lines
#!/usr/bin/env python3
"""Mechanical Stop-hook verdict: what colour should this tab's dot be? (TK-11921)
Pure decision module + CLI. Reads the Claude Code Stop-hook stdin JSON, tails the
session transcript, and prints ONE JSON decision. It NEVER paints -- the hook
(~/.claude/hooks/dot-floor.sh) dispatches exactly one engine call from `action`.
{"verdict": "orange|purple|yellow|lightblue|green|pink|none",
"action": "set|set-variant|repaint|none",
"variant": "" | "monitoring" | "stopped",
"label": "<label to paint>",
"rule": "O1|U1|U2|Q1|W1|S1|S2|S3|P1|W2|D1|D2|D3|IDLE|BASE_NEEDS_STEVE|BASE_PINK|
BASE_UNKNOWN|NO_TRANSCRIPT|NO_TEXT",
"snippet": "<=80 chars", "spinning": [...], "time": "HH:MM"|null,
"base": "...", "base_variant": "", "idle_pink": bool, "signals": {id: snippet}}
"""
import argparse
import glob
import json
import os
import re
import sys
from pathlib import Path
AUTO = " ·auto"
DEFAULT_TAIL = 120
SKIP_TYPES = {"attachment", "queue-operation", "file-history-snapshot", "bridge-session"}
NEEDS_STEVE = ("yellow", "purple", "orange")
PARKED = ("lightblue", "pink")
LABEL_MAX_BYTES = 500 # under terminal_status.valid_label's 512-byte ceiling
I = re.IGNORECASE
TIME = r"\b\d{1,2}:\d{2}\b"
DONE_WORD = r"\b(?:done|finished|complete[d]?)\b"
NEGATION = re.compile(
r"\b(?:not|isn'?t|aren'?t|wasn'?t|never|until|once|when|before|after|if|unless|yet)\b"
r"[^.\n]{0,40}" + DONE_WORD, I)
DONE_YET = re.compile(r"\b(?:done|finished)\s+yet\b", I)
TK_DONE = re.compile(r"\btk\s+done\b", I)
# Colons split sentences EXCEPT inside HH:MM, so "Done 13:42" survives as one sentence.
SENTENCE_SEP = re.compile(r"([.!?;]+|—|–|(?<!\d):(?!\d)|\n)")
D1_A = re.compile(DONE_WORD + r"[^\n]{0,25}?(" + TIME + ")", I)
D1_B = re.compile("(" + TIME + r")[^\n]{0,25}?" + DONE_WORD, I)
D2_PHRASES = {"done", "finished", "complete", "completed", "all set", "shipped",
"landed", "wrapped up", "fixed"}
D2_PREFIX = re.compile(r"^[\s✓✅☑\-\*•]+")
D2_SUFFIX = re.compile(r"\s+(?:for now|here)$", I)
D3_LINE_PREFIX = re.compile(r"^[\s\-\*•>#✓✅☑]+")
D3 = re.compile(r"^(?:done|finished|complete[d]?|all set|shipped|landed)\b", I)
O1 = re.compile(r"^\s*!\s+\S", re.M)
U1 = re.compile(r"pending-approval/(?!_done/|_never/)")
U2 = re.compile(r"drafted to pending-approval|APPROVE\s*/\s*REVISE\s*/\s*BLOCK")
W1 = re.compile(
r"waiting on you|until you (?:say|confirm|give|approve|run|paste)|once you(?:'ve| have)"
r"|tell me when|ping me when|when you(?:'ve| have) (?:run|pasted|signed|restarted)", I)
W2 = re.compile(
r"I'?ll hold|holding here|standing by|let me know (?:when|if)"
r"|nothing (?:left |else )?(?:to continue|pending|in flight)", I)
S2 = re.compile(r"NEXT CHECK|monitoring · next|next (?:check|wake|run)[^\n]{0,20}\d{1,2}:\d{2}", I)
S2_TIME = re.compile(r"(?:NEXT CHECK|monitoring · next|next (?:check|wake|run))[^\n]{0,20}?(" + TIME + ")", I)
P1 = re.compile(r"🩷|\bparked\b|tab (?:is )?pink", I)
Q1_TRAIL = re.compile(r"""[\s*)"»'’”]+$""")
WAIT_NEGATION = re.compile(r"\b(?:nothing|not|no|never|isn'?t|aren'?t|without)\b[^.\n;]{0,30}$", I)
BG_START = re.compile(r"Command running in background with ID:\s*([A-Za-z0-9_-]+)")
AGENT_START = re.compile(r"\bagentId:\s*([A-Za-z0-9_-]+)")
TASK_ID = re.compile(r"<task-id>\s*([A-Za-z0-9_-]+)\s*</task-id>")
BOX_CHARS = set("─═━│┃┌┐└┘├┤┬┴┼╔╗╚╝║╠╣╦╩╬█▀▄▌▐░▒▓★☆✦✧•·-=_~*#+ \t")
# ---------------------------------------------------------------- transcript
def tail_lines(path, n):
"""Last n lines of a (possibly huge) file without reading it all."""
with open(path, "rb") as fh:
fh.seek(0, os.SEEK_END)
end = fh.tell()
block, buf, pos = 65536, b"", end
while pos > 0 and buf.count(b"\n") <= n:
step = min(block, pos)
pos -= step
fh.seek(pos)
buf = fh.read(step) + buf
lines = buf.decode("utf-8", "replace").splitlines()
return lines[-n:]
def parse_records(lines):
"""JSON records worth reading; malformed lines and sidechain/noise types dropped."""
out = []
for line in lines:
line = line.strip()
if not line:
continue
try:
rec = json.loads(line)
except ValueError:
continue
if not isinstance(rec, dict) or rec.get("type") in SKIP_TYPES or rec.get("isSidechain"):
continue
out.append(rec)
return out
def _content_texts(rec):
"""Every text string inside a record's message content (text + tool_result blocks)."""
content = (rec.get("message") or {}).get("content")
if isinstance(content, str):
return [content]
texts = []
for block in content or []:
if not isinstance(block, dict):
continue
if block.get("type") == "text" and isinstance(block.get("text"), str):
texts.append(block["text"])
elif block.get("type") == "tool_result":
inner = block.get("content")
if isinstance(inner, str):
texts.append(inner)
else:
texts += [b.get("text", "") for b in inner or []
if isinstance(b, dict) and b.get("type") == "text"]
return texts
def closing_text(records):
last_user = -1
for i, rec in enumerate(records):
if rec.get("type") == "user":
last_user = i
parts = []
for rec in records[last_user + 1:]:
if rec.get("type") != "assistant":
continue
content = (rec.get("message") or {}).get("content")
if isinstance(content, str):
parts.append(content)
continue
for block in content or []:
text = block.get("text") if isinstance(block, dict) else None
if isinstance(block, dict) and block.get("type") == "text" and isinstance(text, str) and text.strip():
parts.append(text)
return "\n".join(parts)
def spinning_tasks(records):
"""Background ids started in the tail with no later completion notification."""
started, completed = {}, set()
for i, rec in enumerate(records):
if rec.get("type") != "user":
continue
for text in _content_texts(rec):
for m in TASK_ID.finditer(text):
if "<task-notification>" in text:
completed.add(m.group(1))
for rx in (BG_START, AGENT_START):
for m in rx.finditer(text):
started.setdefault(m.group(1), i)
return [tid for tid in started if tid not in completed]
# ---------------------------------------------------------------- text views
def clean_view(raw):
s = re.sub(r"```.*?(?:```|\Z)", "", raw, flags=re.S)
s = re.sub(r"`[^`\n]*`", "", s)
s = s.replace("’", "'")
lines = []
for line in s.splitlines():
stripped = line.strip()
if not stripped or stripped.startswith("|"):
continue
if set(stripped) <= BOX_CHARS:
continue
if "".join(c for c in stripped if c not in BOX_CHARS).strip().lower() == "insight":
continue
line = re.sub(r"\*\*|__", "", line)
line = re.sub(r"(?<![\w*_])[*_](?=\S)(.+?)(?<=\S)[*_](?![\w*_])", r"\1", line)
line = re.sub(r"^\s*#+\s*", "", line)
lines.append(line.rstrip())
return "\n".join(lines)
def window(clean, n=6):
lines = [l for l in clean.splitlines() if l.strip()]
return "\n".join(lines[-n:])
def snippet(text):
s = re.sub(r"\s+", " ", text.replace("\n", " ")).replace('"', "").replace("`", "").strip()
return s[:80]
# ---------------------------------------------------------------- signals
def _sentences(text):
"""Alternating [sentence, sep, sentence, sep, ...] so the text can be rebuilt."""
return SENTENCE_SEP.split(text)
def _negated(sentence):
return bool(NEGATION.search(sentence) or DONE_YET.search(sentence) or TK_DONE.search(sentence))
def guarded_text(text):
"""The text with every negation-tainted sentence blanked out (separators kept)."""
parts = _sentences(text)
return "".join(" " if i % 2 == 0 and _negated(p) else p for i, p in enumerate(parts))
def match(text, base_variant=""):
"""All text-only signals present in `text`. Returns (hits, times)."""
hits, times = {}, {}
raw = text
clean = clean_view(raw)
win = window(clean) # Q1 only: the question test looks at the LAST line
# every other text signal scans the WHOLE closing message (Steve: "done ALWAYS leaves green")
m = O1.search(raw)
if m:
hits["O1"] = raw[m.start():].split("\n", 1)[0]
m = U1.search(raw)
if m:
hits["U1"] = raw[max(0, m.start() - 40):m.end() + 40]
m = U2.search(clean)
if m:
hits["U2"] = m.group(0)
last = [l for l in win.splitlines() if l.strip()]
if last and Q1_TRAIL.sub("", last[-1]).endswith("?"):
hits["Q1"] = last[-1]
for rule, rx in (("W1", W1), ("W2", W2)):
for m in rx.finditer(clean):
if not WAIT_NEGATION.search(clean[max(0, m.start() - 60):m.start()]):
hits[rule] = m.group(0)
break
m = P1.search(clean)
if m:
hits["P1"] = m.group(0)
m = S2.search(clean)
if m:
hits["S2"] = m.group(0)
t = S2_TIME.search(clean)
if t:
times["S2"] = t.group(1)
if base_variant == "monitoring":
hits["S3"] = "base variant monitoring"
guarded = guarded_text(clean)
m = D1_A.search(guarded) or D1_B.search(guarded)
if m:
hits["D1"] = m.group(0)
times["D1"] = m.group(1)
parts = _sentences(guarded)
for sent in parts[0::2]:
core = D2_SUFFIX.sub("", D2_PREFIX.sub("", sent).strip()).strip().lower()
core = re.sub(r"\s+", " ", core)
if core in D2_PHRASES and len(core.split()) <= 4:
hits["D2"] = sent.strip()
break
for line in guarded.splitlines():
if D3.search(D3_LINE_PREFIX.sub("", line)):
hits["D3"] = line.strip()
break
return hits, times
# ---------------------------------------------------------------- decision
def fit_label(label):
label = "".join(c for c in label if ord(c) >= 32 and ord(c) != 127)
while len(label.encode("utf-8")) > LABEL_MAX_BYTES:
label = label[:-1]
return label
def _out(verdict, action, rule, *, variant="", label="", snip="", time=None, **extra):
return {"verdict": verdict, "action": action, "variant": variant,
"label": fit_label(label), "rule": rule, "snippet": snippet(snip),
"time": time, **extra}
def decide(hits, times, *, base="unknown", base_variant="", base_label="",
idle_pink=True, spinning_ids=()):
spinning = [r for r in ("S1", "S2", "S3")
if (r == "S1" and spinning_ids) or (r != "S1" and r in hits)]
meta = {"spinning": spinning, "base": base, "base_variant": base_variant,
"idle_pink": idle_pink, "signals": {k: snippet(v) for k, v in hits.items()}}
if base in NEEDS_STEVE:
return _out(base, "set-variant", "BASE_NEEDS_STEVE", variant="stopped",
label=base_label, **meta)
if base in PARKED:
return _out(base, "repaint", "BASE_PINK", label=base_label, **meta)
if "O1" in hits:
return _out("orange", "set", "O1", label="PASTE waiting" + AUTO, snip=hits["O1"], **meta)
for rule in ("U1", "U2"):
if rule in hits:
return _out("purple", "set", rule, label="GATED" + AUTO, snip=hits[rule], **meta)
if "Q1" in hits:
return _out("yellow", "set", "Q1", label="DIRECTION?" + AUTO, snip=hits["Q1"], **meta)
# D1 (done + a clock time) is Steve's strongest signal: it beats the bare word "parked"
# in prose (live 2026-09-18: a "Done 14:40" report that mentioned "parked" came out P1).
if "D1" in hits and "S2" not in hits and "W1" not in hits:
t = times["D1"]
return _out("pink", "set", "D1", label="DONE %s" % t + AUTO, snip=hits["D1"], time=t, **meta)
if "P1" in hits:
return _out("pink", "set", "P1", label="PARKED" + AUTO, snip=hits["P1"], **meta)
if "W1" in hits:
return _out("lightblue", "set", "W1", label="WAITING ON STEVE" + AUTO, snip=hits["W1"], **meta)
# S2 is a FRESH "next check" in this turn's text -> the session rescheduled itself,
# so it legitimately outranks done. S3 is only the PRIOR turn's stored variant; it is
# not evidence the session is still monitoring THIS turn, so an explicit done wins over
# it (memory done-word-always-leaves-green).
if "S2" in hits or ("S3" in hits and not ({"D1", "D2", "D3"} & hits.keys())):
rule = "S2" if "S2" in hits else "S3"
t = times.get("S2")
if t:
label = "WAITING · next %s" % t + AUTO
elif rule == "S3" and base_label:
label = base_label
else:
label = "WAITING · next check" + AUTO
return _out("green", "set", rule, variant="monitoring",
label=label, snip=hits[rule], time=t, **meta)
if "D1" in hits:
t = times["D1"]
return _out("pink", "set", "D1", label="DONE %s" % t + AUTO, snip=hits["D1"], time=t, **meta)
for rule in ("D2", "D3"):
if rule in hits:
return _out("pink", "set", rule, label="DONE" + AUTO, snip=hits[rule], **meta)
if "W2" in hits:
return _out("pink", "set", "W2", label="HOLDING" + AUTO, snip=hits["W2"], **meta)
if spinning_ids:
return _out("green", "repaint", "S1", snip="bg " + " ".join(spinning_ids), **meta)
if idle_pink:
return _out("pink", "set", "IDLE", label="IDLE" + AUTO, **meta)
return _out("green", "repaint", "IDLE", **meta)
# ---------------------------------------------------------------- base state
def read_base():
"""(base, variant, label) for the calling tty via the engine; 'unknown' on any failure."""
try:
os.environ.setdefault("TERMINAL_STATUS_PS_TIMEOUT",
os.environ.get("DOT_FLOOR_BASE_TIMEOUT", "12"))
sys.path.insert(0, str(Path(__file__).resolve().parent))
import terminal_status as ts
rows = ts.processes()
owner = ts.current_owner(rows)
record, _reason = ts.Store().read(owner)
if not record:
return "none", "", ""
return record.get("state", "unknown"), record.get("variant", ""), record.get("label", "")
except Exception:
return "unknown", "", ""
# ---------------------------------------------------------------- driver
def idle_pink_env():
return os.environ.get("DOT_FLOOR_IDLE_PINK", "1") not in ("0", "false", "no", "")
def tail_n():
try:
return max(1, int(os.environ.get("DOT_FLOOR_TAIL", DEFAULT_TAIL)))
except ValueError:
return DEFAULT_TAIL
def verdict_from_lines(lines, *, base, base_variant, base_label, idle_pink):
records = parse_records(lines)
text = closing_text(records)
if not text.strip():
return _out("none", "none", "NO_TEXT", spinning=[], base=base, base_variant=base_variant,
idle_pink=idle_pink, signals={})
hits, times = match(text, base_variant)
return decide(hits, times, base=base, base_variant=base_variant, base_label=base_label,
idle_pink=idle_pink, spinning_ids=spinning_tasks(records))
BG_DONE_STATES = {"completed", "done", "failed", "killed", "cancelled", "canceled", "stopped", "exited"}
def live_background_ids(background_tasks):
"""Ids of still-running tasks from the Stop hook's background_tasks list (any shape)."""
ids = []
for t in background_tasks or []:
if isinstance(t, str):
ids.append(t)
elif isinstance(t, dict):
status = str(t.get("status") or t.get("state") or "").lower()
if status and status in BG_DONE_STATES:
continue
ids.append(str(t.get("id") or t.get("task_id") or t.get("taskId") or "task"))
return ids
def verdict_from_stdin(stdin, *, base=None, base_variant=None, idle_pink=None):
"""Prefer the hook's own stdin fields (last_assistant_message, background_tasks) over the
transcript: at Stop time the final assistant record is often NOT yet flushed (TK-11921 live
miss), whereas stdin always carries the finished message. Returns None if stdin lacks it."""
if not isinstance(stdin, dict):
return None
text = stdin.get("last_assistant_message")
if not isinstance(text, str) or not text.strip():
return None
idle_pink = idle_pink_env() if idle_pink is None else idle_pink
if base is None:
base, base_variant, base_label = read_base()
else:
base_variant, base_label = base_variant or "", ""
hits, times = match(text, base_variant)
d = decide(hits, times, base=base, base_variant=base_variant, base_label=base_label,
idle_pink=idle_pink, spinning_ids=live_background_ids(stdin.get("background_tasks")))
d["source"] = "stdin"
return d
def verdict_for_transcript(path, *, base=None, base_variant=None, idle_pink=None):
idle_pink = idle_pink_env() if idle_pink is None else idle_pink
if not path or not os.path.isfile(path) or not os.access(path, os.R_OK):
return _out("none", "none", "NO_TRANSCRIPT", spinning=[], base="unknown",
base_variant="", idle_pink=idle_pink, signals={})
if base is None:
base, base_variant, base_label = read_base()
else:
base_variant, base_label = base_variant or "", ""
try:
lines = tail_lines(path, tail_n())
except OSError:
return _out("none", "none", "NO_TRANSCRIPT", spinning=[], base=base,
base_variant=base_variant, idle_pink=idle_pink, signals={})
return verdict_from_lines(lines, base=base, base_variant=base_variant,
base_label=base_label, idle_pink=idle_pink)
def load_fixture(path):
with open(path, encoding="utf-8") as fh:
lines = fh.read().splitlines()
header = json.loads(lines[0]) if lines else {}
if header.get("type") != "fixture":
raise ValueError("%s: first line must be a {\"type\":\"fixture\"} header" % path)
return header, lines[1:]
def run_fixture(path):
header, lines = load_fixture(path)
if header.get("fault"):
os.environ["DOT_FLOOR_FAULT"] = header["fault"]
try:
check_fault()
finally:
os.environ.pop("DOT_FLOOR_FAULT", None)
return verdict_from_lines(lines[-tail_n():], base=header.get("base", "green"),
base_variant=header.get("variant", ""),
base_label=header.get("label", ""),
idle_pink=header.get("idle_pink", True))
def fixture_dir():
return Path(__file__).resolve().parent / "tests" / "stop-verdict"
def selftest(out=sys.stdout):
failures = 0
files = sorted(glob.glob(str(fixture_dir() / "*.jsonl")))
if not files:
print("FAIL no fixtures under %s" % fixture_dir(), file=out)
return 1
for path in files:
name = os.path.basename(path)
try:
header, _ = load_fixture(path)
except (ValueError, KeyError, OSError) as exc:
failures += 1
print("FAIL %-40s bad fixture header: %s" % (name, exc), file=out)
continue
expect = header.get("expect", {})
try:
got = run_fixture(path)
except RuntimeError as exc:
got = {"raises": type(exc).__name__}
bad = {k: (expect[k], got.get(k)) for k in expect if got.get(k) != expect[k]}
if bad:
failures += 1
print("FAIL %-40s %s" % (name, " ".join("%s: want %r got %r" % (k, w, g)
for k, (w, g) in bad.items())), file=out)
else:
print("PASS %-40s %s" % (name, got.get("rule", "")), file=out)
print("%d fixtures, %d failed" % (len(files), failures), file=out)
return 1 if failures else 0
def check_fault():
if os.environ.get("DOT_FLOOR_FAULT") == "detector":
raise RuntimeError("injected detector fault")
def main(argv=None):
check_fault()
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
ap.add_argument("--test", metavar="FIXTURE", help="decide a fixture .jsonl; never touches the engine")
ap.add_argument("--selftest", action="store_true", help="run every tests/stop-verdict/*.jsonl")
ap.add_argument("--base", help="override the base colour (skips the engine read)")
ap.add_argument("--variant", default=None, help="override the base variant")
args = ap.parse_args(argv)
if args.selftest:
return selftest()
if args.test:
decision = run_fixture(args.test)
else:
try:
stdin = json.loads(sys.stdin.read() or "{}")
except ValueError:
stdin = {}
decision = verdict_from_stdin(stdin, base=args.base, base_variant=args.variant)
if decision is None:
path = stdin.get("transcript_path") if isinstance(stdin, dict) else None
decision = verdict_for_transcript(path, base=args.base, base_variant=args.variant)
decision["source"] = "transcript"
print(json.dumps(decision, ensure_ascii=False))
return 0
if __name__ == "__main__":
sys.exit(main())