← back to Terminal Status
test_stop_verdict.py
338 lines
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_done_anywhere_in_closing_message(self):
# Steve: "the word done ALWAYS leaves green" — a done-signal on the FIRST line of a long
# report counts (live miss 2026-09-18: "Done 14:33" + 20 lines came out IDLE).
text = "Done.\n" + "\n".join("line %d" % i for i in range(6))
self.assertIn("D2", sv.match(text)[0])
self.assertIn("D2", sv.match("\n".join("line %d" % i for i in range(5)) + "\nDone.")[0])
def test_question_is_last_line_only(self):
text = "Is this right?\n" + "\n".join("line %d" % i for i in range(6))
self.assertNotIn("Q1", sv.match(text)[0])
self.assertIn("Q1", sv.match("line 1\nline 2\nIs this right?")[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, "")
class NullTextBlockTest(unittest.TestCase):
"""TK-11921 review: a text block whose text is null/non-string must not crash (exit-0 invariant)."""
def test_null_text_block_is_skipped(self):
import json as _json
lines = [
_json.dumps({"type": "user", "message": {"role": "user", "content": "go"}}),
_json.dumps({"type": "assistant", "message": {"content": [
{"type": "text", "text": None}, {"type": "text", "text": 42}, {"type": "text", "text": "Done."}]}}),
]
d = sv.verdict_from_lines(lines, base="green", base_variant="", base_label="", idle_pink=True)
self.assertEqual(d["verdict"], "pink")
self.assertIn(d["rule"], ("D2", "D3"))
class StdinFirstTests(unittest.TestCase):
"""TK-11921: the Stop hook stdin carries last_assistant_message + background_tasks."""
def test_message_from_stdin_beats_missing_transcript(self):
d = sv.verdict_from_stdin({"last_assistant_message": "All landed.\n\nDone 14:40.",
"background_tasks": [], "transcript_path": "/nonexistent"},
base="green", base_variant="", idle_pink=True)
self.assertEqual((d["verdict"], d["rule"], d["source"]), ("pink", "D1", "stdin"))
def test_live_background_task_keeps_green(self):
d = sv.verdict_from_stdin({"last_assistant_message": "Kicked off the build; results will land shortly.",
"background_tasks": [{"id": "b1", "status": "running"}]},
base="green", base_variant="", idle_pink=True)
self.assertEqual((d["verdict"], d["rule"]), ("green", "S1"))
d2 = sv.verdict_from_stdin({"last_assistant_message": "Kicked off the build; results will land shortly.",
"background_tasks": [{"id": "b1", "status": "completed"}]},
base="green", base_variant="", idle_pink=True)
self.assertEqual(d2["rule"], "IDLE")
def test_done_with_time_beats_parked_word(self):
d = sv.verdict_from_stdin({"last_assistant_message": "Done 14:40. The detector scans for done, parked and waiting signals.",
"background_tasks": []}, base="green", base_variant="", idle_pink=True)
self.assertEqual((d["verdict"], d["rule"]), ("pink", "D1"))
# ...but a genuine waiting-on-Steve ask in the same message still wins (attention colour)
d = sv.verdict_from_stdin({"last_assistant_message": "Done 14:40. Tell me when you've run the paste.",
"background_tasks": []}, base="green", base_variant="", idle_pink=True)
self.assertEqual((d["verdict"], d["rule"]), ("lightblue", "W1"))
def test_no_message_returns_none(self):
self.assertIsNone(sv.verdict_from_stdin({"transcript_path": "/x"}, base="green"))
self.assertIsNone(sv.verdict_from_stdin({"last_assistant_message": " "}, base="green"))
if __name__ == "__main__":
unittest.main()