← back to Terminal Status
test_departed_asks.py
253 lines
"""TK-12326: a departed session's outstanding asks survive tty reuse.
Exercises the durable register written by Store.settle_owner_change, its three-way
resolution against pending-approval memos + ticket status, ack, and the seed sweep --
all against a tempdir Store with a fake painter. Never touches a real tty, the shared
~/.local/state store, the real pending-approval queue or the real ticket log.
"""
import argparse
import io
import json
import tempfile
import unittest
from contextlib import redirect_stdout, redirect_stderr
from pathlib import Path
import departed_asks as da
import terminal_status as ts
class DepartedAsksTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.old_process = ts.Process(111, 1, "ttys050",
"Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
self.new_process = ts.Process(222, 1, "ttys050",
"Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
self.newer_process = ts.Process(333, 1, "ttys050",
"Fri Sep 11 09:00:00 2026", "/usr/bin/claude")
self.old_owner = self.old_process.owner()
self.new_owner = self.new_process.owner()
self.rows = {222: self.new_process}
self.store = ts.Store(self.root, lambda **_: self.rows, lambda o, r: None)
self.store.pending_dir.mkdir(parents=True)
(self.store.pending_dir / "_done").mkdir()
self.store.tickets_dir.mkdir(parents=True)
# ---- fixtures -------------------------------------------------------
def make_old_record(self, color, label="reason"):
old_rows = {111: self.old_process}
old_store = ts.Store(self.root, lambda **_: old_rows, lambda o, r: None)
return old_store.set(self.old_owner, color, label, rows=old_rows)
def write_legacy(self, title):
path = self.store.legacy["claude"] / (self.old_owner.tty + ".dot")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(title)
def memo(self, name, body="", filed=False):
d = self.store.pending_dir / ("_done" if filed else "")
(d / name).write_text(body)
def ticket_events(self, *events, archive=()):
(self.store.tickets_dir / "events.jsonl").write_text(
"".join(json.dumps(e) + "\n" for e in events))
if archive:
(self.store.tickets_dir / "events-archive-done-2026.jsonl").write_text(
"".join(json.dumps(e) + "\n" for e in archive))
def resolved(self):
return da.resolve(self.store.departed_path, self.store.pending_dir,
self.store.tickets_dir)
def run_cli(self, **kw):
args = argparse.Namespace(json=True, all=False, ack="", reason="", seed=False)
for k, v in kw.items():
setattr(args, k, v)
out, err = io.StringIO(), io.StringIO()
with redirect_stdout(out), redirect_stderr(err):
code = ts.cmd_departed_asks(self.store, args, live_pids=set(self.rows))
return code, out.getvalue()
# ---- register written on carry --------------------------------------
def test_settle_with_attention_registers_departed_ask(self):
self.make_old_record("purple", "TK-11449 · 6 MC appeals")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
[entry] = self.resolved()
self.assertEqual(entry["tty"], "ttys050")
self.assertEqual(entry["prior_pid"], 111)
self.assertEqual(entry["new_pid"], 222)
self.assertEqual(entry["color"], "purple")
self.assertEqual(entry["tickets"], ["TK-11449"])
def test_register_survives_new_owner_explicit_set(self):
# The whole point: TK-11317's carry is erased by the new owner's first explicit
# set; the register must not be.
self.make_old_record("orange", "TK-11450 · GMC sync click")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.store.set(self.new_owner, "green", "working", rows=self.rows)
self.assertEqual(self.store.load(self.new_owner)[0]["state"], "green")
self.assertEqual([e["tickets"] for e in self.resolved()], [["TK-11450"]])
def test_settle_without_attention_registers_nothing(self):
self.make_old_record("green", "working")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(self.resolved(), [])
def test_legacy_only_carry_still_captures_ticket_mention(self):
# prior_attention never BINDS a ticket from a legacy .dot, but the register
# records it as a mention so the ask can be traced to its memo.
self.write_legacy("\U0001F7E3 TK-11638 · legacy gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
[entry] = self.resolved()
self.assertEqual(entry["tickets"], ["TK-11638"])
self.assertIsNone(entry["prior_pid"])
def test_register_is_idempotent_per_departed_owner(self):
self.make_old_record("purple", "TK-11383 · gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(len(self.resolved()), 1)
def test_register_failure_never_blocks_the_carry(self):
self.make_old_record("purple", "gated")
self.store.departed_path = self.root / "no-such-dir" / "\0bad"
record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(record["state"], "purple")
# ---- three-way resolution -------------------------------------------
def test_open_memo_makes_ask_outstanding(self):
self.make_old_record("purple", "TK-11449 · appeals")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.memo("2026-09-13-TK-11449-appeals.md")
self.memo("unrelated.md", "mentions TK-114490 and TK-1144 but not the id")
[entry] = self.resolved()
self.assertEqual(entry["state"], "OUTSTANDING")
self.assertEqual(entry["evidence"]["TK-11449"]["open_memos"],
["2026-09-13-TK-11449-appeals.md"])
def test_memo_body_mention_counts(self):
self.make_old_record("purple", "TK-11635 · gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.memo("gmc-titlefix.md", "Ticket: TK-11635\n")
self.assertEqual(self.resolved()[0]["state"], "OUTSTANDING")
def test_filed_memo_is_not_open(self):
self.make_old_record("purple", "TK-11635 · gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.memo("TK-11635-x.md", filed=True)
self.ticket_events({"type": "create", "id": "TK-11635-x"},
{"type": "status", "id": "TK-11635-x", "status": "done"})
self.assertEqual(self.resolved()[0]["state"], "RESOLVED")
def test_done_ticket_without_memo_is_resolved(self):
self.make_old_record("orange", "TK-11450 · paste")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.ticket_events(archive=[{"type": "create", "id": "TK-11450-gmc"}])
self.assertEqual(self.resolved()[0]["state"], "RESOLVED")
def test_open_ticket_without_memo_is_unverified_not_resolved(self):
# A paste/question leaves no artifact: absence of a memo is NOT evidence the
# ask was answered (TK-11431 amendment 1).
self.make_old_record("yellow", "TK-11489 · which fork?")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.ticket_events({"type": "create", "id": "TK-11489-q"},
{"type": "status", "id": "TK-11489-q", "status": "blocked"})
entry = self.resolved()[0]
self.assertEqual(entry["state"], "UNVERIFIED")
self.assertEqual(entry["evidence"]["TK-11489"]["status"], "blocked")
def test_unknown_ticket_and_no_ticket_are_unverified(self):
self.make_old_record("lightblue", "90210 iOS resubmit")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.assertEqual(self.resolved()[0]["state"], "UNVERIFIED")
self.assertEqual(self.resolved()[0]["tickets"], [])
def test_deep_body_mention_is_weak_evidence(self):
# A memo that merely cross-references the ticket below its header is not the
# ask itself: not OUTSTANDING, but it still blocks a RESOLVED verdict.
self.make_old_record("purple", "TK-11317 · gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.ticket_events(archive=[{"type": "create", "id": "TK-11317-a"}])
self.memo("TK-12277-other.md", "\n" * 20 + "see also TK-11317\n")
entry = self.resolved()[0]
self.assertEqual(entry["state"], "UNVERIFIED")
self.assertEqual(entry["evidence"]["TK-11317"]["open_memos"], [])
self.assertEqual(entry["evidence"]["TK-11317"]["mentioned_in"], ["TK-12277-other.md"])
def test_open_memo_beats_done_ticket(self):
self.make_old_record("purple", "TK-11383 · gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.ticket_events(archive=[{"type": "create", "id": "TK-11383-a"}])
self.memo("TK-11383-still-open.md")
self.assertEqual(self.resolved()[0]["state"], "OUTSTANDING")
# ---- ack + CLI -------------------------------------------------------
def test_ack_resolves_and_unknown_ack_is_refused(self):
self.make_old_record("yellow", "which fork?")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
eid = self.resolved()[0]["id"]
with self.assertRaises(KeyError):
da.ack(self.store.departed_path, "ttys999:1:x")
da.ack(self.store.departed_path, eid, "answered in chat")
self.assertEqual(self.resolved()[0]["state"], "RESOLVED")
def test_cli_exit_code_tracks_unresolved(self):
code, out = self.run_cli()
self.assertEqual(code, 0)
self.assertEqual(json.loads(out)["counts"]["OUTSTANDING"], 0)
self.make_old_record("purple", "TK-11449 · appeals")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.memo("TK-11449.md")
code, out = self.run_cli()
self.assertEqual(code, 1)
payload = json.loads(out)
self.assertEqual(payload["counts"]["OUTSTANDING"], 1)
self.assertEqual(len(payload["asks"]), 1)
def test_cli_hides_resolved_unless_all(self):
self.make_old_record("orange", "TK-11450 · paste")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.ticket_events(archive=[{"type": "create", "id": "TK-11450-gmc"}])
code, out = self.run_cli()
self.assertEqual((code, json.loads(out)["asks"]), (0, []))
_, out = self.run_cli(all=True)
self.assertEqual(len(json.loads(out)["asks"]), 1)
# ---- seed sweep ------------------------------------------------------
def test_seed_registers_dead_owner_attention_records_only(self):
self.make_old_record("purple", "TK-11449 · appeals") # owner 111: dead
live = ts.Process(444, 1, "ttys051", "Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
self.rows[444] = live
ts.Store(self.root, lambda **_: self.rows, lambda o, r: None).set(
live.owner(), "orange", "live paste", rows=self.rows) # owner 444: live
code, out = self.run_cli(seed=True)
asks = json.loads(out)["asks"]
self.assertEqual([a["tty"] for a in asks], ["ttys050"])
self.assertEqual(asks[0]["tickets"], ["TK-11449"])
self.run_cli(seed=True) # idempotent
self.assertEqual(len(self.resolved()), 1)
def test_seed_registers_carried_record_under_original_owner(self):
self.make_old_record("purple", "TK-11383 · gated")
self.store.settle_owner_change(self.new_owner, rows=self.rows)
self.store.departed_path.unlink() # simulate a carry made before the register
self.run_cli(seed=True)
[entry] = self.resolved()
self.assertEqual((entry["prior_pid"], entry["new_pid"]), (111, 222))
def test_mentions_word_boundaries(self):
self.assertEqual(da.mentions("TK-1 tk-22, xTK-3 TK-44a TK-5-slug"),
["TK-1", "TK-22", "TK-5"])
if __name__ == "__main__":
unittest.main()