← back to Terminal Status
TK-12326: durable departed-asks register + resolver so a reused tty's prior asks outlive the carry
468627f893e4aff1ed409d36b9edf2e83ee9faf8 · 2026-09-26 09:03:11 -0700 · Steve Abrams
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdkK32KZQYqwzJKEUv17jR
Files touched
A departed_asks.pyM terminal_status.pyA test_departed_asks.py
Diff
commit 468627f893e4aff1ed409d36b9edf2e83ee9faf8
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Sep 26 09:03:11 2026 -0700
TK-12326: durable departed-asks register + resolver so a reused tty's prior asks outlive the carry
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VdkK32KZQYqwzJKEUv17jR
---
departed_asks.py | 226 ++++++++++++++++++++++++++++++++++++++++++++
terminal_status.py | 118 +++++++++++++++++++++++
test_departed_asks.py | 252 ++++++++++++++++++++++++++++++++++++++++++++++++++
3 files changed, 596 insertions(+)
diff --git a/departed_asks.py b/departed_asks.py
new file mode 100644
index 0000000..8131ecd
--- /dev/null
+++ b/departed_asks.py
@@ -0,0 +1,226 @@
+"""TK-12326: recover a DEPARTED session's outstanding asks when its tty is reused.
+
+TK-11317 carries a prior owner's attention COLOUR + label onto the new owner's record
+so a reused tty never silently blanks a needs-Steve dot. That carry is one-shot and
+shallow: the moment the new owner sets its own colour the signal is gone, and it
+never says WHICH asks were outstanding. This module is the durable other half:
+
+ * record() -- called by Store.settle_owner_change when it carries attention. Appends
+ one line per departure to <state>/departed-asks.jsonl (tty, prior
+ owner, colour, label, every TK id mentioned). Append-only, so a later
+ repaint by the new owner cannot erase it.
+ * resolve() -- joins each entry to the evidence of whether the ask is still open:
+ open memos in ~/.claude/yolo-queue/pending-approval/ (top level only;
+ _done/ etc. are filed) that name the ticket, and the ticket's status.
+ * ack() -- explicit human/agent dismissal of an entry.
+
+Three states per entry, never two (CLAUDE.md TK-11431 amendment 1):
+ OUTSTANDING -- an open pending-approval memo NAMES the ticket (filename or header).
+ RESOLVED -- every ticket is done/archived and no open memo names or mentions it,
+ or the entry was acked.
+ UNVERIFIED -- nothing proves it either way (no ticket id, unknown ticket, or a ticket
+ still open with no memo -- a paste/question leaves no artifact). This is
+ NOT-MEASURED and is listed, never silently dropped.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import fcntl
+import json
+import re
+from pathlib import Path
+
+TK_RE = re.compile(r"(?<![0-9A-Za-z])TK-(\d+)(?![0-9A-Za-z])", re.I)
+DONE_STATES = frozenset(("done", "closed", "cancelled", "canceled", "wontfix"))
+
+
+def mentions(*texts):
+ """Every distinct short ticket id (TK-<digits>) mentioned in the given texts, in
+ first-seen order."""
+ seen = []
+ for text in texts:
+ for m in TK_RE.finditer(text or ""):
+ tk = "TK-" + m[1]
+ if tk not in seen:
+ seen.append(tk)
+ return seen
+
+
+def _now():
+ return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
+
+
+def _read_lines(path):
+ try:
+ text = path.read_text()
+ except OSError:
+ return []
+ out = []
+ for line in text.splitlines():
+ try:
+ out.append(json.loads(line))
+ except ValueError:
+ continue
+ return out
+
+
+def _append(path, entry):
+ path.parent.mkdir(parents=True, exist_ok=True)
+ with open(path, "a") as fh:
+ fcntl.flock(fh, fcntl.LOCK_EX)
+ try:
+ fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
+ finally:
+ fcntl.flock(fh, fcntl.LOCK_UN)
+
+
+def entry_id(tty, prior_pid, prior_started):
+ return f"{tty}:{prior_pid}:{prior_started}"
+
+
+def record(path, *, tty, runtime, prior, prior_info, new_pid, raw_texts=()):
+ """Append one departure. Idempotent on (tty, prior owner): a second settle for the
+ same departed owner is a no-op. Returns the entry written, or None if skipped."""
+ info = prior_info or {}
+ eid = entry_id(tty, info.get("pid"), info.get("started"))
+ if info.get("pid") is None:
+ # legacy-.dot-only carry has no stored owner; key on the colour+label instead so
+ # the same stale mirror is not re-recorded on every new owner of this tty.
+ eid = f"{tty}:legacy:{prior['color']}:{prior.get('label', '')}"
+ for e in _read_lines(path):
+ if e.get("kind", "departed") == "departed" and e.get("id") == eid:
+ return None
+ tickets = mentions(prior.get("ticket", ""), prior.get("label", ""), *raw_texts)
+ entry = {"kind": "departed", "id": eid, "ts": _now(), "tty": tty,
+ "runtime": runtime, "prior_pid": info.get("pid"),
+ "prior_started": info.get("started"), "new_pid": new_pid,
+ "color": prior["color"], "label": prior.get("label", ""),
+ "tickets": tickets}
+ _append(path, entry)
+ return entry
+
+
+def ack(path, eid, reason="", agent=""):
+ known = {e.get("id") for e in _read_lines(path) if e.get("kind", "departed") == "departed"}
+ if eid not in known:
+ raise KeyError(eid)
+ entry = {"kind": "ack", "id": eid, "ts": _now(), "reason": reason, "agent": agent}
+ _append(path, entry)
+ return entry
+
+
+def open_memos(pending_dir):
+ """{memo filename: text} for OPEN memos only -- top-level *.md. Sub-directories
+ (_done, _approved, _archive ...) are filed decisions, not open asks."""
+ out = {}
+ try:
+ paths = sorted(Path(pending_dir).glob("*.md"))
+ except OSError:
+ return out
+ for p in paths:
+ try:
+ out[p.name] = p.read_text(errors="replace")
+ except OSError:
+ out[p.name] = ""
+ return out
+
+
+HEADER_LINES = 15
+
+
+def memos_naming(ticket, memos):
+ """(named, mentioned): memos whose FILENAME or header (first HEADER_LINES lines --
+ title / Ticket: line / frontmatter) names the ticket -- the memo IS that ask -- vs
+ memos that only mention it deeper in the body (a cross-reference). Only `named`
+ proves OUTSTANDING; `mentioned` still blocks RESOLVED (it is unexamined evidence)."""
+ pat = re.compile(r"(?<![0-9A-Za-z])" + re.escape(ticket) + r"(?![0-9])", re.I)
+ named, mentioned = [], []
+ for name, body in sorted(memos.items()):
+ head = "\n".join(body.splitlines()[:HEADER_LINES])
+ if pat.search(name) or pat.search(head):
+ named.append(name)
+ elif pat.search(body):
+ mentioned.append(name)
+ return named, mentioned
+
+
+def ticket_statuses(tickets_dir, wanted):
+ """Latest status per SHORT id (TK-<digits>) for the wanted set, from the canonical
+ event log. A ticket present only in an events-archive-done-* file is 'done'. A
+ ticket found nowhere is absent from the result (UNVERIFIED, never assumed done)."""
+ wanted = set(wanted)
+ if not wanted:
+ return {}
+ root = Path(tickets_dir)
+ short = re.compile(r"^(TK-\d+)(?:-|$)")
+ status = {}
+
+ def scan(path, archive):
+ try:
+ fh = open(path, errors="replace")
+ except OSError:
+ return
+ with fh:
+ for line in fh:
+ # cheap pre-filter before json.loads on a ~65MB log; spacing-agnostic
+ if "TK-" not in line or ('"create"' not in line and '"status"' not in line):
+ continue
+ try:
+ ev = json.loads(line)
+ except ValueError:
+ continue
+ m = short.match(ev.get("id", ""))
+ if not m or m[1] not in wanted:
+ continue
+ if ev.get("type") not in ("create", "status"):
+ continue
+ if archive:
+ status[m[1]] = "done"
+ elif ev.get("type") == "status" and ev.get("status"):
+ status[m[1]] = ev["status"]
+ else:
+ status.setdefault(m[1], "open")
+
+ for archive in sorted(root.glob("events-archive-done-*.jsonl")):
+ scan(archive, True)
+ scan(root / "events.jsonl", False)
+ return status
+
+
+def resolve(path, pending_dir, tickets_dir, *, statuses=None, memos=None):
+ """Every departed entry with its evidence and a three-way state. `statuses` /
+ `memos` are injection seams for tests (never passed by the CLI)."""
+ lines = _read_lines(path)
+ acks = {e["id"]: e for e in lines if e.get("kind") == "ack" and e.get("id")}
+ entries = [e for e in lines if e.get("kind", "departed") == "departed"]
+ wanted = {tk for e in entries for tk in e.get("tickets", [])}
+ if memos is None:
+ memos = open_memos(pending_dir)
+ if statuses is None:
+ statuses = ticket_statuses(tickets_dir, wanted)
+ out = []
+ for e in entries:
+ tks = e.get("tickets", [])
+ evidence = {}
+ for tk in tks:
+ named, mentioned = memos_naming(tk, memos)
+ evidence[tk] = {"status": statuses.get(tk, "unknown"),
+ "open_memos": named, "mentioned_in": mentioned}
+ if e["id"] in acks:
+ state = "RESOLVED"
+ elif any(v["open_memos"] for v in evidence.values()):
+ state = "OUTSTANDING"
+ elif tks and all(v["status"] in DONE_STATES and not v["mentioned_in"]
+ for v in evidence.values()):
+ state = "RESOLVED"
+ else:
+ state = "UNVERIFIED"
+ out.append(dict(e, state=state, evidence=evidence, acked=acks.get(e["id"])))
+ return out
+
+
+def summary(resolved):
+ counts = {"OUTSTANDING": 0, "UNVERIFIED": 0, "RESOLVED": 0}
+ for r in resolved:
+ counts[r["state"]] += 1
+ return counts
diff --git a/terminal_status.py b/terminal_status.py
index 84ee386..5d1001f 100644
--- a/terminal_status.py
+++ b/terminal_status.py
@@ -19,6 +19,7 @@ import tempfile
import time
import uuid
import ticket_binding as tickets
+import departed_asks
# TK-11505: the ps scan ceiling was hardcoded at 60s. At ~2200 procs under
# concurrent load (N sessions each scanning all N sessions' processes) this is
@@ -825,6 +826,11 @@ class Store:
self.painter = painter
self.ticket_evidence = {}
self.known_tickets = None
+ # TK-12326: durable register of asks left behind by departed owners, and the
+ # evidence sources the resolver joins it against.
+ self.departed_path = self.root / "departed-asks.jsonl"
+ self.pending_dir = self.user_root / ".claude/yolo-queue/pending-approval"
+ self.tickets_dir = self.user_root / ".claude/tickets"
def ticket_info(self, owner, record=None):
record = record or {}
@@ -998,6 +1004,35 @@ class Store:
return {"pid": r["owner"].get("pid"), "started": r["owner"].get("started")}
return None
+ def _prior_raw_texts(self, owner):
+ """Raw label/title text the prior owner left in BOTH sources, so ticket ids a
+ legacy-.dot-only carry deliberately does not bind (prior_attention) are still
+ captured as MENTIONS for the departed-asks register (TK-12326)."""
+ texts = []
+ r = self._read_raw_for_prior(owner)
+ if r is not None:
+ texts += [r.get("title", ""), r.get("ticket", "")]
+ try:
+ texts.append((self.legacy[owner.runtime] / (owner.tty + ".dot")).read_text())
+ except OSError:
+ pass
+ return texts
+
+ def record_departed(self, owner, prior, info):
+ """TK-12326: append the departing owner's ask to the durable register before
+ the carry is written, so it outlives the new owner's first explicit set.
+ Best-effort by design -- a register failure must never block restoring the
+ carried colour itself."""
+ if info and info.get("pid") == owner.pid and info.get("started") == owner.started:
+ return None # the "prior" record is our own -- nothing departed
+ try:
+ return departed_asks.record(
+ self.departed_path, tty=owner.tty, runtime=owner.runtime,
+ prior=prior, prior_info=info, new_pid=owner.pid,
+ raw_texts=self._prior_raw_texts(owner))
+ except (OSError, ValueError, KeyError, TypeError):
+ return None
+
def settle_owner_change(self, owner, *, rows=None, lock_wait=None):
"""TK-11317 (Steve's TK-11620 guard ruling, 2026-09-26): a tty-reuse
owner_changed record must NEVER silently blank or green-floor a prior
@@ -1011,6 +1046,7 @@ class Store:
prior = self.prior_attention(owner)
info = self.prior_owner_info(owner)
if prior is not None:
+ self.record_departed(owner, prior, info)
label = prior["label"] or COLORS[prior["color"]][2]
try:
return self.set(owner, prior["color"], label,
@@ -1771,6 +1807,77 @@ def cmd_auto_bind(store, owner, prompt, project, *, now=None, tk_new=None, ticke
return {"action": "created", "ticket": new_id}
+def seed_departed(store, live_pids):
+ """TK-12326 backfill: every stored ATTENTION record whose owner pid is no longer
+ live is an ask its session left behind -- whether or not a new owner has reused
+ the tty yet. Also registers records that already CARRY a prior ask (TK-11317
+ carried_from) under the original departed owner. Returns entries written."""
+ written = []
+ for path in sorted(store.root.glob("ttys*.json")):
+ try:
+ r = json.loads(path.read_text())
+ except (OSError, ValueError):
+ continue
+ if not isinstance(r, dict) or r.get("state") not in ATTENTION:
+ continue
+ owner = r.get("owner") if isinstance(r.get("owner"), dict) else {}
+ tty = path.stem
+ label, _ = label_ticket(r.get("label", ""))
+ prior = {"color": r["state"], "label": label, "ticket": r.get("ticket", "")}
+ carried = r.get("carried_from") if isinstance(r.get("carried_from"), dict) else None
+ if carried and carried.get("pid") not in live_pids:
+ info, new_pid = carried, owner.get("pid")
+ elif owner.get("pid") is not None and owner.get("pid") not in live_pids:
+ info, new_pid = {"pid": owner.get("pid"), "started": owner.get("started")}, None
+ else:
+ continue
+ e = departed_asks.record(store.departed_path, tty=tty,
+ runtime=owner.get("runtime", ""), prior=prior,
+ prior_info=info, new_pid=new_pid,
+ raw_texts=(r.get("title", ""),))
+ if e:
+ written.append(e)
+ return written
+
+
+def cmd_departed_asks(store, args, live_pids=None):
+ if args.ack:
+ try:
+ departed_asks.ack(store.departed_path, args.ack, args.reason,
+ os.environ.get("TK_AGENT", ""))
+ except KeyError:
+ print("no such departed ask: " + args.ack, file=sys.stderr)
+ return 2
+ print("acked " + args.ack)
+ return 0
+ if args.seed:
+ if live_pids is None:
+ live_pids = set(processes(fresh=True))
+ written = seed_departed(store, live_pids)
+ print(f"seeded {len(written)} departed ask(s)", file=sys.stderr)
+ resolved = departed_asks.resolve(store.departed_path, store.pending_dir, store.tickets_dir)
+ shown = [r for r in resolved if args.all or r["state"] != "RESOLVED"]
+ counts = departed_asks.summary(resolved)
+ if args.json:
+ print(json.dumps({"counts": counts, "asks": shown}, ensure_ascii=False))
+ else:
+ for r in shown:
+ icon = COLORS.get(r["color"], ("",))[0]
+ tks = ", ".join(r["tickets"]) or "(no ticket)"
+ print(f'{r["state"]:<11} {r["tty"]:<8} {icon} {tks} · {r["label"]} [id {r["id"]}]')
+ for tk, ev in r["evidence"].items():
+ for memo in ev["open_memos"]:
+ print(f" ↳ {tk} [{ev['status']}] memo: {memo}")
+ for memo in ev.get("mentioned_in", []):
+ print(f" ↳ {tk} [{ev['status']}] mentioned in: {memo}")
+ if not ev["open_memos"] and not ev.get("mentioned_in"):
+ print(f" ↳ {tk} [{ev['status']}] no open memo")
+ print(f'OUTSTANDING {counts["OUTSTANDING"]} · UNVERIFIED {counts["UNVERIFIED"]} · '
+ f'RESOLVED {counts["RESOLVED"]}')
+ # Non-zero while anything is unresolved, so a canary/cron can key off the exit code.
+ return 1 if counts["OUTSTANDING"] or counts["UNVERIFIED"] else 0
+
+
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
sub = parser.add_subparsers(dest="command", required=True)
@@ -1839,9 +1946,20 @@ def main(argv=None):
bf = sub.add_parser("backfill") # TK-11870 scheduled always-on self-heal
bf.add_argument("--quiet", action="store_true")
sub.add_parser("selftest") # TK-11791/TK-11870 negative tests (never run by a plist)
+ da = sub.add_parser("departed-asks",
+ help="TK-12326: asks left behind by sessions whose tty was reused")
+ da.add_argument("--json", action="store_true")
+ da.add_argument("--all", action="store_true", help="include RESOLVED entries")
+ da.add_argument("--ack", default="", metavar="ID", help="dismiss one entry by id")
+ da.add_argument("--reason", default="")
+ da.add_argument("--seed", action="store_true",
+ help="register attention records whose owner process is dead "
+ "(asks orphaned before this register existed)")
args = parser.parse_args(argv)
if args.command == "selftest":
return selftest()
+ if args.command == "departed-asks":
+ return cmd_departed_asks(Store(), args)
store = Store()
if args.command == "backfill":
# TK-11870: the SOLE scheduled process-table PRODUCER + the always-on
diff --git a/test_departed_asks.py b/test_departed_asks.py
new file mode 100644
index 0000000..d96056d
--- /dev/null
+++ b/test_departed_asks.py
@@ -0,0 +1,252 @@
+"""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()
← 76859a5 TK-11317: live acceptance evidence (Claude+Codex readback, s
·
back to Terminal Status
·
TK-11317: fix deferential-set bypass + ticket-rebind hole; a f171fc9 →