← back to Terminal Status
TK-11794: pulse the 4 needs-Steve dot states; green/pink stay solid
585ee0698b5d82e927fdc708a1298c43f9d43e3e · 2026-09-15 18:33:17 -0700 · vp-engineering
Render-time only: the waiting states (yellow/purple/orange/lightblue) carry a
two-frame attention marker (revision-keyed) in the tab badge; green/pink render
unchanged. Marker is NOT in the stored/validated title, so load() recomputes
byte-identically and no existing record goes invalid. Zero background loops,
zero new subprocess/scan cost. The real animated pulse lives in desktop-dotbar.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0169ph94FSR3sVnot1p5pNX1
Files touched
M terminal_status.pyM test_terminal_status.py
Diff
commit 585ee0698b5d82e927fdc708a1298c43f9d43e3e
Author: vp-engineering <steve@designerwallcoverings.com>
Date: Tue Sep 15 18:33:17 2026 -0700
TK-11794: pulse the 4 needs-Steve dot states; green/pink stay solid
Render-time only: the waiting states (yellow/purple/orange/lightblue) carry a
two-frame attention marker (revision-keyed) in the tab badge; green/pink render
unchanged. Marker is NOT in the stored/validated title, so load() recomputes
byte-identically and no existing record goes invalid. Zero background loops,
zero new subprocess/scan cost. The real animated pulse lives in desktop-dotbar.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0169ph94FSR3sVnot1p5pNX1
---
terminal_status.py | 169 ++++++++++++++++++++++++++++++++++++++++++++++--
test_terminal_status.py | 60 ++++++++++++++++-
2 files changed, 224 insertions(+), 5 deletions(-)
diff --git a/terminal_status.py b/terminal_status.py
index 769c39d..7f23d36 100644
--- a/terminal_status.py
+++ b/terminal_status.py
@@ -61,6 +61,18 @@ COLORS = {
}
PRIORITY = {name: i for i, name in enumerate(
("lightblue", "orange", "purple", "yellow", "green", "pink", "none"))}
+
+# Steve's 2026-09-15 dot-flash directive: green (WORKING) and pink (PARKED) stay SOLID; the
+# four needs-Steve states PULSE. A true timed blink of the TAB itself is impossible without a
+# background per-tab loop -- and that loop IS the orphaned-stuck-color mechanism Steve's memory
+# warns against (flash-red loops, tab-flash.pid drift). So the tab side is honest: the badge
+# char carries a two-frame ATTENTION marker toggled by the ALREADY-PERSISTED `revision` counter,
+# advancing only on a legitimate repaint -- never on a timer, never a spawned loop, zero new
+# scan/subprocess cost. The REAL animated pulse lives on the desktop-dotbar (one already-running
+# process animating the waiting rows in CSS). Both frames are attention glyphs, so even a session
+# that never repaints still reads unmistakably as "waiting", just not mid-animation.
+WAITING_STATES = ("yellow", "purple", "orange", "lightblue")
+PULSE_FRAMES = ("✧", "✦") # ✧ hollow / ✦ filled — a twinkle that toggles across repaints
TTY = re.compile(r"ttys[0-9]+")
@@ -120,16 +132,32 @@ def issued_clock():
STOPPED_MARK = "\U0001F6D1" # 🛑 — "stopped, needs Steve", shown NEXT TO the base dot
-def status_title(color, label, ticket="", variant=""):
+def status_title(color, label, ticket="", variant="", pulse=None):
# variant "stopped" (Steve, 2026-09-11): keep the ORIGINAL colour dot and place the
# blue dot next to it — additive marker, never a replacement. Reverts to the base dot
# alone as soon as the session is running again.
+ #
+ # `pulse` is a RENDER-ONLY concern (the record's revision counter). It stays None for the
+ # STORED/validated title so load() recomputes byte-identically and no existing record goes
+ # invalid; it is passed the revision ONLY at the two render sites (the OSC badge below and
+ # Store.row() for the bar), where it adds the waiting-state two-frame attention marker.
if color == "none":
return ""
dot = COLORS[color][0] + (STOPPED_MARK if variant == "stopped" else "")
+ if pulse is not None and color in WAITING_STATES:
+ dot += PULSE_FRAMES[pulse % 2]
return dot + " " + " · ".join(p for p in (ticket, label) if p)
+def display_title(record):
+ """The tab/bar RENDER title: the stored title PLUS the waiting-state pulse marker.
+ record['title'] itself is never mutated (it is load-validated), so the marker lives only
+ at render time — here for the OSC badge/window-title, and in Store.row() for the desktop
+ bar. For green/pink/none this returns exactly record['title']."""
+ return status_title(record["state"], record["label"], record.get("ticket", ""),
+ record.get("variant", ""), pulse=record.get("revision", 0))
+
+
def label_ticket(label):
label = label.strip(" ·")
match = re.match(r"^(TK-\d+)(?=$|[ ·:-])", label, re.I)
@@ -425,6 +453,111 @@ def current_owner(rows, _retried=False, refetch=processes):
raise StatusError("No owning Claude/Codex terminal in this process ancestry")
+def paint_forced():
+ return os.environ.get("CLAUDE_COLORDOTS_FORCE") == "1"
+
+
+def owner_for_paint(rows, force=False, boot=False, refetch=processes):
+ # CANONICAL paintability guard — the SINGLE decision both engines defer to
+ # (color.sh's /color hue painter AND this engine's dot setters), so they
+ # paint-or-refuse IDENTICALLY in the same context (TK-11791 consolidation).
+ #
+ # Ancestry (current_owner) resolves WHICH tty and refuses a truly-headless
+ # or nested-through-a-separate-process agent. But it CANNOT catch the mirror
+ # bug: an Agent-tool subagent / bridge session runs INSIDE the parent claude
+ # process, so its ancestry is byte-identical to the parent's — same
+ # CLAUDE_PID, same ttys — and current_owner happily resolves the PARENT's
+ # tty (proven live TK-11791). Painting then corrupts Steve's live pane. The
+ # only bash-level signal that a run is a child/bridge context is
+ # CLAUDE_CODE_CHILD_SESSION, which a normally-launched top-level tab leaves
+ # UNSET (memory colordots-sweeps-refuse-bridge-sessions). So the rail is:
+ # refuse an own-session paint whenever that flag is set.
+ #
+ # boot=True bypasses the rail: a SessionStart/UserPromptSubmit/Stop HOOK only
+ # ever fires for a real pane-owning session, NEVER for a subagent, so a
+ # hook-invoked paint is always a legitimate own-pane write even when the flag
+ # is set (resumed / nested / bridge sessions). force=True (or
+ # CLAUDE_COLORDOTS_FORCE=1) is the conscious human opt-in for a bridge session
+ # KNOWN to own its pane. Neither loosens the headless/nested refusal above.
+ owner = current_owner(rows, refetch=refetch)
+ if not boot and not force and not paint_forced() and os.environ.get("CLAUDE_CODE_CHILD_SESSION"):
+ raise StatusError(
+ "subagent/bridge context; refusing to paint (it would write to the "
+ "parent session's live pane). Use --tty for an external verified "
+ "target, --boot from a session hook, or --force if this session owns "
+ "its pane.")
+ return owner
+
+
+def selftest():
+ # NEGATIVE TEST (TK-11791, CLAUDE.md "ships with a test that goes red on an
+ # injected fault"). Proves the canonical guard PAINTS a real top-level
+ # interactive session and REFUSES a true subagent — the two contexts are
+ # byte-identical in ancestry (same tty resolved), distinguished ONLY by the
+ # CLAUDE_CODE_CHILD_SESSION flag. Guarded behind the `selftest` subcommand so
+ # no plist ever runs it. Exit 0 = all cases pass; exit 1 = a case regressed.
+ pid = os.getpid()
+ owner_proc = Process(4242, 1, "ttys099", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
+ me = Process(pid, 4242, "??", "Wed Sep 9 08:35:08 2026", "python3")
+ rows = {4242: owner_proc, pid: me} # my ancestry: python3 -> claude(ttys099 owner)
+ refetch = lambda **_: rows
+ want = owner_proc.owner()
+ checks = []
+
+ def case(name, ok):
+ checks.append((name, ok))
+ print((" PASS " if ok else " FAIL ") + name)
+
+ # 1. Top-level (flag UNSET) → PAINTS the resolved owner.
+ env = {k: v for k, v in os.environ.items() if k != "CLAUDE_CODE_CHILD_SESSION"
+ and k != "CLAUDE_COLORDOTS_FORCE"}
+ with patch_environ(env):
+ try:
+ got = owner_for_paint(rows, refetch=refetch)
+ case("top-level interactive session PAINTS", got == want)
+ except StatusError as exc:
+ case("top-level interactive session PAINTS (got refusal: %s)" % exc, False)
+
+ # 2. Subagent/bridge (flag SET) → REFUSES, even though ancestry resolves the parent.
+ with patch_environ(dict(env, CLAUDE_CODE_CHILD_SESSION="1")):
+ try:
+ got = owner_for_paint(rows, refetch=refetch)
+ case("subagent REFUSES (painted %s instead!)" % (got.tty,), False)
+ except StatusError:
+ case("subagent REFUSES", True)
+
+ # 3. --boot bypass (session hook) PAINTS even with the flag set.
+ try:
+ got = owner_for_paint(rows, boot=True, refetch=refetch)
+ case("session-hook --boot PAINTS", got == want)
+ except StatusError as exc:
+ case("session-hook --boot PAINTS (got refusal: %s)" % exc, False)
+
+ # 4. --force bypass (known-pane bridge opt-in) PAINTS with the flag set.
+ try:
+ got = owner_for_paint(rows, force=True, refetch=refetch)
+ case("--force opt-in PAINTS", got == want)
+ except StatusError as exc:
+ case("--force opt-in PAINTS (got refusal: %s)" % exc, False)
+
+ ok = all(v for _, v in checks)
+ print(("selftest: PASS (%d/%d)" if ok else "selftest: FAIL (%d/%d)")
+ % (sum(v for _, v in checks), len(checks)))
+ return 0 if ok else 1
+
+
+class patch_environ:
+ def __init__(self, env):
+ self.env = env
+ def __enter__(self):
+ self.saved = dict(os.environ)
+ os.environ.clear()
+ os.environ.update(self.env)
+ def __exit__(self, *a):
+ os.environ.clear()
+ os.environ.update(self.saved)
+
+
def valid_label(label):
return (isinstance(label, str) and len(label.encode("utf-8")) <= 512
and not any(ord(c) < 32 or ord(c) == 127 for c in label))
@@ -455,7 +588,7 @@ def atomic_write(path, text):
def osc_payload(record):
if record["state"] == "none":
return b"\x1b]6;1;bg;*;default\x07\x1b]1;\x07\x1b]2;\x07\x1b]1337;SetBadgeFormat=\x07"
- title = record["title"]
+ title = display_title(record)
rgb = (0, 150, 136) if record.get("variant") == "monitoring" else COLORS[record["state"]][1]
text = "".join(
f"\x1b]6;1;bg;{channel};brightness;{value}\x07"
@@ -771,7 +904,7 @@ class Store:
result.update(color="none", label="⚪ " + (ticket["id"] or "TK REQUIRED") + " · " + labels.get(reason, "Status not set"))
return result
label, _ = label_ticket(r["label"])
- result.update(color=r["state"], label=status_title(r["state"], label, ticket["id"] or "TK REQUIRED", r.get("variant", "")) or "⚪ " + (ticket["id"] or "TK REQUIRED") + " · Status cleared",
+ result.update(color=r["state"], label=status_title(r["state"], label, ticket["id"] or "TK REQUIRED", r.get("variant", ""), pulse=r.get("revision", 0)) or "⚪ " + (ticket["id"] or "TK REQUIRED") + " · Status cleared",
variant=r.get("variant", ""),
updated_at=r.get("updated_at"), revision=r.get("revision"))
warnings = []
@@ -875,8 +1008,20 @@ def main(argv=None):
help="target ANOTHER live session (e.g. ttys018) so an external supervisor "
"can mark it. assert_owner() still verifies the owner record against "
"the live process table, so a dead/reassigned tty is refused.")
+ sv.add_argument("--force", action="store_true", help=argparse.SUPPRESS)
+ sv.add_argument("--boot", action="store_true", help=argparse.SUPPRESS)
for name in ("set", "clear", "repaint", "current", "start", "ticket", "status", "headers"):
cmd = sub.add_parser(name)
+ # TK-11791: --force / --boot ride the canonical owner_for_paint rail. --force =
+ # conscious bridge opt-in (CLAUDE_COLORDOTS_FORCE=1 also honoured); --boot = an
+ # own-pane session hook (SessionStart/UserPromptSubmit/Stop). --paintable on
+ # `current` runs the FULL guard (ancestry + child-session rail) so /color can
+ # defer to it: exit 0 + owner json if paintable, exit 1 if refused.
+ cmd.add_argument("--force", action="store_true", help=argparse.SUPPRESS)
+ cmd.add_argument("--boot", action="store_true", help=argparse.SUPPRESS)
+ if name == "current":
+ cmd.add_argument("--paintable", action="store_true",
+ help="apply the paint rail (subagent/bridge refusal); exit 1 if not paintable")
if name == "set":
cmd.add_argument("color", choices=[c for c in COLORS if c != "none"])
cmd.add_argument("label", nargs="?", default="")
@@ -906,7 +1051,10 @@ def main(argv=None):
cmd = sub.add_parser(name)
cmd.add_argument("--json", action="store_true")
cmd.add_argument("--tsv", action="store_true")
+ sub.add_parser("selftest") # TK-11791 negative test (never run by a plist)
args = parser.parse_args(argv)
+ if args.command == "selftest":
+ return selftest()
store = Store()
rows = processes()
store.ticket_evidence, store.known_tickets = tickets.discover(
@@ -915,7 +1063,7 @@ def main(argv=None):
data = scan(store, rows)
if args.tsv:
for r in data:
- print("\t".join((r["tty"], r["color"], "true", r["pid"], r["label"])))
+ print("\t".join((r["tty"], r["color"], "true", r["pid"], r["label"], r.get("variant", ""))))
elif args.json:
print(json.dumps(data, ensure_ascii=False))
else:
@@ -945,11 +1093,24 @@ def main(argv=None):
# a caller at all, so resolving one must not be a precondition. Before this, every
# scheduled run failed on that raise while launchctl still reported `last exit code = 0`
# — a textbook false green: loaded, exit 0, runs=1, accomplishing nothing. (2026-09-11)
+ # TK-11791: own-session MUTATING commands (set/clear/ticket/set-variant with no
+ # --tty) resolve through the canonical owner_for_paint rail, so a subagent/bridge
+ # cannot paint the parent's live pane via any dot skill (the mirror bug). --tty
+ # (external supervisor, e.g. greendot-agent) keeps its verified-owner path
+ # unchanged; repaint (Stop hook) + start (color.sh, which passes --boot) stay on
+ # plain ancestry; reads (current/status/headers) never refuse — except
+ # `current --paintable`, the guard /color defers to.
+ force = getattr(args, "force", False)
+ boot = getattr(args, "boot", False)
+ guarded = args.command in ("set", "clear", "ticket", "set-variant") or (
+ args.command == "current" and getattr(args, "paintable", False))
if getattr(args, "tty", ""):
try:
caller = current_owner(rows)
except StatusError:
caller = None
+ elif guarded:
+ caller = owner_for_paint(rows, force=force, boot=boot)
else:
caller = current_owner(rows)
if args.command == "set-variant":
diff --git a/test_terminal_status.py b/test_terminal_status.py
index 4b86bc2..faaf347 100644
--- a/test_terminal_status.py
+++ b/test_terminal_status.py
@@ -482,6 +482,30 @@ class StatusTests(unittest.TestCase):
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.
+ 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")}
+ 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_paint_failure_is_reported_and_auditable(self):
def fail(*args):
raise OSError("fixture: unavailable terminal")
@@ -496,9 +520,43 @@ class StatusTests(unittest.TestCase):
r = self.store.set(self.owner, "yellow", "Choose direction")
packet = ts.osc_payload(r)
self.assertIn(b"green;brightness;204", packet)
- self.assertEqual(packet.count(r["title"].encode()), 2)
+ # 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
← e0a025a terminal-status: keep lightblue label+ticket consistent (TK-
·
back to Terminal Status
·
TK-11794: green 'waiting' variant — brighter static green (0 ea41eb3 →