← back to Terminal Status
stop_verdict: mechanical done/waiting dot verdict for the Stop hook (TK-11921)
f5399f8e3411c98a099d08a02abf518c3f882b42 · 2026-09-18 14:27:06 -0700 · Steve Abrams
Pure decision module + CLI: reads the Stop-hook stdin JSON, tails the
transcript, prints ONE JSON verdict (never paints). 36 fixtures, 34 unit
tests, --selftest, DOT_FLOOR_FAULT negative seam.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YT3be6iadPzjKJDiaTasEE
Files touched
A stop_verdict.pyA test_stop_verdict.py
Diff
commit f5399f8e3411c98a099d08a02abf518c3f882b42
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Fri Sep 18 14:27:06 2026 -0700
stop_verdict: mechanical done/waiting dot verdict for the Stop hook (TK-11921)
Pure decision module + CLI: reads the Stop-hook stdin JSON, tails the
transcript, prints ONE JSON verdict (never paints). 36 fixtures, 34 unit
tests, --selftest, DOT_FLOOR_FAULT negative seam.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YT3be6iadPzjKJDiaTasEE
---
stop_verdict.py | 478 +++++++++++++++++++++++++++++++++++++++++++++++++++
test_stop_verdict.py | 282 ++++++++++++++++++++++++++++++
2 files changed, 760 insertions(+)
diff --git a/stop_verdict.py b/stop_verdict.py
new file mode 100644
index 0000000..03cf946
--- /dev/null
+++ b/stop_verdict.py
@@ -0,0 +1,478 @@
+#!/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 []:
+ if isinstance(block, dict) and block.get("type") == "text" and block.get("text", "").strip():
+ parts.append(block["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)
+
+ 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(win)
+ 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(win):
+ if not WAIT_NEGATION.search(win[:m.start()]):
+ hits[rule] = m.group(0)
+ break
+ m = P1.search(win)
+ if m:
+ hits["P1"] = m.group(0)
+ m = S2.search(win)
+ if m:
+ hits["S2"] = m.group(0)
+ t = S2_TIME.search(win)
+ if t:
+ times["S2"] = t.group(1)
+ if base_variant == "monitoring":
+ hits["S3"] = "base variant monitoring"
+
+ guarded = guarded_text(win)
+ 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)
+ 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)
+ if "S2" in hits or "S3" in hits:
+ 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))
+
+
+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)
+ header, _ = load_fixture(path)
+ 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 = {}
+ path = stdin.get("transcript_path") if isinstance(stdin, dict) else None
+ decision = verdict_for_transcript(path, base=args.base, base_variant=args.variant)
+ print(json.dumps(decision, ensure_ascii=False))
+ return 0
+
+
+if __name__ == "__main__":
+ sys.exit(main())
diff --git a/test_stop_verdict.py b/test_stop_verdict.py
new file mode 100644
index 0000000..16f0515
--- /dev/null
+++ b/test_stop_verdict.py
@@ -0,0 +1,282 @@
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+import stop_verdict as sv
+
+
+HERE = Path(__file__).resolve().parent
+
+
+class MatchTests(unittest.TestCase):
+ """One-line closing texts -> which signals fire (and which must not)."""
+
+ CASES = [
+ # text, expected signal ids (subset that must be present), forbidden ids
+ ("Done.", {"D2"}, set()),
+ ("done", {"D2"}, set()),
+ ("✓ done", {"D2"}, set()),
+ ("✅ Done for now.", {"D2"}, set()),
+ ("Understood — done.", {"D2"}, set()),
+ ("All set.", {"D2"}, set()),
+ ("Shipped.", {"D2"}, set()),
+ ("Landed.", {"D2"}, set()),
+ ("Wrapped up here.", {"D2"}, set()),
+ ("Fixed — reload the page", {"D2"}, set()),
+ ("Finished the migration across all 12 repos.", {"D3"}, set()),
+ ("Completed the sweep; nothing else needed.", {"D3"}, set()),
+ ("Done with the review, report is above.", {"D3"}, set()),
+ ("Done 13:42", {"D1"}, set()),
+ ("Finished at 09:15.", {"D1"}, set()),
+ ("13:42 — done.", {"D1"}, set()),
+ ("not done yet", set(), {"D1", "D2", "D3"}),
+ ("This is not finished.", set(), {"D1", "D2", "D3"}),
+ ("`tk done`", set(), {"D1", "D2", "D3"}),
+ ("Run `tk done TK-1` next.", set(), {"D1", "D2", "D3"}),
+ ("tk done TK-11921", set(), {"D1", "D2", "D3"}),
+ ("once that's done I'll rerun the canary.", set(), {"D1", "D2", "D3"}),
+ ("when done", set(), {"D1", "D2", "D3"}),
+ ("Not done until 14:00.", set(), {"D1", "D2", "D3"}),
+ ("It isn't complete.", set(), {"D1", "D2", "D3"}),
+ ("You've hit your weekly limit", set(), {"D1", "D2", "D3", "W1", "W2", "Q1"}),
+ ("| lane | done |\n|---|---|", set(), {"D1", "D2", "D3"}),
+ ("```sh\ntk done TK-1\n```", set(), {"D1", "D2", "D3", "O1"}),
+ ("tell me when Developer Mode shows", {"W1"}, set()),
+ ("Waiting on you to approve the DNS change.", {"W1"}, set()),
+ ("Nothing gated, nothing waiting on you.", set(), {"W1", "W2"}),
+ ("Not waiting on you for anything.", set(), {"W1"}),
+ ("I'll continue once you've pasted it.", {"W1"}, set()),
+ ("Ping me when the cert lands.", {"W1"}, set()),
+ ("I'll hold here.", {"W2"}, set()),
+ ("Standing by.", {"W2"}, set()),
+ ("Let me know if you want the full table.", {"W2"}, set()),
+ ("Nothing left in flight.", {"W2"}, set()),
+ ("Parked — ping me when ready", {"P1", "W1"}, set()),
+ ("🩷 parked", {"P1"}, set()),
+ ("Should I proceed?", {"Q1"}, set()),
+ ("Which one do you want?)", {"Q1"}, set()),
+ ('Do you want both?"', {"Q1"}, set()),
+ ("Is that right? No — it is fine.", set(), {"Q1"}),
+ ("NEXT CHECK 14:30", {"S2"}, set()),
+ ("monitoring · next 08:40", {"S2"}, set()),
+ ("next wake at 07:05 sharp", {"S2"}, set()),
+ ("Backfill at 40%. NEXT CHECK 14:30", {"S2"}, {"D1"}),
+ ("```bash\n! sudo pmset -c sleep 0\n```", {"O1"}, set()),
+ ("! ssh root@45.61.58.125 'pm2 reload x'", {"O1"}, set()),
+ ("Warning! this is loud", set(), {"O1"}),
+ ("Memo: ~/.claude/yolo-queue/pending-approval/2026-09-18-x.md", {"U1"}, set()),
+ ("Filed to pending-approval/_done/2026-09-18-x.md", set(), {"U1"}),
+ ("Filed to pending-approval/_never/x.md", set(), {"U1"}),
+ ("Drafted to pending-approval; APPROVE / REVISE / BLOCK is yours.", {"U2"}, set()),
+ ]
+
+ def test_signal_table(self):
+ for text, want, forbid in self.CASES:
+ hits, _ = sv.match(text)
+ got = set(hits)
+ with self.subTest(text=text):
+ self.assertTrue(want <= got, "missing %s in %s" % (want - got, sorted(got)))
+ self.assertFalse(forbid & got, "forbidden %s fired" % sorted(forbid & got))
+
+ def test_weekly_limit_matches_nothing(self):
+ hits, times = sv.match("You've hit your weekly limit")
+ self.assertEqual(hits, {})
+ self.assertEqual(times, {})
+
+ def test_d1_captures_time(self):
+ hits, times = sv.match("Done 13:42")
+ self.assertIn("D1", hits)
+ self.assertEqual(times["D1"], "13:42")
+
+ def test_s2_time_is_the_next_check_not_the_done_time(self):
+ _, times = sv.match("Done 14:02. NEXT CHECK 14:30")
+ self.assertEqual(times["S2"], "14:30")
+ self.assertEqual(times["D1"], "14:02")
+
+ def test_s3_comes_from_base_variant(self):
+ hits, _ = sv.match("Still importing.", base_variant="monitoring")
+ self.assertIn("S3", hits)
+ self.assertNotIn("S3", sv.match("Still importing.")[0])
+
+ def test_window_is_last_six_clean_lines(self):
+ text = "Done.\n" + "\n".join("line %d" % i for i in range(6))
+ self.assertNotIn("D2", sv.match(text)[0])
+ self.assertIn("D2", sv.match("\n".join("line %d" % i for i in range(5)) + "\nDone.")[0])
+
+ def test_clean_view_strips_fences_code_tables_rules_emphasis(self):
+ raw = "```\n! paste\n```\n`tk done`\n| a | done |\n─────\n★ Insight ─────\n**Done** _now_"
+ self.assertEqual(sv.clean_view(raw), "Done now")
+
+
+class DecideTests(unittest.TestCase):
+ def d(self, hits, times=None, **kw):
+ kw.setdefault("base", "green")
+ return sv.decide(hits, times or {}, **kw)
+
+ def h(self, *ids):
+ return {i: i for i in ids}
+
+ def test_needs_steve_base_only_gets_stopped_variant(self):
+ for base in ("yellow", "purple", "orange"):
+ d = self.d(self.h("D2", "O1"), base=base, base_label="TK-1 · why")
+ self.assertEqual((d["verdict"], d["action"], d["variant"], d["rule"]),
+ (base, "set-variant", "stopped", "BASE_NEEDS_STEVE"))
+
+ def test_parked_base_only_repaints(self):
+ for base in ("pink", "lightblue"):
+ d = self.d(self.h("D2", "O1"), base=base)
+ self.assertEqual((d["verdict"], d["action"], d["rule"]), (base, "repaint", "BASE_PINK"))
+
+ def test_order_paste_beats_everything(self):
+ d = self.d(self.h("O1", "U1", "Q1", "W1", "S2", "P1", "D1", "W2"), {"D1": "13:42"})
+ self.assertEqual((d["verdict"], d["rule"], d["label"]), ("orange", "O1", "PASTE waiting ·auto"))
+
+ def test_order_gated_beats_question(self):
+ d = self.d(self.h("U1", "Q1", "D2"))
+ self.assertEqual((d["verdict"], d["rule"], d["label"]), ("purple", "U1", "GATED ·auto"))
+ self.assertEqual(self.d(self.h("U2", "Q1"))["rule"], "U2")
+
+ def test_order_question_beats_waiting(self):
+ d = self.d(self.h("Q1", "W1", "D2"))
+ self.assertEqual((d["verdict"], d["rule"], d["label"]), ("yellow", "Q1", "DIRECTION? ·auto"))
+
+ def test_order_waiting_on_steve_beats_monitoring_and_done(self):
+ d = self.d(self.h("W1", "S2", "D1"), {"D1": "13:42", "S2": "14:00"})
+ self.assertEqual((d["verdict"], d["rule"], d["label"]),
+ ("lightblue", "W1", "WAITING ON STEVE ·auto"))
+
+ def test_next_check_beats_done(self):
+ d = self.d(self.h("S2", "D1", "D2"), {"S2": "14:30", "D1": "14:02"})
+ self.assertEqual((d["verdict"], d["action"], d["variant"], d["rule"], d["label"], d["time"]),
+ ("green", "set", "monitoring", "S2", "WAITING · next 14:30 ·auto", "14:30"))
+ self.assertEqual(d["spinning"], ["S2"])
+
+ def test_s3_keeps_base_label_when_no_time(self):
+ d = self.d(self.h("S3"), base_variant="monitoring", base_label="TK-1 · monitoring · next 09:10")
+ self.assertEqual((d["rule"], d["variant"], d["label"]),
+ ("S3", "monitoring", "TK-1 · monitoring · next 09:10"))
+ self.assertEqual(self.d(self.h("S3"), base_variant="monitoring")["label"], "WAITING · next check ·auto")
+
+ def test_explicit_park_beats_waiting_on_steve(self):
+ d = self.d(self.h("P1", "W1", "W2"))
+ self.assertEqual((d["verdict"], d["rule"], d["label"]), ("pink", "P1", "PARKED ·auto"))
+
+ def test_parked_beats_hold_and_done(self):
+ d = self.d(self.h("P1", "W2", "D2"))
+ self.assertEqual((d["verdict"], d["rule"], d["label"]), ("pink", "P1", "PARKED ·auto"))
+
+ def test_done_labels(self):
+ self.assertEqual(self.d(self.h("D1", "D2"), {"D1": "13:42"})["label"], "DONE 13:42 ·auto")
+ self.assertEqual(self.d(self.h("D2", "D3"))["label"], "DONE ·auto")
+ self.assertEqual(self.d(self.h("D3"))["rule"], "D3")
+ self.assertEqual(self.d(self.h("W2"))["label"], "HOLDING ·auto")
+
+ def test_done_beats_silent_spinning_task(self):
+ d = self.d(self.h("D2"), spinning_ids=["bva1m1uhx"])
+ self.assertEqual((d["verdict"], d["action"], d["rule"], d["spinning"]), ("pink", "set", "D2", ["S1"]))
+
+ def test_silent_spinning_task_alone_stays_green(self):
+ d = self.d({}, spinning_ids=["bva1m1uhx"])
+ self.assertEqual((d["verdict"], d["action"], d["rule"], d["spinning"]), ("green", "repaint", "S1", ["S1"]))
+
+ def test_idle_pink_toggle(self):
+ on = self.d({}, idle_pink=True)
+ self.assertEqual((on["verdict"], on["action"], on["rule"], on["label"]), ("pink", "set", "IDLE", "IDLE ·auto"))
+ off = self.d({}, idle_pink=False)
+ self.assertEqual((off["verdict"], off["action"], off["rule"], off["label"]), ("green", "repaint", "IDLE", ""))
+
+ def test_unknown_base_is_evaluated_like_green(self):
+ self.assertEqual(self.d(self.h("D2"), base="unknown")["verdict"], "pink")
+ self.assertEqual(self.d(self.h("D2"), base="none")["verdict"], "pink")
+
+ def test_output_contract_keys(self):
+ d = self.d(self.h("D2"))
+ self.assertEqual(set(d), {"verdict", "action", "variant", "label", "rule", "snippet",
+ "spinning", "time", "base", "base_variant", "idle_pink", "signals"})
+ self.assertLessEqual(len(d["snippet"]), 80)
+
+ def test_labels_respect_engine_valid_label(self):
+ import terminal_status as ts
+ d = self.d(self.h("S3"), base_variant="monitoring", base_label="x" * 600)
+ self.assertTrue(ts.valid_label(d["label"]))
+
+
+class TranscriptTests(unittest.TestCase):
+ def rec(self, kind, text, **extra):
+ content = text if kind == "user" else [{"type": "text", "text": text}]
+ return json.dumps({"type": kind, "isSidechain": False, "message": {"content": content}, **extra})
+
+ def test_closing_text_is_after_last_user_line(self):
+ lines = [self.rec("assistant", "Done."), self.rec("user", "more"), self.rec("assistant", "Working on it?")]
+ d = sv.verdict_from_lines(lines, base="green", base_variant="", base_label="", idle_pink=True)
+ self.assertEqual(d["rule"], "Q1")
+
+ def test_sidechain_and_noise_records_are_skipped(self):
+ lines = [self.rec("user", "go"), self.rec("assistant", "Done 13:42.", isSidechain=True),
+ json.dumps({"type": "attachment", "attachment": {}}), self.rec("assistant", "Report above.")]
+ d = sv.verdict_from_lines(lines, base="green", base_variant="", base_label="", idle_pink=True)
+ self.assertEqual(d["rule"], "IDLE")
+
+ def test_malformed_lines_are_skipped_and_no_text_is_none(self):
+ d = sv.verdict_from_lines(["{nope", "", "[1]"], base="green", base_variant="", base_label="", idle_pink=True)
+ self.assertEqual((d["verdict"], d["action"], d["rule"]), ("none", "none", "NO_TEXT"))
+
+ def test_missing_transcript_is_benign(self):
+ d = sv.verdict_for_transcript("/nonexistent/x.jsonl", base="green")
+ self.assertEqual((d["verdict"], d["action"], d["rule"]), ("none", "none", "NO_TRANSCRIPT"))
+
+ def test_tail_lines_reads_only_the_end(self):
+ with tempfile.NamedTemporaryFile("w", suffix=".jsonl", delete=False) as fh:
+ for i in range(5000):
+ fh.write(self.rec("assistant", "line %d" % i) + "\n")
+ self.addCleanup(os.unlink, fh.name)
+ tail = sv.tail_lines(fh.name, 3)
+ self.assertEqual(len(tail), 3)
+ self.assertIn("line 4999", tail[-1])
+
+ def test_spinning_pairs_bg_start_with_notification(self):
+ start = json.dumps({"type": "user", "message": {"content": [{"type": "tool_result", "content":
+ "Command running in background with ID: abc123. You will be notified."}]}})
+ agent = json.dumps({"type": "user", "message": {"content": [{"type": "tool_result", "content":
+ [{"type": "text", "text": "Async agent launched.\nagentId: a9816d499be105861 (internal)"}]}]}})
+ notif = json.dumps({"type": "user", "message": {"content":
+ "<task-notification>\n<task-id>abc123</task-id>\n<status>completed</status>"}})
+ self.assertEqual(sv.spinning_tasks(sv.parse_records([start, agent])), ["abc123", "a9816d499be105861"])
+ self.assertEqual(sv.spinning_tasks(sv.parse_records([start, agent, notif])), ["a9816d499be105861"])
+
+
+class CliTests(unittest.TestCase):
+ def run_cli(self, *args, stdin="", env=None):
+ return subprocess.run([sys.executable, str(HERE / "stop_verdict.py"), *args], input=stdin,
+ capture_output=True, text=True, env={**os.environ, **(env or {})})
+
+ def test_selftest_passes(self):
+ r = self.run_cli("--selftest")
+ self.assertEqual(r.returncode, 0, r.stdout + r.stderr)
+ self.assertNotIn("FAIL", r.stdout)
+ self.assertRegex(r.stdout, r"\d+ fixtures, 0 failed")
+
+ def test_stdin_missing_transcript_exits_zero(self):
+ r = self.run_cli(stdin='{"transcript_path": "/nonexistent/x.jsonl"}')
+ self.assertEqual(r.returncode, 0)
+ self.assertEqual(json.loads(r.stdout)["rule"], "NO_TRANSCRIPT")
+
+ def test_fixture_flag_with_base_override_never_touches_engine(self):
+ r = self.run_cli("--test", str(HERE / "tests/stop-verdict/01-done-plain.jsonl"),
+ env={"TERMINAL_STATUS_PS_TIMEOUT": "0.001"})
+ self.assertEqual(r.returncode, 0, r.stderr)
+ self.assertEqual(json.loads(r.stdout)["label"], "DONE ·auto")
+
+ def test_injected_fault_goes_red(self):
+ r = self.run_cli(stdin="{}", env={"DOT_FLOOR_FAULT": "detector"})
+ self.assertNotEqual(r.returncode, 0)
+ self.assertIn("injected detector fault", r.stderr)
+ self.assertEqual(r.stdout, "")
+
+
+if __name__ == "__main__":
+ unittest.main()
← de433d8 auto-data-snapshot: 2026-09-18T14:19:46 (36 data files) — te
·
back to Terminal Status
·
integrations: sync dot-floor.sh with the live TK-11921 Stop ec7cc19 →