← back to Terminal Status
test_terminal_status.py
1220 lines
import dataclasses
import datetime as dt
import json
import os
from pathlib import Path
import re
import tempfile
import subprocess
import sys
import unittest
from unittest.mock import patch
import terminal_status as ts
import ticket_binding as tb
class StatusTests(unittest.TestCase):
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.process = ts.Process(123, 1, "ttys010",
"Wed Sep 9 08:35:08 2026", "/bin/codex")
self.rows = {123: self.process}
self.owner = self.process.owner()
self.paints = []
self.store = ts.Store(self.root, lambda: self.rows,
lambda owner, record: self.paints.append(record.copy()))
def legacy(self, runtime, title, stamp=None):
path = self.store.legacy[runtime] / (self.owner.tty + ".dot")
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(title)
when = stamp if stamp is not None else self.owner.epoch + 10
os.utime(path, (when, when))
return path
def test_ticket_survives_every_generic_color_and_clear(self):
self.store.set(self.owner, "green", "TK-11317 · initial")
for color in ts.COLORS:
record = self.store.set(self.owner, color)
self.assertEqual(record["ticket"], "TK-11317")
self.assertEqual(record["title"].count("TK-11317"), 0 if color == "none" else 1)
def test_event_clock_is_comparable_between_python_processes(self):
before = ts.issued_clock()
child = int(subprocess.check_output([sys.executable, '-c',
'import terminal_status; print(terminal_status.issued_clock())'], text=True))
after = ts.issued_clock()
self.assertLessEqual(before, child)
self.assertLessEqual(child, after)
def test_ticket_header_is_read_only_and_does_not_resurrect_clear(self):
self.store.set(self.owner, "none", ticket_update="TK-11317")
before = self.store.path(self.owner).read_bytes()
outputs = []
title = self.store.header(self.owner, lambda owner, data: outputs.append(data))
self.assertIn("⚪ TK-11317 · Status cleared", title)
self.assertEqual(before, self.store.path(self.owner).read_bytes())
self.assertNotIn(b']6;', outputs[0])
def test_new_assignment_wins_and_generic_hook_preserves_binding(self):
self.store.ticket_evidence = {self.owner.tty: {"id": "TK-1", "at": 10, "source": "ticket_ledger"}}
self.store.set(self.owner, "green")
self.store.ticket_evidence[self.owner.tty].update(id="TK-2", at=20)
self.assertIn("TK-2", self.store.row(self.owner)["label"])
r = self.store.set(self.owner, "purple")
self.assertEqual(r["ticket"], "TK-2")
self.store.ticket_evidence[self.owner.tty].update(id="TK-1", at=10)
self.assertEqual(self.store.set(self.owner, "green")["ticket"], "TK-2")
def test_ticket_validation_and_tty_reuse(self):
self.store.known_tickets = {"TK-11317"}
with self.assertRaises(ts.StatusError):
self.store.set(self.owner, "green", ticket_update="TK-99999999")
self.store.set(self.owner, "green", ticket_update="TK-11317")
self.process.started = "Wed Sep 9 09:35:08 2026"
self.assertEqual(self.store.row(self.process.owner())["ticket"], "")
def test_monitoring_shade_remains_green_and_survives_repaint(self):
record = self.store.set(self.owner, "green", "Next 10:30", variant="monitoring")
self.assertIn(b"green;brightness;150", ts.osc_payload(record))
self.store.repaint(self.owner)
self.assertEqual(self.paints[-1]["variant"], "monitoring")
self.assertEqual(self.store.row(self.owner)["color"], "green")
def test_waiting_shade_is_brighter_green_and_survives_repaint(self):
# Parallel to the monitoring variant: waiting paints a distinct brighter STATIC
# green rgb(0,255,120), stays state=="green" (a green variant, not a new colour),
# and survives a repaint.
record = self.store.set(self.owner, "green", "Idle", variant="waiting")
packet = ts.osc_payload(record)
self.assertIn(b"red;brightness;0", packet)
self.assertIn(b"green;brightness;255", packet)
self.assertIn(b"blue;brightness;120", packet)
self.store.repaint(self.owner)
self.assertEqual(self.paints[-1]["variant"], "waiting")
self.assertEqual(self.store.row(self.owner)["color"], "green")
def test_lightblue_preserves_underlying_needs_steve_reason(self):
# T1-c (TK-11779): an explicit lightblue on a purple/orange/yellow base must NOT
# erase the specific reason — it rides as the additive 🔵 stopped marker on the
# preserved base colour, so a later resume reveals the still-pending reason.
for base in ("purple", "orange", "yellow"):
with self.subTest(base=base):
self.store.set(self.owner, base, "TK-11779 · the real reason")
r = self.store.set(self.owner, "lightblue")
self.assertEqual(r["state"], base, "base colour must be preserved")
self.assertEqual(r["variant"], "stopped", "needs-Steve rides as 🔵 marker")
self.assertEqual(r["label"], "the real reason", "reason label preserved")
self.assertEqual(r["ticket"], "TK-11779", "ticket preserved")
self.assertIn(ts.STOPPED_MARK, r["title"], "🔵 stopped marker shown")
# resuming (clear the stopped variant) reveals the still-pending reason
back = self.store.set_variant(self.owner, "")
self.assertEqual(back["state"], base)
self.assertEqual(back["variant"], "")
self.store.set(self.owner, "none") # reset for next subTest
def test_lightblue_is_solid_when_no_reason_to_preserve(self):
# A green/pink/none base carries no needs-Steve reason, so lightblue paints solid.
for base, seed in (("green", lambda: self.store.set(self.owner, "green", "Working")),
("pink", lambda: self.store.set(self.owner, "pink", "parked")),
("none", lambda: None)):
with self.subTest(base=base):
seed()
r = self.store.set(self.owner, "lightblue")
self.assertEqual(r["state"], "lightblue")
self.assertEqual(r["variant"], "")
self.store.set(self.owner, "none")
def test_explicit_lightblue_reason_wins_over_preserved_base(self):
# An explicit `set(lightblue, "TK-new · reason")` on a purple/orange/yellow base
# must NOT staple the NEW ticket onto the OLD label. The caller's reason wins:
# paints a solid lightblue carrying the new label AND the new ticket, consistently.
for base in ("purple", "orange", "yellow"):
with self.subTest(base=base):
self.store.set(self.owner, base, "TK-11779 · the OLD reason")
r = self.store.set(self.owner, "lightblue", "TK-11780 · the NEW reason")
self.assertEqual(r["state"], "lightblue", "explicit reason paints solid lightblue")
self.assertEqual(r["variant"], "")
self.assertEqual(r["label"], "the NEW reason", "new label must not be dropped")
self.assertEqual(r["ticket"], "TK-11780", "new ticket must not pair with old label")
self.assertNotIn("OLD", r["title"], "no leak of the preserved base's reason")
self.store.set(self.owner, "none")
def test_lightblue_never_mismatches_old_label_with_new_ticket(self):
# TK-11826 / TK-11779 Finding 2: guard the coherence invariant permanently — a
# lightblue paint on a purple/orange/yellow base must never produce the mismatch of
# an OLD preserved label carrying a NEW caller ticket. Every lightblue result must
# take its label and ticket from the SAME source: both preserved (bare set) or both
# the caller's explicit reason. The T1-c block pins ticket_update=None defensively.
for base in ("purple", "orange", "yellow"):
with self.subTest(base=base, path="bare-preserve"):
self.store.set(self.owner, base, "TK-11779 · reason A")
r = self.store.set(self.owner, "lightblue") # no reason -> preserve whole
self.assertEqual(r["state"], base, "base colour preserved")
self.assertEqual(r["variant"], "stopped")
self.assertEqual(r["label"], "reason A", "preserved label")
self.assertEqual(r["ticket"], "TK-11779",
"preserved ticket must stay with the preserved label")
self.store.set(self.owner, "none")
with self.subTest(base=base, path="explicit-ticket-only"):
# An explicit ticket_update with NO label must not staple the new ticket onto
# the preserved base's old label — it paints a coherent solid lightblue.
self.store.set(self.owner, base, "TK-11779 · reason A")
r = self.store.set(self.owner, "lightblue", ticket_update="TK-11780")
self.assertEqual(r["state"], "lightblue", "explicit ticket paints solid lightblue")
self.assertEqual(r["ticket"], "TK-11780", "new ticket taken")
self.assertNotIn("reason A", r["title"], "no old label paired with the new ticket")
self.store.set(self.owner, "none")
def test_ledger_discovery_rejects_reused_pid_and_nested_agent(self):
self.rows[124] = ts.Process(124, 123, "??", self.process.started, "/bin/node")
self.rows[125] = ts.Process(125, 123, "??", self.process.started, "/bin/codex")
self.rows[126] = ts.Process(126, 125, "??", self.process.started, "/bin/node")
path = self.root / ".claude/tickets/events.jsonl"
path.parent.mkdir(parents=True)
def event(pid, seconds, ticket):
return {"id": ticket, "type": "create", "correlation_id": f"create-abc-{pid}-xyz",
"ts": dt.datetime.fromtimestamp(self.owner.epoch + seconds, dt.timezone.utc).isoformat()}
events = [event(124, -1, "TK-1-old"), event(124, 10, "TK-2-current"),
event(126, 20, "TK-3-child"), event(999, 30, "TK-4-unrelated")]
path.write_text("\n".join(json.dumps(e) for e in events))
found, known = ts.tickets.discover(self.root, self.rows, ts.owners(self.rows), ts.ancestors, argv={})
self.assertEqual(found["ttys010"]["id"], "TK-2")
self.assertEqual(len(known), 4)
# TK-11631: run-ticket.sh's own `export TK_AGENT=<prefix>-<IDNUM>` carries only
# the DRIVING ticket's number, so it binds sessions the bare-id scan must abstain
# on. RUN_TICKET_ARGV is the verbatim argv of live pid 48691 (2026-09-13).
RUN_TICKET_ARGV = (
"claude --model opus export TK_AGENT=claude-run-11630. You are driving ticket "
"TK-11630-tk-11340-follow-on-2-live-gated-memos-we to completion. First run: tk "
"inbox (act on any DMs), then tk show TK-11630-tk-11340-follow-on-2-live-gated-"
"memos-we for full context. You now OWN this ticket: tk take TK-11630-tk-11340-"
"follow-on-2-live-gated-memos-we.")
def ledger(self, *tickets):
path = self.root / ".claude/tickets/events.jsonl"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text("\n".join(json.dumps(
{"id": t, "type": "create",
"ts": dt.datetime.fromtimestamp(self.owner.epoch - 60, dt.timezone.utc).isoformat()})
for t in tickets))
def session_ledger(self, tsid, *events, created=()):
"""Write a ledger of claude@<tsid> events (each (type, ticket, offset_s))
plus optional standalone create events, for the TK-11665 session route."""
path = self.root / ".claude/tickets/events.jsonl"
path.parent.mkdir(parents=True, exist_ok=True)
lines = []
for tk in created:
lines.append(json.dumps({"id": tk, "type": "create",
"ts": dt.datetime.fromtimestamp(self.owner.epoch - 3600, dt.timezone.utc).isoformat()}))
for etype, ticket, offset in events:
lines.append(json.dumps({"id": ticket, "type": etype,
"agent": "claude@" + tsid,
"ts": dt.datetime.fromtimestamp(self.owner.epoch + offset, dt.timezone.utc).isoformat()}))
path.write_text("\n".join(lines))
def test_session_ledger_binds_resumed_session_via_term_session_id(self):
# TK-11665. A bare `claude --continue` carries NO ticket in argv, so every
# argv source abstains. Its work is logged under claude@<TERM_SESSION_ID>,
# and a resume in the same pane keeps that id -- so events written BEFORE
# this process started (offset -3600, i.e. before owner.epoch) are still
# this session's own work and MUST bind. Keyed PID->TERM_SESSION_ID via env.
tsid = "w7t0p0:E2D0289F-FA6C-45F7-8D42-42A35D406E46"
self.session_ledger(tsid,
("assign", "TK-11665-fix-x", -3600),
("action", "TK-11665-fix-x", -1800),
created=("TK-11665-fix-x",))
found, _ = ts.tickets.discover(
self.root, self.rows, ts.owners(self.rows), ts.ancestors,
argv={123: "claude --continue"}, env={123: tsid})
self.assertEqual(found.get("ttys010"),
{"id": "TK-11665", "at": self.owner.epoch - 1800,
"source": "session_ledger"})
def test_session_ledger_is_non_vacuous_and_never_launders(self):
# TK-11665 NEGATIVE test (TK-11431 rule 3): the source must go quiet on
# every shape it must not bind, so a green result is evidence, not habit.
tsid = "w7t0p0:AAAAAAAA-0000-0000-0000-000000000000"
base = dict(argv={123: "claude --continue"}, env={123: tsid})
bind = lambda: ts.tickets.discover(self.root, self.rows,
ts.owners(self.rows), ts.ancestors, **base)
# (a) `read` events are board-viewing NOISE, never a binding.
self.session_ledger(tsid, ("read", "TK-11665-x", -60), created=("TK-11665-x",))
self.assertNotIn("ttys010", bind()[0])
# (b) a ticket the session touched but that was NEVER created cannot bind
# (the `in known` safety -- it can never invent a ticket).
self.session_ledger(tsid, ("action", "TK-99999999-ghost", -60))
self.assertNotIn("ttys010", bind()[0])
# (c) no TERM_SESSION_ID in env (a session under a self-chosen TK_AGENT --
# the ticket's KNOWN CEILING): stays honest, binds nothing.
self.session_ledger(tsid, ("action", "TK-11665-x", -60), created=("TK-11665-x",))
self.assertNotIn("ttys010",
ts.tickets.discover(self.root, self.rows, ts.owners(self.rows),
ts.ancestors, argv={123: "claude --continue"},
env={})[0])
# (d) an argv ticket present: the session route must NEVER override a bound
# argv source (it is the last resort, not a competitor).
self.session_ledger(tsid,
("action", "TK-11665-x", -60),
created=("TK-11665-x", "TK-11317"))
found, _ = ts.tickets.discover(
self.root, self.rows, ts.owners(self.rows), ts.ancestors,
argv={123: "claude drive TK-11317 to done"}, env={123: tsid})
self.assertEqual(found["ttys010"]["source"], "main_process_argument")
self.assertEqual(found["ttys010"]["id"], "TK-11317")
def bind(self, argv):
return ts.tickets.discover(self.root, self.rows, ts.owners(self.rows),
ts.ancestors, argv=argv)
def test_declared_agent_binds_driving_ticket_when_slug_embeds_a_second_id(self):
self.rows[124] = ts.Process(124, 1, "ttys011", self.process.started, "/bin/claude")
self.ledger("TK-11630-tk-11340-follow-on-2-live-gated-memos-we", "TK-11317")
# The pre-existing rule cannot bind this argv: the slug embeds a second id.
self.assertEqual(len(set(re.findall(r"\bTK-\d+\b", self.RUN_TICKET_ARGV, re.I))), 2)
# The launcher emits `export ` then the token then a literal period, and the
# digits are anchored between them: a greedy capture would yield
# "claude-run-11630." and bind nothing.
self.assertIn("export TK_AGENT=claude-run-11630. You are", self.RUN_TICKET_ARGV)
self.assertEqual(ts.tickets.AGENT.findall(self.RUN_TICKET_ARGV), ["11630"])
found, known = self.bind({123: self.RUN_TICKET_ARGV,
124: "claude --model opus Please drive TK-11317 to done"})
self.assertEqual(found.get("ttys010"), {"id": "TK-11630", "at": self.owner.epoch,
"source": "main_process_agent"})
# ...and the single-id rule it defers to is untouched.
self.assertEqual(found["ttys011"], {"id": "TK-11317", "at": self.owner.epoch,
"source": "main_process_argument"})
self.assertEqual(known, {"TK-11630", "TK-11317"})
def test_declared_agent_never_invents_a_ticket_nor_launders_a_bare_session(self):
self.rows[124] = ts.Process(124, 1, "ttys011", self.process.started, "/bin/claude")
self.ledger("TK-11630-tk-11340-follow-on-2-live-gated-memos-we")
unknown = self.RUN_TICKET_ARGV.replace("TK_AGENT=claude-run-11630",
"TK_AGENT=claude-run-99999999")
found, _ = self.bind({123: "claude", 124: unknown})
# A declared id that was never created binds nothing -- and does NOT fall
# back to guessing one of the ids the slug happens to mention.
self.assertNotIn("ttys011", found)
# A genuinely ticketless session stays unbound and still reads TK REQUIRED.
self.assertNotIn("ttys010", found)
self.store.ticket_evidence = found
self.store.set(self.owner, "green", "Working")
self.assertIn("TK REQUIRED", self.store.row(self.owner)["label"])
# TK-11369/TK-11505: an unavailable `ps` degrades the label, never the paint.
self.assertEqual(self.bind({})[0], {})
def test_declared_agent_needs_an_export_and_exactly_one_declaration(self):
# TK-11631 follow-up. The first cut matched a bare `TK_AGENT=` anywhere and took
# re.search's FIRST hit, so an agent whose prompt merely NAMED the ticket it was
# debugging got bound to it (memory: detector-argv-substring-self-match). Every
# argv below is a real shape; `driving` is the only session actually launched for
# a ticket. Expected value is (id, source) or None for "nothing binds at all".
self.ledger("TK-11631-chronic-x", "TK-11630-tk-11340-x", "TK-11076-tk-11073-x")
agent, argument = "main_process_agent", "main_process_argument"
cases = {
"driving": (
"claude --model opus export TK_AGENT=claude-run-11631. You are driving "
"ticket TK-11631-chronic-x to completion.", ("TK-11631", agent)),
# A prose mention with no `export ` must not reach the new source at all.
# The loose regex bound TK-11630 here off nothing but a sentence; now NOTHING
# binds, because the old rule also abstains (the slug holds two bare ids).
"prose_only": (
"claude fix the bug where TK_AGENT=claude-run-11630 fails to bind on "
"TK-11630-tk-11340-x", None),
# Same prose mention, but a single bare id the OLD rule can still bind: proves
# that rule is untouched, and that the prose TK_AGENT did not steer it.
"prose_only_old_rule_unaffected": (
"claude fix the bug where TK_AGENT=claude-run-11630 fails to bind on "
"TK-11631", ("TK-11631", argument)),
"export_plus_prose": (
"claude export TK_AGENT=claude-run-11631. Compare against "
"TK_AGENT=claude-run-99999 behaviour", ("TK-11631", agent)),
# The argv of an agent working THIS ticket: three mentions, one export.
"export_plus_two_prose": (
"claude --model opus export TK_AGENT=claude-run-11631. Root-cause why "
"TK_AGENT=claude-run-11630 and TK_AGENT=claude-run-11076 read "
"TK REQUIRED", ("TK-11631", agent)),
# Identical content, prose FIRST: proves the bind is not order-luck.
"prose_before_export": (
"claude --model opus TK_AGENT=claude-run-11630 and "
"TK_AGENT=claude-run-11076 read TK REQUIRED. Root-cause: export "
"TK_AGENT=claude-run-11631. Fix", ("TK-11631", agent)),
# Two real declarations are ambiguous: abstain, never pick one.
"two_exports": (
"claude export TK_AGENT=claude-run-11631. Then export "
"TK_AGENT=claude-run-11630. Compare", None),
# `export ` present but no trailing period: not the launcher's shape.
"export_without_period": (
"claude the launcher should export TK_AGENT=claude-run-11630 but does "
"not", None),
# Trailing period present but NO `export `: a sentence that simply ends on
# the token. Isolates the `export ` half of the anchor -- without it this
# case binds another session's ticket off prose.
"period_without_export": (
"claude why does the tab for TK_AGENT=claude-run-11630. read "
"TK REQUIRED", None),
}
for name, (command, expected) in cases.items():
with self.subTest(name):
bound = self.bind({123: command})[0].get("ttys010")
actual = (bound["id"], bound["source"]) if bound else None
self.assertEqual(actual, expected)
# The two that MUST differ from the loose regex do so for opposite reasons.
self.assertEqual(ts.tickets.AGENT.findall(cases["prose_only"][0]), [])
self.assertEqual(len(set(ts.tickets.AGENT.findall(cases["two_exports"][0]))), 2)
def test_all_colors_write_canonical_and_identical_mirrors(self):
for color in ts.COLORS:
r = self.store.set(self.owner, color, "" if color == "none" else "Task")
self.assertEqual(self.store.load(self.owner)[0], r)
for directory in self.store.legacy.values():
self.assertEqual((directory / "ttys010.dot").read_text(), r["title"])
def test_canonical_green_beats_stale_orange_title_and_cache(self):
self.store.set(self.owner, "green", "Working")
self.legacy("claude", "🟠 stale gate")
row = self.store.row(self.owner, "🟠 stale gate")
self.assertEqual(row["color"], "green")
self.assertIn("window_title_disagrees", row["warnings"])
self.assertIn("claude_mirror_disagrees", row["warnings"])
def test_repaint_never_changes_record_or_mirror_timestamps(self):
self.store.set(self.owner, "orange", "A real paste request")
paths = [self.store.path(self.owner)] + [
d / "ttys010.dot" for d in self.store.legacy.values()]
before = [(p.read_bytes(), p.stat().st_mtime_ns) for p in paths]
self.store.repaint(self.owner)
self.assertEqual(before, [(p.read_bytes(), p.stat().st_mtime_ns) for p in paths])
def test_clear_tombstone_prevents_title_cache_and_repaint_resurrection(self):
self.store.set(self.owner, "orange", "Old")
clear = self.store.set(self.owner, "none")
self.legacy("codex", "🟠 old copy returned")
self.assertEqual(self.store.row(self.owner, "🟠 old")["color"], "none")
self.store.repaint(self.owner)
self.assertEqual(self.paints[-1]["state"], "none")
self.assertEqual(self.store.load(self.owner)[0], clear)
def test_old_request_cannot_overwrite_a_newer_clear(self):
self.store.set(self.owner, "green", issued_ns=10)
clear = self.store.set(self.owner, "none", issued_ns=30)
with self.assertRaises(ts.StatusError):
self.store.set(self.owner, "orange", issued_ns=20)
self.assertEqual(self.store.load(self.owner)[0], clear)
def test_pid_reuse_and_runtime_change_fail_closed(self):
self.store.set(self.owner, "green")
# TK-11317: row() no longer renders a bare "none"/"status not set" for
# reason owner_changed -- it renders the would-be settle_owner_change()
# preview. The prior record here is GREEN (not an ATTENTION colour), so
# the preview is the yellow "New owner - status needed" floor, flagged
# owner_status_unverified -- never "none" and never a silent green carry.
for new in (dataclasses.replace(self.owner, started="Wed Sep 9 09:35:08 2026"),
dataclasses.replace(self.owner, runtime="claude"),
dataclasses.replace(self.owner, pid=124)):
row = self.store.row(new, "🟢 stale")
self.assertEqual(row["color"], "yellow")
self.assertIn("owner_status_unverified", row["warnings"])
def test_dead_writer_cannot_modify_new_terminal_owner(self):
self.store.set(self.owner, "green")
before = self.store.path(self.owner).read_bytes()
self.rows = {124: dataclasses.replace(self.process, pid=124)}
with self.assertRaises(ts.StatusError):
self.store.set(self.owner, "orange")
self.assertEqual(self.store.path(self.owner).read_bytes(), before)
def test_corrupt_record_never_falls_back_to_green(self):
self.store.set(self.owner, "green")
self.store.path(self.owner).write_text("{broken")
self.assertEqual(self.store.row(self.owner, "🟢 stale")["color"], "none")
with self.assertRaises(ts.StatusError):
self.store.repaint(self.owner)
def test_control_sequences_cannot_be_injected_through_labels(self):
for label in ("bad\nlabel", "\x1b]2;bad\x07", "a" * 513):
with self.assertRaises(ts.StatusError):
self.store.set(self.owner, "green", label)
self.assertFalse(self.store.path(self.owner).exists())
def test_legacy_is_runtime_scoped_and_stale_records_are_rejected(self):
self.legacy("claude", "🟠 other runtime")
self.assertIsNone(self.store.read(self.owner)[0])
self.legacy("codex", "🟢 previous session", self.owner.epoch - 10)
self.assertIsNone(self.store.read(self.owner)[0])
self.legacy("codex", "🟢 current")
self.assertEqual(self.store.read(self.owner)[0]["state"], "green")
def test_ambiguous_legacy_gate_never_guesses_green(self):
self.legacy("codex", "🟢 legacy")
row = self.store.row(self.owner, "🟠 pending paste")
self.assertEqual(row["color"], "none")
self.assertEqual(row["status_source"], "legacy_conflict")
self.assertFalse(self.store.path(self.owner).exists())
def test_repaint_never_adopts_unbound_legacy_state(self):
self.legacy("codex", "🟢 fresh but unbound")
with self.assertRaises(ts.StatusError):
self.store.repaint(self.owner)
self.assertFalse(self.paints)
def test_concurrent_writers_keep_the_newest_event_and_valid_json(self):
import concurrent.futures
def update(sequence):
try:
self.store.set(self.owner, "green", str(sequence), issued_ns=sequence)
except ts.StatusError as exc:
self.assertIn("Obsolete status update", str(exc))
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as pool:
list(pool.map(update, [6, 2, 5, 1, 3, 4]))
record, reason = self.store.load(self.owner)
self.assertEqual(reason, "canonical")
self.assertEqual(record["label"], "6")
def test_scan_survives_terminal_api_outage_and_drops_dead_sessions(self):
self.store.set(self.owner, "purple", "Approval needed")
result = ts.scan(self.store, sessions={})
self.assertEqual(result[0]["color"], "purple")
self.assertFalse(result[0]["terminal_visible"])
self.rows = {}
self.assertEqual(ts.scan(self.store, sessions={"ttys010": "🟢 stale"}), [])
def test_scan_elevates_stopped_tab_above_same_colour_and_higher_colours(self):
# TK-11779 follow-up: a preserved-reason needs-Steve stop keeps its base colour but
# must still rank at the top. Build three sessions: a plain orange (PRIORITY 1, the
# highest base colour here), a plain purple (PRIORITY 2), and a purple that was
# explicitly declared needs-Steve (stored purple + variant=stopped).
other = dataclasses.replace(self.process, pid=200, tty="ttys011")
third = dataclasses.replace(self.process, pid=201, tty="ttys012")
self.rows[200] = other
self.rows[201] = third
self.store.set(other.owner(), "orange", "paste waiting") # ttys011
self.store.set(third.owner(), "purple", "plain gated") # ttys012
self.store.set(self.owner, "purple", "TK-11779 · the reason") # ttys010 base
stopped = self.store.set(self.owner, "lightblue") # -> purple+stopped
self.assertEqual((stopped["state"], stopped["variant"]), ("purple", "stopped"))
order = [(r["color"], r.get("variant", "")) for r in ts.scan(self.store, sessions={})]
# The stopped tab leads despite being purple, ahead of both the higher-priority
# plain orange and the same-colour plain purple.
self.assertEqual(order[0], ("purple", "stopped"))
self.assertEqual(order[1:], [("orange", ""), ("purple", "")])
def test_runtime_identification_is_exact_not_command_substrings(self):
self.rows[456] = ts.Process(456, 1, "ttys011", self.owner.started,
"/usr/bin/not-codex")
self.rows[457] = ts.Process(457, 123, "ttys010", self.owner.started, "codex")
self.assertEqual(ts.owners(self.rows), {"ttys010": self.owner})
def test_headless_nested_agent_cannot_paint_parent(self):
self.rows[456] = ts.Process(456, 123, "??", self.owner.started, "codex")
self.rows[os.getpid()] = ts.Process(os.getpid(), 456, "??",
self.owner.started, "python3")
# TK-11672: inject the fixture as the fresh-rescan source so the retry
# cannot escape into the live process table. Without this seam the suite
# inherits whatever real session it runs inside and the assertion below
# silently passes/fails on live ancestry instead of the fixture.
refetch = lambda **_: self.rows
with self.assertRaises(ts.StatusError):
ts.current_owner(self.rows, refetch=refetch)
with patch.dict(os.environ, {"CLAUDE_CODE_CHILD_SESSION": "1"}):
with self.assertRaises(ts.StatusError):
ts.current_owner(self.rows, refetch=refetch)
def test_owner_for_paint_rail_paints_top_level_refuses_subagent(self):
# TK-11791 canonical guard. Ancestry resolves the SAME owner in both
# contexts (a subagent runs inside the parent process); only the
# CLAUDE_CODE_CHILD_SESSION flag distinguishes them, so the flag is the
# rail. Mirrors the CLI `selftest`, kept here so the standard suite covers it.
# TK-12167: CLAUDE_CODE_BRIDGE_SESSION_ID is also scrubbed from the
# baseline `env` so this stays hermetic when the suite runs INSIDE a
# real bridge session's own shell (routine) -- otherwise the ambient
# bridge id would leak into the "subagent" case below and flip it.
owner_proc = ts.Process(4242, 1, "ttys099", self.owner.started, "/usr/bin/claude")
me = ts.Process(os.getpid(), 4242, "??", self.owner.started, "python3")
rows = {4242: owner_proc, os.getpid(): me}
refetch = lambda **_: rows
want = owner_proc.owner()
env = {k: v for k, v in os.environ.items()
if k not in ("CLAUDE_CODE_CHILD_SESSION", "CLAUDE_COLORDOTS_FORCE",
"CLAUDE_CODE_BRIDGE_SESSION_ID")}
with patch.dict(os.environ, env, clear=True):
self.assertEqual(ts.owner_for_paint(rows, refetch=refetch), want)
with patch.dict(os.environ, dict(env, CLAUDE_CODE_CHILD_SESSION="1"), clear=True):
with self.assertRaises(ts.StatusError): # the mirror bug is refused
ts.owner_for_paint(rows, refetch=refetch)
self.assertEqual(ts.owner_for_paint(rows, boot=True, refetch=refetch), want)
self.assertEqual(ts.owner_for_paint(rows, force=True, refetch=refetch), want)
with patch.dict(os.environ,
dict(env, CLAUDE_CODE_CHILD_SESSION="1", CLAUDE_COLORDOTS_FORCE="1"),
clear=True):
self.assertEqual(ts.owner_for_paint(rows, refetch=refetch), want)
def test_owner_for_paint_bridge_session_paints_when_it_owns_its_tty_directly(self):
# TK-12167 (a). Live-evidenced bug: pid 33946 on ttys003, a Remote
# Control bridge session, resolves owner == its own process directly
# (same rows shape as the mirror-bug fixture above) yet was refused
# solely because CLAUDE_CODE_CHILD_SESSION was set. The additional
# CLAUDE_CODE_BRIDGE_SESSION_ID marker (never carried by a plain
# internal Agent-tool subagent) now lets a session that verifiably
# owns its own pane paint.
owner_proc = ts.Process(4242, 1, "ttys099", self.owner.started, "/usr/bin/claude")
me = ts.Process(os.getpid(), 4242, "??", self.owner.started, "python3")
rows = {4242: owner_proc, os.getpid(): me}
refetch = lambda **_: rows
want = owner_proc.owner()
env = {k: v for k, v in os.environ.items()
if k not in ("CLAUDE_CODE_CHILD_SESSION", "CLAUDE_COLORDOTS_FORCE",
"CLAUDE_CODE_BRIDGE_SESSION_ID")}
with patch.dict(os.environ,
dict(env, CLAUDE_CODE_CHILD_SESSION="1",
CLAUDE_CODE_BRIDGE_SESSION_ID="session_test"),
clear=True):
self.assertEqual(ts.owner_for_paint(rows, refetch=refetch), want)
def test_owner_for_paint_bridge_marker_never_rescues_an_intermediate_agent(self):
# TK-12167 (b). A genuinely nested/headless child claude process (its
# own separate pid, no tty of its own) sitting between the caller and
# the real tty owner must stay refused even with a bridge id present --
# the marker identifies bridge IDENTITY, not a free pass past a real
# intermediate agent. current_owner() already raises for this shape
# (a headless/nested-through-a-separate-process agent), before
# owner_for_paint's bridge check is ever reached.
owner_proc = ts.Process(4242, 1, "ttys098", self.owner.started, "/usr/bin/claude")
intermediate = ts.Process(9000, 4242, "??", self.owner.started, "/usr/bin/claude")
me = ts.Process(os.getpid(), 9000, "??", self.owner.started, "python3")
rows = {4242: owner_proc, 9000: intermediate, os.getpid(): me}
refetch = lambda **_: rows
env = {k: v for k, v in os.environ.items()
if k not in ("CLAUDE_CODE_CHILD_SESSION", "CLAUDE_COLORDOTS_FORCE",
"CLAUDE_CODE_BRIDGE_SESSION_ID")}
with patch.dict(os.environ,
dict(env, CLAUDE_CODE_CHILD_SESSION="1",
CLAUDE_CODE_BRIDGE_SESSION_ID="session_test"),
clear=True):
with self.assertRaises(ts.StatusError):
ts.owner_for_paint(rows, refetch=refetch)
def test_owner_for_paint_bridge_marker_never_rescues_undeterminable_ownership(self):
# TK-12167 (c). No claude/codex anywhere in ancestry at all (ownership
# cannot be determined) → refused, fail CLOSED, even with a bridge id.
me = ts.Process(os.getpid(), 1, "??", self.owner.started, "python3")
rows = {os.getpid(): me}
refetch = lambda **_: rows
env = {k: v for k, v in os.environ.items()
if k not in ("CLAUDE_CODE_CHILD_SESSION", "CLAUDE_COLORDOTS_FORCE",
"CLAUDE_CODE_BRIDGE_SESSION_ID")}
with patch.dict(os.environ,
dict(env, CLAUDE_CODE_CHILD_SESSION="1",
CLAUDE_CODE_BRIDGE_SESSION_ID="session_test"),
clear=True):
with self.assertRaises(ts.StatusError):
ts.owner_for_paint(rows, refetch=refetch)
def test_paint_failure_is_reported_and_auditable(self):
def fail(*args):
raise OSError("fixture: unavailable terminal")
self.store.painter = fail
with self.assertRaises(ts.StatusError):
self.store.set(self.owner, "green")
r, _ = self.store.load(self.owner)
self.assertEqual(r["render_status"], "failed")
self.assertIn("display_update_failed", self.store.row(self.owner)["warnings"])
def test_complete_osc_packet_contains_one_consistent_title_and_badge(self):
r = self.store.set(self.owner, "yellow", "Choose direction")
packet = ts.osc_payload(r)
self.assertIn(b"green;brightness;204", packet)
# yellow is a waiting state, so the window-title and badge both carry the pulse
# marker (display_title), and must still be BYTE-IDENTICAL to each other.
self.assertEqual(packet.count(ts.display_title(r).encode()), 2)
self.assertIn(b"SetBadgeFormat=", packet)
def test_waiting_states_pulse_and_solid_states_stay_solid(self):
# Steve's 2026-09-15 dot-flash directive: green/pink stay solid; the four needs-Steve
# states carry the render-time pulse marker on BOTH the tab badge and the bar label.
frames = set(ts.PULSE_FRAMES)
for color in ("yellow", "purple", "orange", "lightblue"):
with self.subTest(color=color):
rec = self.store.set(self.owner, color, "the reason")
# the STORED/validated title stays marker-free (load() stays byte-identical)
self.assertFalse(frames & set(rec["title"]), "stored title must not carry the marker")
# the rendered tab badge/window-title pulses
self.assertTrue(frames & set(ts.osc_payload(rec).decode()), "tab render must pulse")
# the desktop-bar / allcolordots label pulses
self.assertTrue(frames & set(self.store.row(self.owner)["label"]), "bar label must pulse")
self.store.set(self.owner, "none")
for color, seed in (("green", "Working"), ("pink", "parked")):
with self.subTest(color=color):
rec = self.store.set(self.owner, color, seed)
self.assertFalse(frames & set(ts.osc_payload(rec).decode()), "solid state must not pulse")
self.assertFalse(frames & set(self.store.row(self.owner)["label"]), "solid bar must not pulse")
self.store.set(self.owner, "none")
def test_pulse_marker_toggles_across_legitimate_repaints(self):
# "Pulse the badge char, no loop": the two-frame marker advances ONLY on a legitimate
# new event (revision++), never on a timer. A plain repaint keeps the same frame.
a = self.store.set(self.owner, "yellow", "q1")
b = self.store.set(self.owner, "yellow", "q2")
fa = set(ts.osc_payload(a).decode()) & set(ts.PULSE_FRAMES)
fb = set(ts.osc_payload(b).decode()) & set(ts.PULSE_FRAMES)
self.assertTrue(fa and fb and fa != fb, "badge char toggles frames across repaints")
again = ts.osc_payload(self.store.repaint(self.owner)).decode()
self.assertEqual(set(again) & set(ts.PULSE_FRAMES), fb, "a bare repaint does not advance the frame")
def test_two_ttys_do_not_share_status(self):
second = dataclasses.replace(self.process, pid=124, tty="ttys011")
self.rows[124] = second
self.store.set(self.owner, "green", "First")
self.store.set(second.owner(), "purple", "Second")
self.assertEqual(self.store.row(self.owner)["color"], "green")
self.assertEqual(self.store.row(second.owner())["color"], "purple")
def test_busy_target_skips_fast_with_short_lock_wait(self):
# TK-11835 NEGATIVE TEST: a cross-tty caller must SKIP a busy target FAST
# (short lock_wait), not inherit the 65s self-writer backstop that made a
# sweep hang on every mid-turn tab. Hold the target's per-tty lock, then a
# set(lock_wait=short) must raise a "busy ... skipped" StatusError WELL
# within the budget -- not block for 65s. Goes RED if the wait override
# regresses (a bare set() would wait the full backstop and this fails on
# the elapsed-time bound / or hang the suite past the injected fault).
import fcntl
self.store.set(self.owner, "green") # create the record + lockdir
lockpath = self.store.root / ".locks" / (self.owner.tty + ".lock")
held = open(lockpath, "a")
fcntl.flock(held, fcntl.LOCK_EX | fcntl.LOCK_NB)
try:
start = ts.time.monotonic()
with self.assertRaises(ts.StatusError) as cm:
self.store.set(self.owner, "purple", lock_wait=0.2)
elapsed = ts.time.monotonic() - start
self.assertIn("busy", str(cm.exception))
self.assertIn("skipped", str(cm.exception))
self.assertIn(self.owner.tty, str(cm.exception))
self.assertLess(elapsed, 5.0, "fail-fast: nowhere near the 65s backstop")
finally:
fcntl.flock(held, fcntl.LOCK_UN)
held.close()
# Positive control: with the lock free the same short-budget paint lands.
record = self.store.set(self.owner, "purple", lock_wait=0.2)
self.assertEqual(record["state"], "purple")
class TicketPromptHelperTests(unittest.TestCase):
"""TK-12168: the pure, side-effect-free detectors cmd_auto_bind builds on."""
def test_explicit_ticket_extraction(self):
self.assertEqual(tb.explicit_ticket_in_prompt("work on TK-500 please"), "TK-500")
self.assertEqual(tb.explicit_ticket_in_prompt("tk-500 lowercase too"), "TK-500")
self.assertEqual(tb.explicit_ticket_in_prompt("no ticket mentioned here"), "")
self.assertEqual(tb.explicit_ticket_in_prompt(""), "")
self.assertEqual(tb.explicit_ticket_in_prompt(None), "")
# Two distinct ids -> ambiguous, abstain (same rule as AGENT/the bare-id
# scan in discover()).
self.assertEqual(tb.explicit_ticket_in_prompt("TK-500 or TK-501?"), "")
# The SAME id repeated is not ambiguous.
self.assertEqual(tb.explicit_ticket_in_prompt("TK-500 ... yeah TK-500"), "TK-500")
def test_trivial_prompt_guard(self):
trivial = ["", " ", "ok", "OK", "Ok.", "yes", "yes.", "y", "no", "n",
"thanks", "k", "/foo", "/status", "hi there", "two words"]
for prompt in trivial:
with self.subTest(prompt=prompt):
self.assertTrue(tb.is_trivial_prompt(prompt), prompt)
substantive = [
"fix the login bug on checkout",
"/deploy the checkout service now",
"why is the build failing today",
"TK-500 needs a follow-up patch",
]
for prompt in substantive:
with self.subTest(prompt=prompt):
self.assertFalse(tb.is_trivial_prompt(prompt), prompt)
def test_short_topic_collapses_and_caps(self):
self.assertEqual(tb.short_topic(" fix the\nlogin bug "), "fix the login bug")
long_prompt = "x " * 100
self.assertEqual(len(tb.short_topic(long_prompt, limit=70)), 70)
class AutoBindTests(unittest.TestCase):
"""TK-12168: a plain top-level interactive session that never ran `tk` and
carries no TK- in its argv sits unbound ("TK REQUIRED") forever, because
ticket_binding.discover() is deliberately read-only. cmd_auto_bind is the
UserPromptSubmit-hook orchestration that closes that gap.
NEGATIVE-TEST NOTE (CLAUDE.md TK-11431 amendment 3): every test below
exercises `ts.cmd_auto_bind`, which did not exist before this change --
on the pre-TK-12168 tree these all fail with
`AttributeError: module 'terminal_status' has no attribute 'cmd_auto_bind'`,
which is exactly the "10/14 tabs show TK REQUIRED forever" bug this closes.
"""
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
self.process = ts.Process(321, 1, "ttys077",
"Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
self.rows = {321: self.process}
self.owner = self.process.owner()
self.paints = []
self.store = ts.Store(self.root, lambda: self.rows,
lambda owner, record: self.paints.append(record.copy()))
self.store.known_tickets = {"TK-500", "TK-12168"}
self.tk_new_calls = []
def fake_tk_new(self, ticket_id="TK-99999", raises=None):
def _fn(owner, topic, project):
self.tk_new_calls.append((owner, topic, project))
if raises:
raise raises
return ticket_id
return _fn
def test_unbound_plus_substantive_prompt_creates_and_binds(self):
result = ts.cmd_auto_bind(self.store, self.owner,
"fix the login bug on the checkout page",
"my-project", tk_new=self.fake_tk_new("TK-99999"))
self.assertEqual(result, {"action": "created", "ticket": "TK-99999"})
self.assertEqual(len(self.tk_new_calls), 1)
_, topic, project = self.tk_new_calls[0]
self.assertEqual(project, "my-project")
self.assertIn("login bug", topic)
record, _ = self.store.load(self.owner)
self.assertEqual(record["ticket"], "TK-99999")
# A second call must see the session as already bound and never call
# tk_new again (the SAME "unbound" check that skipped it before).
result2 = ts.cmd_auto_bind(self.store, self.owner, "another real task here",
"my-project", tk_new=self.fake_tk_new())
self.assertEqual(result2["action"], "already-bound")
self.assertEqual(len(self.tk_new_calls), 1)
def test_trivial_prompts_never_create_a_ticket(self):
for prompt in ("", "ok", "Ok", "yes", " yes ", "/foo", "/foo\n", "hi there"):
with self.subTest(prompt=prompt):
result = ts.cmd_auto_bind(self.store, self.owner, prompt, "proj",
tk_new=self.fake_tk_new())
self.assertEqual(result["action"], "skipped-trivial", prompt)
self.assertEqual(self.tk_new_calls, [])
record, reason = self.store.load(self.owner)
self.assertEqual(reason, "missing") # never wrote anything
def test_slash_command_with_arguments_is_substantive(self):
# A bare "/foo" is trivial (no task named); "/foo do the actual thing"
# names real work and must still earn a ticket.
result = ts.cmd_auto_bind(self.store, self.owner, "/deploy the checkout service",
"proj", tk_new=self.fake_tk_new("TK-1"))
self.assertEqual(result["action"], "created")
self.assertEqual(len(self.tk_new_calls), 1)
def test_explicit_ticket_in_prompt_binds_without_creating(self):
result = ts.cmd_auto_bind(self.store, self.owner,
"let's keep working on TK-500 today",
"proj", tk_new=self.fake_tk_new())
self.assertEqual(result, {"action": "bound-explicit", "ticket": "TK-500"})
self.assertEqual(self.tk_new_calls, []) # never shells out when the user named one
record, _ = self.store.load(self.owner)
self.assertEqual(record["ticket"], "TK-500")
self.assertEqual(record["ticket_source"], "explicit")
def test_explicit_ticket_not_in_known_ledger_fails_open(self):
result = ts.cmd_auto_bind(self.store, self.owner,
"let's work on TK-777777 today",
"proj", tk_new=self.fake_tk_new())
self.assertEqual(result["action"], "explicit-bind-failed")
self.assertEqual(result["ticket"], "TK-777777")
record, reason = self.store.load(self.owner)
self.assertEqual(reason, "missing") # refused, never fabricated a binding
def test_ambiguous_multiple_tickets_in_prompt_falls_through_to_create(self):
# Two distinct TK- mentions -> abstain on "which one", same conservatism
# as every other detector in ticket_binding.py -- falls through to the
# substantive-prompt create path instead of guessing.
result = ts.cmd_auto_bind(self.store, self.owner,
"is this related to TK-500 or TK-501 or something else",
"proj", tk_new=self.fake_tk_new("TK-2"))
self.assertEqual(result["action"], "created")
def test_never_creates_a_duplicate_for_the_same_session(self):
r1 = ts.cmd_auto_bind(self.store, self.owner, "build the new export feature",
"proj", tk_new=self.fake_tk_new("TK-1"))
self.assertEqual(r1["action"], "created")
# Simulate the record's own ticket field going empty again (e.g. an
# explicit clear) WITHOUT clearing our autobind state file -- a second
# call must REBIND the same id, never mint TK-2.
self.store.set(self.owner, "none", ticket_update="")
r2 = ts.cmd_auto_bind(self.store, self.owner, "build the new export feature again",
"proj", tk_new=self.fake_tk_new("TK-2"))
self.assertEqual(r2, {"action": "rebound", "ticket": "TK-1"})
self.assertEqual(len(self.tk_new_calls), 1) # only the first call actually created
def test_concurrent_prompt_on_same_tty_cannot_double_create(self):
import fcntl
# Hold the lock the way a concurrent auto-bind invocation would.
state_path = ts._autobind_state_path(self.store, self.owner)
lock_path = state_path.with_suffix(".lock")
fd = os.open(str(lock_path), os.O_CREAT | os.O_RDWR, 0o600)
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
try:
result = ts.cmd_auto_bind(self.store, self.owner, "do the real work now",
"proj", tk_new=self.fake_tk_new())
finally:
fcntl.flock(fd, fcntl.LOCK_UN)
os.close(fd)
self.assertEqual(result, {"action": "locked"})
self.assertEqual(self.tk_new_calls, [])
def test_fails_open_when_tk_errors_and_rate_limits_the_retry(self):
boom = self.fake_tk_new(raises=ts.StatusError("tk: connection refused"))
r1 = ts.cmd_auto_bind(self.store, self.owner, "do the real work now",
"proj", tk_new=boom, now=1000.0)
self.assertEqual(r1["action"], "create-failed")
self.assertIn("connection refused", r1["error"])
record, reason = self.store.load(self.owner)
self.assertEqual(reason, "missing") # a failed create never fakes a binding
# A retry moments later is throttled -- never hammers a broken `tk`.
r2 = ts.cmd_auto_bind(self.store, self.owner, "do the real work now",
"proj", tk_new=boom, now=1005.0)
self.assertEqual(r2["action"], "rate-limited")
self.assertEqual(len(self.tk_new_calls), 1)
# Well past the rate-limit window, it tries again.
r3 = ts.cmd_auto_bind(self.store, self.owner, "do the real work now",
"proj", tk_new=self.fake_tk_new("TK-3"), now=1000.0 + 3600)
self.assertEqual(r3["action"], "created")
def test_trivial_prompt_never_consumes_the_rate_limit_window(self):
# A trivial "ok" right before the real prompt must not block it.
r1 = ts.cmd_auto_bind(self.store, self.owner, "ok", "proj",
tk_new=self.fake_tk_new(), now=1000.0)
self.assertEqual(r1["action"], "skipped-trivial")
r2 = ts.cmd_auto_bind(self.store, self.owner, "now actually fix the bug",
"proj", tk_new=self.fake_tk_new("TK-4"), now=1000.1)
self.assertEqual(r2["action"], "created")
def test_never_binds_an_empty_ticket_id(self):
# A helper that returns "" instead of raising (defensive belt-and-
# suspenders on _default_tk_new's own contract) must fail open, not
# bind a blank ticket that would look like a real dot with none at all.
result = ts.cmd_auto_bind(self.store, self.owner, "do the real work now",
"proj", tk_new=lambda o, t, p: "")
self.assertEqual(result["action"], "create-failed")
record, reason = self.store.load(self.owner)
self.assertEqual(reason, "missing")
class EnrichmentCacheTests(unittest.TestCase):
"""TK-11831: the shared pid-keyed cache for discover()'s two `ps` enrichment calls.
These ran on EVERY self-paint (every UserPromptSubmit AND every Stop hook)
across ~68 concurrent sessions -- measured 2.46s each at 18 live owners and
~4.4s under overnight load, which is the `ps` storm behind this ticket.
"""
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.root = Path(self.temp.name)
events = self.root / ".claude/tickets/events.jsonl"
events.parent.mkdir(parents=True, exist_ok=True)
events.write_text("") # discover() returns early without this
cache = self.root / "psenrich.json"
patcher = patch.object(tb, "_PS_CACHE_PATH", str(cache))
patcher.start()
self.addCleanup(patcher.stop)
self.cache = cache
# A real, live pid so `ps` genuinely answers: this process.
started = dt.datetime.fromtimestamp(
os.path.getmtime(__file__)).strftime("%a %b %d %H:%M:%S %Y")
self.proc = ts.Process(os.getpid(), 1, "ttys010", started, "/bin/claude")
self.rows = {self.proc.pid: self.proc}
self.live = {self.proc.tty: self.proc.owner()}
def discover(self):
return tb.discover(self.root, self.rows, self.live, ts.ancestors)
def counted(self):
"""Count only the enrichment `ps` calls discover() shells out to."""
real, calls = tb.subprocess.run, []
def spy(cmd, *a, **kw):
if isinstance(cmd, (list, tuple)) and cmd and cmd[0] == "ps":
calls.append(list(cmd))
return real(cmd, *a, **kw)
return patch.object(tb.subprocess, "run", spy), calls
def test_warm_cache_is_exact_and_shells_out_zero_times(self):
# The whole claim in one test: a second pass must return the IDENTICAL
# answer while making NO `ps` call at all. Goes RED if the cache is
# bypassed (calls reappear) or if it is lossy (answers diverge).
spy, calls = self.counted()
with spy:
cold, cold_known = self.discover()
self.assertEqual(len(calls), 2, "cold pass makes both enrichment reads")
calls.clear()
warm, warm_known = self.discover()
self.assertEqual(calls, [], "warm pass must not touch ps at all")
self.assertEqual(cold, warm)
self.assertEqual(cold_known, warm_known)
self.assertTrue(self.cache.exists(), "the warm pass had something to hit")
def test_cache_key_is_pid_reuse_safe(self):
# NEGATIVE TEST (CLAUDE.md TK-11431 amendment 3). argv/env are immutable
# per process, which is what makes caching EXACT -- but only while the
# pid means the same process. A recycled pid carries a DIFFERENT start
# time, so it must MISS and be re-read; serving the dead process's argv
# would bind a tab to a ticket that is not running. Mutation-verified:
# dropping `started` from _enrich_key() makes this go RED.
self.discover() # populate
recycled = dataclasses.replace(self.proc, started="Wed Sep 9 08:35:08 2026")
self.rows = {recycled.pid: recycled}
self.live = {recycled.tty: recycled.owner()}
spy, calls = self.counted()
with spy:
self.discover()
self.assertEqual(len(calls), 2,
"a recycled pid must re-read ps, never serve the dead pid's argv")
def test_unreadable_cache_degrades_to_the_uncached_answer(self):
# Fail-open: a torn or hostile cache file must cost correctness nothing.
# Every cache fault degrades to today's behaviour, exactly as a slow ps
# already degrades the label but never the paint.
expected, _ = self.discover()
for junk in ("", "{", "null", '{"entries": "not-a-dict"}', '{"entries": {"x": 7}}'):
self.cache.write_text(junk)
self.assertEqual(self.discover()[0], expected, "corrupt cache: " + repr(junk))
def test_inconclusive_read_is_never_cached_as_a_negative(self):
# CODEX-CHECK FINDING (sticky negative cache). If a pid is absent from the
# ps output -- it exited mid-call, or ps was truncated -- that is "not yet
# known", NOT "known to have no argv". Caching "" for it would make ONE
# transient miss STICKY for the whole TTL, leaving that tab unbound for an
# hour, where the uncached path simply retried on the next paint. A cache
# must never be WORSE than the thing it replaces.
real = tb.subprocess.run
def blind(cmd, *a, **kw):
out = real(cmd, *a, **kw)
if isinstance(cmd, (list, tuple)) and cmd and cmd[0] == "ps":
return subprocess.CompletedProcess(cmd, 0, "", "") # pid absent
return out
with patch.object(tb.subprocess, "run", blind):
self.discover()
spy, calls = self.counted()
with spy:
self.discover()
self.assertEqual(len(calls), 2,
"an inconclusive read must be retried, not cached as empty")
def test_merge_never_downgrades_a_peer_value(self):
# CODEX-CHECK FINDING (poison overwrite). ~68 processes read-modify-write
# this file with no lock. A slow writer merging onto its START-OF-PASS
# snapshot could blank an entry a peer had already learned -- and a blanked
# entry is a WRONG ticket binding, not merely a missed cache. Values are
# immutable per process, so "never downgrade" makes concurrent writes
# commutative: no writer can lose information, lock-free.
tb._enrich_cache_merge({"K": {"a": "claude TK-1", "s": "REAL"}})
tb._enrich_cache_merge({"K": {"a": "", "s": ""}}) # the poison pass
kept = tb._enrich_cache_read()["K"]
self.assertEqual(kept["a"], "claude TK-1")
self.assertEqual(kept["s"], "REAL", "empty must never overwrite a real answer")
def test_merge_reads_fresh_and_keeps_a_concurrent_peers_entry(self):
# The other half of the same race: merging must start from the file as it
# is NOW, not from this pass's stale snapshot, or a peer's brand-new key is
# silently dropped by the full-file replace.
stale = tb._enrich_cache_read() # empty snapshot, as a slow pass holds
tb._enrich_cache_merge({"PEER": {"s": "PEER-TSID"}}) # peer lands mid-flight
tb._enrich_cache_merge({"MINE": {"s": "MY-TSID"}}) # our late write-back
self.assertEqual(stale, {})
after = tb._enrich_cache_read()
self.assertIn("PEER", after, "a concurrent peer's entry must survive our write")
self.assertIn("MINE", after)
def test_non_string_cache_values_are_not_served(self):
# CODEX-CHECK FINDING: the uncached path can only ever put a str into the
# argv map. A hand-edited {"a": null} served as a hit would be a cache
# answer that DIFFERS from the uncached one -- the one thing this may
# never do -- and would blow up downstream string ops.
self.cache.write_text(json.dumps({"entries": {
"K": {"a": None, "s": 7, "t": tb.time.time()}}}))
self.assertEqual(tb._enrich_cache_read(), {}, "typed junk is not a hit")
def test_stale_entries_age_out(self):
# A pid-keyed cache on a box up for days must not grow without bound or
# serve an entry whose process is long gone; the TTL is the backstop.
self.discover()
with patch.object(tb, "_PS_CACHE_TTL", -1):
spy, calls = self.counted()
with spy:
self.discover()
self.assertEqual(len(calls), 2, "an expired entry must be re-read")
class ItermEnumCacheTests(unittest.TestCase):
"""TK-11879: the iTerm session enumeration collapses the herd through a
cross-process cache and degrades to that cache when a live enumeration goes
blind, instead of the {} -> confidently-wrong-green fall-through."""
def setUp(self):
self.temp = tempfile.TemporaryDirectory()
self.addCleanup(self.temp.cleanup)
self.cache = Path(self.temp.name) / "iterm.json"
p = patch.object(ts, "_iterm_cache_path", lambda: self.cache)
p.start()
self.addCleanup(p.stop)
# TK-11831 single-flight lock isolated to this test's tempdir.
self.lock = Path(self.temp.name) / "iterm.lock"
lp = patch.object(ts, "_iterm_lock_path", lambda: self.lock)
lp.start()
self.addCleanup(lp.stop)
# HOME redirected so a test blind's enum_blind trail never pollutes the real file.
hp = patch.dict(os.environ, {"HOME": self.temp.name})
hp.start()
self.addCleanup(hp.stop)
def _osascript(self, *, rc, stdout=""):
"""A subprocess.run stub that answers ONLY the osascript enumeration and
records how many times it was called (to prove the herd was collapsed)."""
self.calls = 0
real = subprocess.run
def fake(cmd, *a, **k):
if cmd[:1] == ["osascript"]:
self.calls += 1
return subprocess.CompletedProcess(cmd, rc, stdout, "")
return real(cmd, *a, **k)
return patch.object(ts.subprocess, "run", fake)
def test_success_enumerates_once_then_siblings_hit_the_cache(self):
out = "/dev/ttys010\t🟢 one\n/dev/ttys011\t🟣 two\n"
with self._osascript(rc=0, stdout=out):
first, api1 = ts.iterm_sessions()
second, api2 = ts.iterm_sessions()
third, api3 = ts.iterm_sessions()
self.assertEqual(first, {"ttys010": "🟢 one", "ttys011": "🟣 two"})
self.assertEqual((api1, api2, api3), ("available", "cached", "cached"))
self.assertEqual(second, first)
# osascript ran exactly ONCE for three enumerations -- the herd collapsed.
self.assertEqual(self.calls, 1)
def test_fresh_forces_a_live_enumeration_past_the_cache(self):
with self._osascript(rc=0, stdout="/dev/ttys010\t🟢 one\n"):
ts.iterm_sessions() # warms cache
_, api = ts.iterm_sessions(fresh=True) # must bypass it
self.assertEqual(api, "available")
self.assertEqual(self.calls, 2)
def test_blind_falls_back_to_the_cached_map_not_green(self):
# Seed a real recent map, then a live enumeration goes blind (rc 1).
with self._osascript(rc=0, stdout="/dev/ttys010\t🟢 one\n"):
ts.iterm_sessions()
with patch.object(ts, "_ITERM_TTL", -1): # force past the fresh TTL
with self._osascript(rc=1):
sessions, api = ts.iterm_sessions()
self.assertEqual(api, "stale")
self.assertEqual(sessions, {"ttys010": "🟢 one"}) # REAL map, not {}
# A recovered blind must NOT pollute the health failure trail.
trail = Path(self.temp.name) / ".claude/skills/terminal-status-health/data/repaint-failures.jsonl"
self.assertFalse(trail.exists(),
"a blind recovered from cache must not record enum_blind")
def test_truly_blind_with_no_cache_records_enum_blind_and_is_unavailable(self):
# NEGATIVE / fault-injection (CLAUDE.md TK-11431 amendment 3): with NO usable
# cache the detector MUST still fire -- record enum_blind + return unavailable.
with self._osascript(rc=1):
sessions, api = ts.iterm_sessions()
self.assertEqual((sessions, api), ({}, "unavailable"))
trail = Path(self.temp.name) / ".claude/skills/terminal-status-health/data/repaint-failures.jsonl"
self.assertTrue(trail.exists(), "a genuine blind MUST record enum_blind")
rec = json.loads(trail.read_text().strip().splitlines()[-1])
self.assertEqual(rec["rc"], "enum_blind")
def test_stale_map_past_the_bound_is_not_trusted(self):
with self._osascript(rc=0, stdout="/dev/ttys010\t🟢 one\n"):
ts.iterm_sessions()
# Both the fresh TTL and the stale bound are in the past -> no fallback.
with patch.object(ts, "_ITERM_TTL", -1), patch.object(ts, "_ITERM_STALE_MAX", -1):
with self._osascript(rc=1):
sessions, api = ts.iterm_sessions()
self.assertEqual((sessions, api), ({}, "unavailable"))
def test_empty_successful_enum_is_not_cached_so_siblings_re_enumerate(self):
# rc 0 but zero sessions is a racey/partial answer; caching it would poison
# siblings with `cached {}` for a whole TTL and record no enum_blind.
with self._osascript(rc=0, stdout=""):
first, api1 = ts.iterm_sessions()
second, api2 = ts.iterm_sessions()
self.assertEqual((first, api1), ({}, "available"))
# The cache was NOT written, so the sibling enumerates again (not "cached").
self.assertFalse(self.cache.exists())
self.assertEqual((second, api2), ({}, "available"))
self.assertEqual(self.calls, 2)
def test_single_flight_peer_holding_lock_coalesces_and_fires_no_osascript(self):
# TK-11831: a peer is already enumerating (holds the lock). A fresh map it
# publishes must be COALESCED without this caller firing its own osascript
# into iTerm's serial queue -- the herd collapses to ONE round-trip.
import fcntl as _f
fd = os.open(str(self.lock), os.O_CREAT | os.O_RDWR, 0o600)
_f.flock(fd, _f.LOCK_EX | _f.LOCK_NB) # simulate the peer/winner
# The winner has just published its map (real file, fresh mtime for the
# wait loop's mtime>=started gate).
self.cache.write_text(json.dumps({"sessions": {"ttys010": "🟢 one"}}))
fresh = ({"ttys010": "🟢 one"}, 0.0)
# Initial lookup MISSES (forces single-flight); the wait-loop lookup HITS
# (the winner has published) -> deterministic coalesce.
calls = {"n": 0}
def staged_get(_max_age):
calls["n"] += 1
return None if calls["n"] == 1 else fresh
try:
with self._osascript(rc=0, stdout="/dev/ttysZZZ\tSHOULD-NOT-RUN\n"):
with patch.object(ts, "_iterm_cache_get", staged_get), \
patch.object(ts, "_ITERM_SF_WAIT", 1.0):
sessions, api = ts.iterm_sessions()
finally:
os.close(fd)
# ZERO osascript calls -- the whole point of single-flight -- and the
# coalesced map is returned as a cache result.
self.assertEqual(self.calls, 0,
"a peer already enumerating must not trigger a second osascript")
self.assertEqual((sessions, api), ({"ttys010": "🟢 one"}, "cached"))
def test_single_flight_peer_lock_no_fresh_map_blinds_without_osascript(self):
# Peer holds the lock but publishes nothing usable within the wait -> this
# caller degrades to the SAME blind path an osascript timeout takes, and
# STILL fires no osascript (never a second round-trip on the busy queue).
import fcntl as _f
fd = os.open(str(self.lock), os.O_CREAT | os.O_RDWR, 0o600)
_f.flock(fd, _f.LOCK_EX | _f.LOCK_NB)
try:
with self._osascript(rc=0, stdout="/dev/ttysZZZ\tSHOULD-NOT-RUN\n"):
with patch.object(ts, "_ITERM_SF_WAIT", 0.3):
sessions, api = ts.iterm_sessions()
finally:
os.close(fd)
self.assertEqual(self.calls, 0)
self.assertEqual((sessions, api), ({}, "unavailable"))
def test_another_users_cache_file_is_never_trusted(self):
with self._osascript(rc=0, stdout="/dev/ttys010\t🟢 one\n"):
ts.iterm_sessions()
# Simulate a file owned by another uid: _iterm_cache_get rejects it.
real_uid = os.getuid()
with patch.object(ts.os, "getuid", lambda: real_uid + 99999):
self.assertIsNone(ts._iterm_cache_get(ts._ITERM_TTL))
if __name__ == "__main__":
unittest.main(verbosity=2)