← back to Terminal Status
verification/TK-11317-integration/terminal_status.py.preimage
1912 lines
#!/usr/bin/env python3
"""One semantic status source for terminal painters and the desktop bar."""
from __future__ import annotations
import argparse
import base64
import contextlib
import dataclasses
import datetime as dt
import fcntl
import json
import os
from pathlib import Path
import re
import stat
import subprocess
import sys
import tempfile
import time
import uuid
import ticket_binding as tickets
# 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
# routinely exceeded, causing every dot paint to raise StatusError and exit 1 --
# which callers see as a failure, not a paint. Made configurable so deployments
# with more processes can raise it without a code change. Default raised to 90s
# (60 was the previous "raised" value; 90 gives headroom for the measured
# worst-case of ~70s at 42 live sessions + 2k procs). The cross-process disk
# cache (TK-11398, committed ce04960) reduces HOW OFTEN a fresh scan fires, so
# the ceiling is hit far less frequently -- but when it IS needed the window
# must be wide enough to actually finish.
#
# TK-11511: DOCUMENTED ROOT-CAUSE FINDING. Mac2 host saturation (load avg
# spiking) is driven by per-session MCP fan-out: ~70 live Claude sessions each
# launch ~19 MCP servers = ~1330 long-running processes beyond the normal
# userland count. terminal-status itself launches ZERO MCP servers and has no
# MCP calls anywhere in this file or ticket_binding.py. The load guard that
# mitigates TK-11511 from terminal-status's side is the disk cache (TK-11398):
# instead of 70 concurrent ps -ax scans each racing to complete within the
# ceiling, sibling invocations within _DISK_TTL seconds share one result. Under
# load that reduces ~70 concurrent scans to ~1 scan per _DISK_TTL window. A
# load-average-based skip gate was considered and rejected: skipping a scan when
# load is high would produce false "no owning terminal" negatives for the
# negative-rescan valve (current_owner path), which is worse than a slow scan.
_PS_TIMEOUT = float(os.environ.get("TERMINAL_STATUS_PS_TIMEOUT", "90"))
COLORS = {
"green": ("🟢", (0, 200, 83), "WORKING"),
"yellow": ("🟡", (255, 204, 0), "DIRECTION?"),
"purple": ("🟣", (148, 0, 211), "GATED"),
"orange": ("🟠", (255, 140, 0), "PASTE waiting"),
"pink": ("🩷", (255, 105, 180), "PARKED"),
# lightblue (Steve, 2026-09-11): the UMBRELLA "any stop that requires my input on any
# terminal". Deliberately overlaps yellow/DIRECTION?, purple/GATED and orange/PASTE —
# those keep their specific meanings; lightblue is the one colour to SCAN for when you
# only want to know "has this stopped and does it need me?". Ranks ABOVE the three in
# every consumer's priority order, because it is the explicit needs-Steve declaration.
"lightblue": ("🔵", (0, 176, 240), "NEEDS STEVE"),
"none": ("", (0, 0, 0), ""),
}
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]+")
def _record_enum_blind(reason):
"""Record that iTerm session enumeration went blind (TK-11369).
A timeout here used to return ({}, "unavailable") silently, which makes
`start` paint green instead of restoring a sticky semantic dot. Never
raises and never blocks: a failure to log must not break painting.
"""
try:
d = os.path.join(os.path.expanduser("~"),
".claude/skills/terminal-status-health/data")
os.makedirs(d, exist_ok=True)
line = json.dumps({
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"pid": os.getpid(),
"rc": "enum_blind",
"err": str(reason)[:300],
}) + "\n"
path = os.path.join(d, "repaint-failures.jsonl")
try:
if os.path.getsize(path) > 262144:
os.replace(path, path + ".1")
except OSError:
pass
# Single O_APPEND write keeps the record atomic across concurrent writers.
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
try:
os.write(fd, line.encode())
finally:
os.close(fd)
except Exception:
pass
VERSION = 1
def issued_clock():
# Python <3.10 on macOS made monotonic_ns() process-relative. Read the
# underlying system-wide raw clock so queued events compare across runtimes.
return time.clock_gettime_ns(getattr(time, "CLOCK_MONOTONIC_RAW", time.CLOCK_MONOTONIC))
# STOPPED_MARK must NOT be the same codepoint as any BASE colour dot (fixed 2026-09-13).
# It was "\U0001F535", which is byte-identical to the lightblue base colour on line 59.
# status_title() renders state + variant into ONE string, so "state=lightblue" and
# "state=purple, variant=stopped" came out visually IDENTICAL -- no string-level reader,
# and no human eye on the tab bar, could tell "this tab's whole purpose is needs-Steve"
# from "this gated tab has gone idle". Two already-distinct canonical fields were being
# collapsed onto one glyph, and every downstream consumer disagreement traced back here.
# \U0001F6D1 (stop sign) differs in BOTH shape (octagon vs circle) and hue (red is unused
# in the palette), so the second axis now rides a second visual channel instead of
# overloading colour. Nothing parses this mark -- it has exactly one render site below --
# so changing it cannot break a reader; consumers that classify read the canonical
# `state`/`variant` fields, and the ones that still grep a glyph only ever OVER-report
# during the repaint window, never under-report.
STOPPED_MARK = "\U0001F6D1" # 🛑 — "stopped, needs Steve", shown NEXT TO the base dot
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)
ticket = match[1].upper() if match else ""
if ticket:
label = label[len(ticket):].lstrip(" ·:-")
return label, ticket
def title_payload(title):
encoded = base64.b64encode(title.encode()).decode()
return (f"\x1b]1;{title}\x07\x1b]2;{title}\x07"
f"\x1b]1337;SetBadgeFormat={encoded}\x07"
f"\x1b]1337;SetUserVar=abramsTitle={encoded}\x07").encode()
class StatusError(Exception):
pass
@dataclasses.dataclass(frozen=True)
class Owner:
tty: str
pid: int
runtime: str
started: str
@property
def epoch(self):
return dt.datetime.strptime(
self.started, "%a %b %d %H:%M:%S %Y").timestamp()
@dataclasses.dataclass
class Process:
pid: int
ppid: int
tty: str
started: str
command: str
@property
def runtime(self):
name = Path(self.command).name
return name if name in ("claude", "codex") else None
def owner(self):
return Owner(self.tty, self.pid, self.runtime, self.started)
# TK-11385 fix 1 of 3: short-TTL cache on the process-table scan.
# Measured on this box: processes() = 14.84s, owners() over the rows = 0.00s.
# A single CLI run calls processes() at least twice -- once to resolve the
# caller (current_owner) and again inside Store.lock() via assert_owner -- so
# every write paid the 14.84s twice. This cache removes the SECOND scan.
# Scope note, deliberately honest: this is a per-PROCESS cache, so it does NOT
# reduce scans across the ~49 concurrent sessions (each still scans once). A
# cross-process file cache would, but a table up to TTL seconds stale could
# fail to see a just-started session and wrongly report "no owning terminal",
# so it is not worth the risk here.
#
# TK-11466-followup (2026-09-11): the 5s default DEFEATED this cache's own
# purpose. The scan is ~15s (up to the 60s ps ceiling under load), so by the
# time the SECOND intra-run call arrives the 5s window has already expired and
# it re-scans anyway -- every paint paid the scan TWICE, which is the ~120s
# hang that let a slow paint leave a stale "two dots" transient on the tab.
# The TTL must exceed one worst-case scan to bridge the two calls WITHIN a run.
# This is a per-PROCESS cache reset each invocation, so a large value only means
# "reuse the one scan across this single run's two calls" -- it has ZERO
# cross-invocation staleness effect (the just-started-session risk above is
# about a cross-PROCESS file cache, which this is not). 120s = 60s ceiling x2.
_PROC_CACHE = {"at": 0.0, "rows": None}
_PROC_TTL = float(os.environ.get("TERMINAL_STATUS_PROC_TTL", "120"))
# TK-11398 (2026-09-11): DOCUMENTED DECISION REVERSAL -- the scope note above
# says a cross-process file cache is "not worth the risk". Reversed here, by
# claude-run-11255, with Steve's explicit approval, on this NEW MEASURED FACT:
# concurrent scans do not cost a CONSTANT ~15s, they mutually slow each other.
# Three consecutive `ps -axo pid,ppid,tty,lstart,comm` runs on this box (1,945
# procs, 42 live claude sessions) measured 2.30s / 4.90s / 7.08s. With ~42
# sessions each shelling out a fresh python3 -- so the per-PROCESS cache above
# is dead on arrival every invocation -- the herd drives a single paint past the
# 60s ps ceiling and NOTHING paints. That is strictly worse than the staleness
# it was avoiding: the observed failure was a tab silently keeping a PREVIOUS
# session's dot (advertising a finished, unrelated ticket) because the paint
# no-oped. A silently wrong dot is the false-green class.
#
# The original objection is preserved exactly, not traded away: a table up to
# TTL seconds stale could miss a just-started session and wrongly report "no
# owning terminal". So a cached table is only ever trusted for a POSITIVE
# answer. Every NEGATIVE ownership conclusion -- current_owner() finding no
# owning runtime, or assert_owner() about to refuse a stale writer -- re-scans
# fresh and re-decides before it is allowed to fail. Worst case a negative
# costs one extra scan; it can never produce a wrong refusal.
# 30s, not 5s: a scan on this box measures 15-70s under load, so a 5s window
# expires before the next invocation can ever reach it -- the same trap the
# per-PROCESS TTL note above describes. Staleness is bounded by the
# negative-rescan valve, not by this number, so the window can afford to be
# wide enough to actually catch the herd.
_DISK_TTL = float(os.environ.get("TERMINAL_STATUS_DISK_TTL", "30"))
def _disk_cache_path():
return Path(tempfile.gettempdir()) / f"terminal-status-proc-{os.getuid()}.json"
def _disk_cache_get():
"""Recent scan from a sibling invocation, or None. Never raises."""
try:
path = _disk_cache_path()
st = path.stat()
if st.st_uid != os.getuid(): # never trust another user's file
return None
if time.time() - st.st_mtime >= _DISK_TTL:
return None
payload = json.loads(path.read_text())
return {int(p[0]): Process(int(p[0]), int(p[1]), p[2], p[3], p[4])
for p in payload}
except (OSError, ValueError, TypeError, KeyError, IndexError):
return None
def _disk_cache_put(rows):
"""Publish this scan for sibling invocations. Best effort, never raises."""
try:
path = _disk_cache_path()
payload = [[p.pid, p.ppid, p.tty, p.started, p.command]
for p in rows.values()]
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".ts-proc-")
try:
with os.fdopen(fd, "w") as handle:
json.dump(payload, handle)
os.chmod(tmp, 0o600)
os.replace(tmp, path) # atomic; readers see whole files
except BaseException:
os.unlink(tmp)
raise
except OSError:
pass
def _splice_self_chain(rows):
"""Make a CACHED table usable for ancestry resolution.
A cached table was written before THIS process existed, so
ancestors(os.getpid(), rows) dead-ends immediately and current_owner()
reports a false negative -- which then forces a full rescan on every call
and cancels the cache's entire benefit. Walk our own chain with targeted
single-pid ps calls (measured 0.01-0.10s each, O(depth ~8) not O(2000))
and splice in whatever the cached table is too old to know about.
Returns None if the chain cannot be resolved completely; the caller then
falls back to a full scan. A partial chain is never returned, so this can
only ever cost an extra scan -- never manufacture a wrong answer.
"""
pid = os.getpid()
for _ in range(32):
if pid <= 1 or pid in rows:
return rows
try:
out = subprocess.run(
["ps", "-o", "ppid=,tty=,lstart=,comm=", "-p", str(pid)],
capture_output=True, text=True, timeout=10,
env={**os.environ, "LC_ALL": "C"})
except (OSError, subprocess.TimeoutExpired):
return None
line = out.stdout.strip()
if out.returncode or not line:
return None
parts = line.split(None, 7)
if len(parts) != 8:
return None
try:
ppid = int(parts[0])
except ValueError:
return None
rows[pid] = Process(pid, ppid, parts[1], " ".join(parts[2:7]), parts[7])
pid = ppid
return None
def processes(fresh=False):
now = time.monotonic()
if not fresh and _PROC_CACHE["rows"] is not None \
and now - _PROC_CACHE["at"] < _PROC_TTL:
return _PROC_CACHE["rows"]
rows = None if fresh else _disk_cache_get()
if rows is not None:
rows = _splice_self_chain(rows)
# Testability seam (TK-11398): a cache whose hit-rate you cannot observe is
# a cache you cannot prove works. Explicit opt-in flag; never set by callers.
if os.environ.get("TERMINAL_STATUS_DEBUG"):
print("proc-cache: %s%s" % ("HIT" if rows is not None else "MISS -> scan",
" (fresh=True forced)" if fresh else ""),
file=sys.stderr)
if rows is None:
rows = _scan_processes()
_disk_cache_put(rows)
_PROC_CACHE["at"] = time.monotonic()
_PROC_CACHE["rows"] = rows
return rows
def _scan_processes():
# `ps -ax` walks the whole process table, so its cost scales with how many
# processes the box is running -- on a workstation with a large MCP fleet
# (~2k procs) it measures ~4-5s idle and ~30s under concurrent load. The old
# 8s ceiling was marginal even at rest, so any parallel work turned it into
# a hard failure: every dot script died with "timed out after 8 seconds"
# and no tab could be painted at all.
# Ownership resolution genuinely needs the full table (owners() proves a
# tty has exactly ONE live runtime before we agree to paint it), so the fix
# is headroom plus an actionable message, not a narrower query.
# TK-11505: timeout is now configurable via TERMINAL_STATUS_PS_TIMEOUT
# (default 90s, set at module load). Exits non-zero on timeout via StatusError.
try:
result = subprocess.run(
["ps", "-axo", "pid=,ppid=,tty=,lstart=,comm="],
capture_output=True, text=True, timeout=_PS_TIMEOUT,
env={**os.environ, "LC_ALL": "C"})
except subprocess.TimeoutExpired:
raise StatusError(
f"Timed out reading the process table (ps -ax took >{_PS_TIMEOUT:.0f}s); "
"the box is overloaded -- check the process count with `ps -A | wc -l`. "
"Set TERMINAL_STATUS_PS_TIMEOUT to a higher value if this box normally "
"has more than ~2000 processes.")
if result.returncode:
raise StatusError("Cannot inspect live terminal owners: " + result.stderr.strip())
rows = {}
for line in result.stdout.splitlines():
parts = line.split(None, 8)
if len(parts) == 9:
p = Process(int(parts[0]), int(parts[1]), parts[2],
" ".join(parts[3:8]), parts[8])
rows[p.pid] = p
return rows
def ancestors(pid, rows):
seen = set()
while pid in rows and pid not in seen:
seen.add(pid)
p = rows[pid]
yield p
pid = p.ppid
def owners(rows):
candidates = {}
for p in rows.values():
if p.runtime and TTY.fullmatch(p.tty):
# Ignore nested agents sharing the parent terminal.
if any(a.runtime for a in ancestors(p.ppid, rows)):
continue
candidates.setdefault(p.tty, []).append(p.owner())
return {tty: group[0] for tty, group in candidates.items() if len(group) == 1}
def current_owner(rows, _retried=False, refetch=processes):
# TK-11672: the fresh re-scan below must re-read the SAME source `rows` came
# from, not silently escape to the live process table. Production leaves
# `refetch` at its default (`processes`) so behaviour is byte-identical; a
# test injects `refetch` so the retry re-reads its fixture instead of the
# live table (which, run inside a real claude session, would resolve a real
# owner and hide the headless-nested-agent refusal this function exists to
# make). The seam is guarded by never being passed in production.
#
# Resolve ownership by ancestry FIRST. CLAUDE_CODE_CHILD_SESSION is set on
# any tool-spawned shell (the Bash tool, hooks), including a legit top-level
# session's own subshell, so it can't be the sole refusal signal — that made
# /dot no-op for the real interactive session. The reliable test is whether
# this process's ancestry resolves to a `claude`/`codex` that is the LIVE
# OWNER of its tty (owners() already excludes nested agents sharing a parent
# terminal); if so, that tty IS ours to paint. This mirrors topic-title.sh.
live = owners(rows)
for p in ancestors(os.getpid(), rows):
if p.runtime:
# Do not walk through a headless nested agent into its parent's tty.
if live.get(p.tty) == p.owner():
return p.owner()
# TK-11398: never let a CACHED table produce a negative. A table up
# to _DISK_TTL old can miss a just-started session; re-scan and
# re-decide before refusing.
if not _retried:
return current_owner(refetch(fresh=True), _retried=True, refetch=refetch)
raise StatusError("Headless or nested agent; refusing parent-terminal write")
# No owning runtime in ancestry at all: only now does the child-session flag
# decide — a true background subagent has no terminal of its own.
if not _retried:
return current_owner(refetch(fresh=True), _retried=True, refetch=refetch)
if os.environ.get("CLAUDE_CODE_CHILD_SESSION"):
raise StatusError("A subagent has no terminal of its own; refusing to paint")
raise StatusError("No owning Claude/Codex terminal in this process ancestry")
def paint_forced():
return os.environ.get("CLAUDE_COLORDOTS_FORCE") == "1"
def _owns_tty_directly(rows, owner):
"""True iff OUR OWN process ancestry resolves `owner` directly, with no
second claude/codex process anywhere between us and it.
current_owner()'s ancestors() walk only ever inspects the CLOSEST
runtime-having ancestor: it returns on that ancestor's first match (a
verified sole live tty owner) or raises (a headless/nested-through-a-
separate-process agent) -- it never walks PAST a non-owning claude/codex
to find a further one. So whenever current_owner() succeeds at all, the
returned owner is, by construction, that single closest ancestor -- this
is therefore always true after a successful current_owner() call today.
Kept as an explicit, defensive re-check (not inferred from "it didn't
raise") so the invariant stays enforced even if current_owner()'s
implementation changes later, per the TK-12167 fix below.
"""
runtime_ancestors = [p for p in ancestors(os.getpid(), rows) if p.runtime]
return len(runtime_ancestors) == 1 and runtime_ancestors[0].owner() == owner
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 a genuine same-PID subagent then
# corrupts Steve's live pane, so CLAUDE_CODE_CHILD_SESSION (set on any
# tool-spawned shell that is NOT the top-level's own direct turn) has to
# stay part of the rail for that class.
#
# TK-12167: that same flag is ALSO carried by a Remote Control (bridge)
# session that genuinely OWNS its own pane — alongside
# CLAUDE_CODE_BRIDGE_SESSION_ID and CLAUDE_CODE_ENTRYPOINT — so the flag
# alone over-refused it too (proven live: pid 33946 on ttys003, `current`
# already resolves owner == {tty: ttys003, pid: 33946}, the session's OWN
# process, yet every dot-script paint was refused for an hour). Narrow the
# refusal from "the flag is set" to "the flag is set AND we are not a
# verified direct-owning bridge": a bridge session is identified by
# CLAUDE_CODE_BRIDGE_SESSION_ID (set only for a genuine Remote Control
# connection, never for a plain internal Agent-tool subagent call), and
# "direct-owning" is the _owns_tty_directly() ownership check above —
# current_owner() having resolved OUR OWN ancestry straight to its live
# sole tty owner, with no other claude/codex process sitting in between.
# A true subagent or a genuinely nested/headless child claude process
# fails one or both of those: it carries no CLAUDE_CODE_BRIDGE_SESSION_ID
# of its own (same-PID mirror-bug case — TK-11791's own selftest fixture),
# or current_owner() has already raised on a real intermediate process
# before this guard is even reached (the nested-through-a-separate-process
# case) — so it is still refused either way. Ownership that cannot be
# determined at all (current_owner() raising) is always refused, never
# rescued by the bridge marker — fail CLOSED.
#
# 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"):
is_verified_bridge = (bool(os.environ.get("CLAUDE_CODE_BRIDGE_SESSION_ID"))
and _owns_tty_directly(rows, owner))
if not is_verified_bridge:
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.
# TK-12167: also scrub CLAUDE_CODE_BRIDGE_SESSION_ID from the baseline so
# this selftest is hermetic when it happens to run INSIDE a real bridge
# session's own shell (it does, routinely) — without this, cases 1-4 below
# would silently inherit a real bridge id from the ambient environment and
# the new bridge-only cases 5-7 would prove nothing.
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_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, NO bridge id) → REFUSES, even though ancestry
# resolves the parent. This fixture is deliberately identical in ps
# ancestry to case 1 -- it models TK-11791's real same-PID mirror bug (an
# Agent-tool subagent carries CLAUDE_CODE_CHILD_SESSION but never its own
# CLAUDE_CODE_BRIDGE_SESSION_ID), so it must keep failing after TK-12167.
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)
# TK-12167 (a): CHILD_SESSION set + a genuine bridge id + the caller owns the
# tty DIRECTLY (same rows as cases 1/2 -- pid 33946 on ttys003 live evidence)
# → PAINTS. This is the exact case that regressed before the fix above: with
# only the guard's ORIGINAL unconditional flag check, this assertion fails.
with patch_environ(dict(env, CLAUDE_CODE_CHILD_SESSION="1",
CLAUDE_CODE_BRIDGE_SESSION_ID="session_test")):
try:
got = owner_for_paint(rows, refetch=refetch)
case("bridge session owning its tty directly PAINTS", got == want)
except StatusError as exc:
case("bridge session owning its tty directly PAINTS (got refusal: %s)" % exc, False)
# TK-12167 (b): CHILD_SESSION set + a genuine bridge id, but a real
# INTERMEDIATE claude process (its own separate pid, detached tty) sits
# between the caller and the tty owner → still REFUSED. The bridge marker
# never rescues a genuinely nested/headless child process.
b_owner = Process(4242, 1, "ttys098", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
b_intermediate = Process(9000, 4242, "??", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
b_me = Process(pid, 9000, "??", "Wed Sep 9 08:35:08 2026", "python3")
b_rows = {4242: b_owner, 9000: b_intermediate, pid: b_me}
b_refetch = lambda **_: b_rows
with patch_environ(dict(env, CLAUDE_CODE_CHILD_SESSION="1",
CLAUDE_CODE_BRIDGE_SESSION_ID="session_test")):
try:
got = owner_for_paint(b_rows, refetch=b_refetch)
case("intermediate claude process REFUSES (painted %s instead!)" % (got.tty,), False)
except StatusError:
case("intermediate claude process REFUSES", True)
# TK-12167 (c): CHILD_SESSION set + a genuine bridge id, but ownership is
# UNDETERMINABLE (no claude/codex anywhere in ancestry at all) → REFUSED.
# Fails CLOSED: the bridge marker alone is never sufficient.
c_me = Process(pid, 1, "??", "Wed Sep 9 08:35:08 2026", "python3")
c_rows = {pid: c_me}
c_refetch = lambda **_: c_rows
with patch_environ(dict(env, CLAUDE_CODE_CHILD_SESSION="1",
CLAUDE_CODE_BRIDGE_SESSION_ID="session_test")):
try:
got = owner_for_paint(c_rows, refetch=c_refetch)
case("undeterminable ownership REFUSES (painted %s instead!)" % (got.tty,), False)
except StatusError:
case("undeterminable ownership REFUSES", True)
# 5. TK-11870 backfill: (a) repaints an owner WITH a valid record, (b) NEVER
# flattens that valid gated dot to green (the key safety invariant — the
# injected fault is a purple/gated tab that must survive a sweep), and (c)
# green-floors a recordless/owner_changed live owner so no live tab stays dark.
# In-memory painter so it runs headless (no real tty); its own tempdir Store.
painted_ttys = []
have = Process(5252, 1, "ttys091", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
lack = Process(5353, 1, "ttys092", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
bf_rows = {5252: have, 5353: lack}
with tempfile.TemporaryDirectory() as bf_dir:
st = Store(user_root=bf_dir, process_provider=lambda **_: bf_rows,
painter=lambda owner, record: painted_ttys.append(owner.tty))
st.set(have.owner(), "purple", "gated reason", rows=bf_rows) # ttys091 = valid purple record
painted_ttys.clear() # ignore set()'s own paint
p, f, s = st.backfill(bf_rows)
purple_rec, _ = st.load(have.owner())
green_rec, _ = st.load(lack.owner())
case("backfill repaints the owner WITH a record", "ttys091" in painted_ttys and p == 1)
case("backfill NEVER flattens a valid gated dot to green",
purple_rec is not None and purple_rec["state"] == "purple")
case("backfill green-floors the recordless live owner",
f == 1 and green_rec is not None and green_rec["state"] == "green")
# 6. TK-11870 heartbeat honesty (CLAUDE.md TK-11431: an unmeasured input is
# NEVER green; severity maps to capability). The injected faults are the
# two false-green shapes the old hardcoded-PASS heartbeat masked.
case("heartbeat WARN + NOT-MEASURED when the ps scan failed",
backfill_heartbeat(0, 0, 0, None, scan_error="ps -ax timed out")["verdict"] == "WARN"
and backfill_heartbeat(0, 0, 0, None, scan_error="x")["live_owners"] is None)
case("heartbeat WARN when live tabs exist but NONE could be asserted",
backfill_heartbeat(0, 0, 3, 3)["verdict"] == "WARN")
case("heartbeat PASS when it acted on the live tabs",
backfill_heartbeat(30, 0, 3, 33)["verdict"] == "PASS")
case("heartbeat PASS on a measured 0-of-0 (no live tabs), still carrying the count",
backfill_heartbeat(0, 0, 0, 0)["verdict"] == "PASS"
and backfill_heartbeat(0, 0, 0, 0)["live_owners"] == 0)
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))
def color_of(title):
# A title mentions other colors legitimately; only the first dot is status.
for color, (dot, _, _) in COLORS.items():
if dot and title.startswith(dot):
return color
return "none"
def atomic_write(path, text):
path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
fd, tmp = tempfile.mkstemp(prefix="." + path.name + ".", dir=path.parent)
try:
with os.fdopen(fd, "w") as stream:
stream.write(text)
stream.flush()
os.fsync(stream.fileno())
os.replace(tmp, path)
finally:
if os.path.exists(tmp):
os.unlink(tmp)
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 = display_title(record)
# Green VARIANT tints (both green-only, guarded at set-time): waiting = a distinct
# brighter STATIC green for an idle/waiting green session; monitoring = teal, a
# scheduled deadline. Any other variant/base uses the base colour's rgb (stopped
# keeps its base dot and rides the additive marker instead of changing the tint).
variant = record.get("variant")
if variant == "waiting":
rgb = (0, 255, 120)
elif variant == "monitoring":
rgb = (0, 150, 136)
else:
rgb = COLORS[record["state"]][1]
text = "".join(
f"\x1b]6;1;bg;{channel};brightness;{value}\x07"
for channel, value in zip(("red", "green", "blue"), rgb))
return text.encode() + title_payload(title)
def write_terminal(owner, payload):
fd = os.open("/dev/" + owner.tty, os.O_WRONLY | os.O_NOCTTY | os.O_NONBLOCK)
try:
if not stat.S_ISCHR(os.fstat(fd).st_mode) or not os.isatty(fd):
raise StatusError("Paint target is not a terminal")
# One write, no background loop, no terminal input injection.
# Non-blocking write: a momentarily busy tty (input buffer full, reader not
# draining) makes os.write raise BlockingIOError (EAGAIN / Errno 35) or return
# a PARTIAL count. Retry the remaining bytes with a short bounded backoff rather
# than turning a transient full buffer into a hard paint failure — an unretried
# EAGAIN latched a 24h FAIL on the fleet-health panel (~1.4% of paints hit it).
# Keep O_NONBLOCK so a dead/gone reader can never hang us; bound the wait instead.
# On genuine wedge (>0.5s) raise "Terminal busy; retry" — the same transient class
# dot-floor.sh already treats as benign, not a false engine fault.
view = memoryview(payload)
sent = 0
deadline = time.monotonic() + 0.5
while sent < len(view):
try:
sent += os.write(fd, view[sent:])
except BlockingIOError:
if time.monotonic() >= deadline:
raise StatusError("Terminal busy; retry")
time.sleep(0.01)
finally:
os.close(fd)
def paint(owner, record):
write_terminal(owner, osc_payload(record))
class Store:
def __init__(self, user_root=None, process_provider=processes, painter=paint):
self.user_root = Path(user_root) if user_root else Path.home()
self.root = self.user_root / ".local/state/abrams-terminal-status"
self.legacy = {runtime: self.user_root / f".{runtime}/tab-dots"
for runtime in ("claude", "codex")}
self.process_provider = process_provider
self.painter = painter
self.ticket_evidence = {}
self.known_tickets = None
def ticket_info(self, owner, record=None):
record = record or {}
saved = {"id": record.get("ticket", ""), "at": record.get("ticket_at", 0),
"source": record.get("ticket_source", "unbound")}
found = self.ticket_evidence.get(owner.tty, {})
if found.get("at", 0) > saved["at"]:
return found
# A migration record was already freshness/identity checked by read().
if not saved["at"] and not saved["id"] and "version" not in record:
_, legacy_ticket = label_ticket(record.get("label", ""))
if legacy_ticket and (self.known_tickets is None or legacy_ticket in self.known_tickets):
return {"id": legacy_ticket, "at": owner.epoch, "source": "fresh_legacy_label"}
return saved
def path(self, owner):
if not TTY.fullmatch(owner.tty):
raise StatusError("Invalid tty")
return self.root / (owner.tty + ".json")
def assert_owner(self, owner, rows=None):
# TK-11385 fix 2 of 3: accept an already-resolved process table.
# This runs INSIDE Store.lock(), so re-scanning here held the per-tty
# lock for the full duration of the scan. The caller has almost always
# just resolved the table; reusing it keeps the guard while removing
# the scan from the critical section.
rows = rows if rows is not None else self.process_provider()
if owners(rows).get(owner.tty) != owner:
# TK-11398: a cached table may simply be stale. Re-scan fresh and
# re-check before refusing a writer that is in fact still the owner.
rows = processes(fresh=True)
if owners(rows).get(owner.tty) != owner:
raise StatusError("Terminal owner changed; refusing stale writer")
@contextlib.contextmanager
def lock(self, owner, rows=None, wait=None):
self.path(owner) # Validate before building any filesystem path.
directory = self.root / ".locks"
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
with open(directory / (owner.tty + ".lock"), "a") as lock:
# TK-11385 fix 3 of 3: was 3s, which a same-tty writer could
# NEVER win -- the holder sat in a 14.84s scan while the waiter gave
# up after 3s, so contention was a guaranteed loss, not a race.
# With fixes 1+2 the lock is now held for ~0s, so this longer
# deadline is a backstop that should essentially never be reached;
# it is NOT licence to hold the lock across expensive work.
#
# TK-11835: a `wait` OVERRIDE lets a CROSS-TTY caller (the `--tty`
# external supervisor / dot sweep) fail FAST instead of inheriting
# this 65s SELF-writer backstop. When a target session is mid-turn it
# fires its own working-state hooks (each grabbing THIS lock), and a
# holder's assert_owner fresh `ps -ax` rescan runs inside the lock --
# under concurrent sweep load that ps hits ~30s (see _scan_processes),
# so without an override a sweeper hung up to 65s per busy tab. A short
# override turns that into a reported skip. `wait=None` preserves the
# exact prior behaviour for a session painting its OWN tab.
budget = wait if wait is not None else float(
os.environ.get("TERMINAL_STATUS_LOCK_WAIT", "65"))
deadline = time.monotonic() + budget
while True:
try:
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
break
except BlockingIOError:
if time.monotonic() >= deadline:
raise StatusError(
"Terminal %s is busy (owner mid-write); skipped after "
"%.1fs -- retry when idle." % (owner.tty, budget))
time.sleep(0.02)
try:
self.assert_owner(owner, rows)
yield
finally:
fcntl.flock(lock, fcntl.LOCK_UN)
def load(self, owner):
path = self.path(owner)
if not path.exists():
return None, "missing"
try:
r = json.loads(path.read_text())
if (r["version"] != VERSION or r["state"] not in COLORS
or not valid_label(r["label"]) or not valid_label(r["title"])
or not isinstance(r["revision"], int)
or not isinstance(r["issued_ns"], int)):
return None, "invalid"
if r["owner"] != dataclasses.asdict(owner):
return None, "owner_changed"
ticket = r.get("ticket", "")
if (ticket and not re.fullmatch(r"TK-\d+", ticket)) or not isinstance(r.get("ticket_at", 0), (int, float)):
return None, "invalid"
if r.get("variant", "") not in ("", "monitoring", "stopped", "waiting"):
return None, "invalid"
expected = status_title(r["state"], r["label"], ticket, r.get("variant", ""))
if r["title"] != expected or (r["state"] == "none" and r["label"]):
return None, "invalid"
return r, "canonical"
except (OSError, ValueError, KeyError, TypeError):
return None, "invalid"
def legacy_status(self, owner, live_title=""):
# Only the owning runtime's legacy file can be considered during rollout.
path = self.legacy[owner.runtime] / (owner.tty + ".dot")
try:
title = path.read_text().strip()
if path.stat().st_mtime < owner.epoch or not valid_label(title):
return None, "legacy_stale"
except (OSError, ValueError):
return None, "missing"
color = color_of(title)
if color == "none":
return None, "legacy_empty"
live_color = color_of(live_title)
if live_color not in ("none", color):
return None, "legacy_conflict"
return {"state": color, "title": title,
"label": title[len(COLORS[color][0]):].strip()}, "legacy"
def read(self, owner, live_title=""):
r, reason = self.load(owner)
if reason == "missing":
return self.legacy_status(owner, live_title)
# A clear tombstone, corruption, or reused tty can never fall back.
return r, reason
def set(self, owner, color, label="", *, issued_ns=None, ticket_update=None,
variant="", deferential=False, rows=None, lock_wait=None):
issued_ns = issued_ns if issued_ns is not None else issued_clock()
if color not in COLORS or not valid_label(label) or variant not in ("", "monitoring", "stopped", "waiting"):
raise StatusError("Invalid status or label")
label, embedded_ticket = label_ticket(label)
if embedded_ticket:
ticket_update = embedded_ticket
if ticket_update is not None:
if ticket_update and (not re.fullmatch(r"TK-\d+", ticket_update) or
self.known_tickets is not None and ticket_update not in self.known_tickets):
raise StatusError("Ticket does not exist in the canonical ledger")
# An EXPLICIT caller reason (a label OR a ticket) must WIN over the T1-c
# preserve-path below. Captured before the default is applied, so a bare
# `set(lightblue)` (no reason) rides the base while `set(lightblue, "TK-x · why")`
# paints its own reason — otherwise the new ticket lands on the OLD label.
caller_gave_reason = bool(label) or ticket_update is not None
label = "" if color == "none" else (label or COLORS[color][2])
# TK-11835: a cross-tty caller passes the already-resolved `rows` (so the
# in-lock assert_owner reuses main's table instead of re-scanning) and a
# short `lock_wait` (so a busy target is skipped fast, not waited on 65s).
# Both default to None -> byte-identical to the prior self-paint path.
with self.lock(owner, rows=rows, wait=lock_wait):
previous, _ = self.load(owner)
if previous and issued_ns <= previous["issued_ns"]:
raise StatusError("Obsolete status update; a newer event already won")
# T1-c (TK-11779) — an EXPLICIT lightblue must not ERASE an underlying
# needs-Steve reason. lightblue is the UMBRELLA "this stop needs Steve"; when
# a SPECIFIC reason already stands on the tab — purple (gated memo), orange
# (paste waiting), yellow (question) — express "needs Steve" as the additive
# 🔵 stopped marker ON that base colour instead of overwriting it. Clearing the
# marker later (session resumes) then REVEALS the still-pending reason instead
# of a dead/dotless tab that silently lost its gated memo. A green/teal/pink/
# none/lightblue base has no reason to preserve, so it takes solid lightblue as
# before. `variant == "stopped"` guards against re-routing the additive path
# itself (the Stop hook already sets the marker directly).
if (color == "lightblue" and variant != "stopped" and not caller_gave_reason
and previous and previous.get("state") in ("purple", "orange", "yellow")):
color = previous["state"]
variant = "stopped"
label = label_ticket(previous.get("label", ""))[0]
# Defensive coherence (TK-11826 / TK-11779 Finding 2): the branch guard
# `not caller_gave_reason` already guarantees ticket_update is None here
# (caller_gave_reason is True whenever ticket_update is not None), so the
# overwrite below cannot fire on this path today. Pin it to None anyway so
# the preserved reason stays coherent as a WHOLE — colour + label + ticket —
# even if a future edit ever loosens the branch condition; never the
# mismatch of an OLD preserved label carrying a NEW caller ticket.
ticket_update = None
# TK-11378 (DTD verdict A) — an AUTOMATIC paint must never erase a pending
# request for Steve's attention. PRIORITY already encodes the ordering
# (orange > purple > yellow > green > pink > none) but until now was only
# used to sort scan() output, never to arbitrate a paint. "Sticky" = anything
# ranked above green, i.e. the three colors that mean a human is blocked.
# Only deferential callers (the per-prompt working-state hook) yield; every
# EXPLICIT paint (/greendot, /color, /pinkdot, --off) stays authoritative, so
# a session can still clear its own dot or reuse the tab for new work.
if (deferential and previous and color == "green"
and PRIORITY.get(previous.get("state"), PRIORITY["none"]) < PRIORITY["green"]):
self.assert_owner(owner, rows)
try:
self.painter(owner, previous)
except (OSError, StatusError):
pass # keep the record authoritative even if the repaint fails
return previous
ticket = self.ticket_info(owner, previous)
if ticket_update is not None:
ticket = {"id": ticket_update, "at": time.time(), "source": "explicit"}
record = {
"version": VERSION, "owner": dataclasses.asdict(owner),
"state": color, "label": label,
"title": status_title(color, label, ticket["id"],
variant if (color == "green" or variant == "stopped") else ""),
"ticket": ticket["id"], "ticket_at": ticket["at"],
"ticket_source": ticket["source"],
# "stopped" is colour-AGNOSTIC: it is an additive 🔵 marker beside ANY base
# dot, and the blocked colours (purple/yellow/orange) are its whole point.
# "monitoring" stays green-only because it is the teal TINT of green.
"variant": variant if (color == "green" or variant == "stopped") else "",
"revision": previous["revision"] + 1 if previous else 1,
"event_id": uuid.uuid4().hex, "issued_ns": issued_ns,
"updated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
"render_status": "pending", "mirror_errors": [],
}
self.assert_owner(owner, rows)
atomic_write(self.path(owner), json.dumps(record, indent=2) + "\n")
error = None
try:
self.assert_owner(owner, rows)
self.painter(owner, record)
record["render_status"] = "applied"
except (OSError, StatusError) as exc:
record["render_status"] = "failed"
record["render_error"] = str(exc)
error = exc
for directory in self.legacy.values():
try:
atomic_write(directory / (owner.tty + ".dot"), record["title"])
except OSError as exc:
record["mirror_errors"].append(str(exc))
atomic_write(self.path(owner), json.dumps(record, indent=2) + "\n")
if error or record["mirror_errors"]:
raise StatusError("Status saved, but display/mirror update failed; run audit")
return record
def repaint(self, owner, *, rows=None, lock_wait=None):
with self.lock(owner, rows=rows, wait=lock_wait):
record, reason = self.load(owner)
if record is None:
raise StatusError("Cannot repaint unknown status: " + reason)
self.assert_owner(owner, rows)
self.painter(owner, record)
# Never refresh timestamps, rewrite mirrors, or resurrect cleared state.
return record
def backfill(self, rows, *, lock_wait=2.5):
"""Make Steve's HARD rule true: EVERY live terminal shows a dot at all times.
Two failures this repairs (TK-11870):
1. DROPPED DOT — under host overload a paint raises a ps-timeout StatusError
and NO-OPS, so the tab loses its dot while the persisted record still holds
the truth. -> pure `repaint` re-emits the stored OSC (no mutation).
2. RECORDLESS/OWNER-CHANGED LIVE TAB — a live claude/codex session whose only
on-disk record belongs to a PRIOR session on the reused tty (the signature
of a boot paint refused by the child-session rail), so it shows ⚪ "status
not set". owner_changed means a genuinely NEW process (pid/started differ),
so there is no prior state to preserve. -> `set` a GREEN floor (working),
exactly what the `start` command already does for owner_changed. The
session's own hooks refine it on its next turn; a green floor beats a dark
⚪ tab. (`missing` is treated the same — a live owner with no record of its
own.)
Safe by construction: it iterates owners(rows) (the authoritative
process-table -> tty resolution), so it repaints/floors each owner's OWN tty
from its OWN record — it NEVER guesses a tty from a label map (memory
every-terminal-always-shows-a-dot: that mis-map nearly clobbered a GATED tab).
A VALID record is only ever repainted, never overwritten, so a real
gated/parked/needs-Steve dot can never be flattened to green. `invalid`/
conflicting records and mid-turn busy tabs are left for manual refresh / the
next cycle. Runs as an EXTERNAL supervisor (clean launchd env), so it passes
the resolved `rows` (no per-owner rescan) and a SHORT lock_wait so a busy tab
is skipped this cycle, not waited on the 65s self-writer backstop.
Returns (painted, floored, skipped)."""
painted = floored = skipped = 0
for owner in owners(rows).values():
_, reason = self.load(owner)
try:
if reason == "canonical":
self.repaint(owner, rows=rows, lock_wait=lock_wait)
painted += 1
elif reason in ("missing", "owner_changed"):
self.set(owner, "green", rows=rows, lock_wait=lock_wait)
floored += 1
else:
skipped += 1 # invalid / legacy_conflict -> manual refresh
except (StatusError, OSError):
skipped += 1 # busy tab this cycle -> caught next cycle
return painted, floored, skipped
def set_variant(self, owner, variant, *, rows=None, lock_wait=None):
"""Toggle ONLY the variant (e.g. the additive 🔵 stopped marker) on an existing
status — colour, label, ticket and timestamps are preserved. This is what the
flasher pulses, so a crashed loop can never corrupt the real dot: the worst case
is the marker left on or off beside an otherwise-correct base colour."""
if variant not in ("", "monitoring", "stopped", "waiting"):
raise StatusError("Invalid variant")
with self.lock(owner, rows=rows, wait=lock_wait):
record, reason = self.load(owner)
if record is None:
raise StatusError("Cannot set variant on unknown status: " + reason)
self.assert_owner(owner, rows)
color = record["state"]
# monitoring AND waiting are green-only tints; drop either on a non-green base.
if variant in ("monitoring", "waiting") and color != "green":
variant = ""
record["variant"] = variant
record["title"] = status_title(color, record["label"], record.get("ticket", ""), variant)
record["revision"] = record.get("revision", 0) + 1
record["event_id"] = uuid.uuid4().hex
record["issued_ns"] = issued_clock()
record["updated_at"] = dt.datetime.now(dt.timezone.utc).isoformat()
record["mirror_errors"] = []
self.assert_owner(owner)
self.painter(owner, record)
# The legacy .dot mirror is what allcolordots and dot-screen-router actually
# read — updating only the JSON would leave every consumer showing the old
# title while the engine reported success.
for directory in self.legacy.values():
try:
atomic_write(directory / (owner.tty + ".dot"), record["title"])
except OSError as exc:
record["mirror_errors"].append(str(exc))
atomic_write(self.path(owner), json.dumps(record, indent=2) + "\n")
return record
def row(self, owner, title=""):
r, reason = self.read(owner, title)
ticket = self.ticket_info(owner, r)
result = {"tty": owner.tty, "pid": str(owner.pid),
"runtime": owner.runtime, "live": True,
"ticket": ticket["id"], "ticket_source": ticket["source"],
"status_source": reason, "owner": dataclasses.asdict(owner)}
if r is None:
labels = {
"legacy_conflict": "Conflicting old status — refresh dot",
"invalid": "Invalid status record — refresh dot",
"owner_changed": "New session — status not set",
}
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", ""), 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 = []
live_color = color_of(title)
if reason == "canonical":
if live_color != "none" and live_color != r["state"]:
warnings.append("window_title_disagrees")
for runtime, directory in self.legacy.items():
try:
if (directory / (owner.tty + ".dot")).read_text() != r["title"]:
warnings.append(runtime + "_mirror_disagrees")
except OSError:
warnings.append(runtime + "_mirror_missing")
if r.get("render_status") != "applied":
warnings.append("display_update_" + r.get("render_status", "unknown"))
result["warnings"] = warnings
return result
def header(self, owner, writer=write_terminal):
with self.lock(owner):
row = self.row(owner)
minutes = max(0, int((time.time() - owner.epoch) / 60))
title = row["label"] + f" · up {minutes // 60}h {minutes % 60}m"
self.assert_owner(owner)
writer(owner, title_payload(title))
return title
# TK-11879: the iTerm2 window/tab/session enumeration below is served by iTerm's
# SINGLE AppleEvent handler, so when ~49 interactive sessions each shell it at start
# under host load they serialize inside iTerm and the later ones time out or return
# rc 1 (628 'osascript returncode 1' blinds in one 7d window). That is the identical
# herd-contention class the ps process-table scan hit, and the fix is the same proven
# one: a short-TTL cross-process disk cache (TK-11398/TK-11831). The FIRST session in
# a window enumerates and publishes the {tty: title} map; every sibling that starts
# within the TTL reads the file and never adds a second AppleEvent to the busy handler.
# When a live enumeration DOES go blind, a recently cached map is a graceful fallback:
# a real (if slightly stale) session map beats the {} that makes `start` fall through
# to a confidently-wrong green. Only when NO usable cache exists is it a genuine blind.
_ITERM_TTL = float(os.environ.get("TERMINAL_STATUS_ITERM_TTL", "30"))
# How much older than the fresh TTL a cached map may be and still serve as the
# blind-fallback. Bounded so a long-abandoned map is not trusted forever.
_ITERM_STALE_MAX = float(os.environ.get("TERMINAL_STATUS_ITERM_STALE_MAX", "300"))
# TK-11831 single-flight: the 30s TTL cache above collapses REPEAT enumerations,
# but a SIMULTANEOUS cold herd (~59 sessions starting at once, empty cache) all
# miss together and each fires its own osascript into iTerm's ONE serial
# AppleScript queue -- the exact stampede this ticket is about (628 'osascript
# rc1' blinds in one 7d window, terminal_api unavailable x57 at the 91s peak).
# The codex-check flagged that the TTL cache alone moves repeat-frequency, not
# the cold-herd tail. Fix: a NON-BLOCKING cross-process lock so exactly ONE
# caller enumerates; every peer waits briefly for that winner to publish its
# fresh map (the coalesced path) and, only if the winner is still enumerating
# past the wait, falls through to the SAME stale-cache/blind path that already
# exists on an osascript timeout -- so a peer NEVER fires a second osascript and
# NEVER manufactures a fresh answer it did not measure. Fail-OPEN: any lock
# error degrades to today's behaviour (enumerate directly). Kill-switch:
# TERMINAL_STATUS_NO_SINGLEFLIGHT=1. The wait is bounded well under startup
# tolerance so a coalescing peer can never hang a session.
_ITERM_SINGLEFLIGHT = os.environ.get("TERMINAL_STATUS_NO_SINGLEFLIGHT", "") == ""
_ITERM_SF_WAIT = float(os.environ.get("TERMINAL_STATUS_ITERM_SF_WAIT", "6"))
def _iterm_cache_path():
return Path(tempfile.gettempdir()) / f"terminal-status-iterm-{os.getuid()}.json"
def _iterm_lock_path():
return Path(tempfile.gettempdir()) / f"terminal-status-iterm-{os.getuid()}.lock"
def _iterm_singleflight_acquire():
"""Try to become the sole enumerator. Returns an OPEN locked fd (caller must
close it to release) if we won, or None if a peer already holds it. Fail-OPEN:
on any error return a sentinel that behaves like 'won' so painting degrades to
the pre-single-flight direct-enumerate path rather than blocking."""
try:
fd = os.open(str(_iterm_lock_path()), os.O_CREAT | os.O_RDWR, 0o600)
except OSError:
return "failopen" # cannot even open the lockfile -> enumerate directly
try:
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
return fd # we are the single flight
except OSError:
os.close(fd)
return None # a peer is enumerating
def _iterm_wait_for_winner():
"""A peer is enumerating. Poll the cross-process cache up to _ITERM_SF_WAIT for
the fresh map the winner publishes. The caller only reaches here AFTER its own
_iterm_cache_get(_ITERM_TTL) already MISSED, so any within-TTL map that appears
now is necessarily the winner's new write -- a plain TTL-freshness poll is
sufficient (no mtime bookkeeping). Return that map on success, else None (caller
then uses the existing stale/blind fallback). Never fires osascript; never raises."""
deadline = time.monotonic() + _ITERM_SF_WAIT
while time.monotonic() < deadline:
hit = _iterm_cache_get(_ITERM_TTL)
if hit is not None and hit[0]:
return hit[0]
time.sleep(0.15)
return None
def _iterm_cache_get(max_age):
"""Most recent sibling enumeration within max_age seconds, or None. Never raises.
Returns (sessions_dict, age_seconds)."""
try:
path = _iterm_cache_path()
st = path.stat()
if st.st_uid != os.getuid(): # never trust another user's file
return None
age = time.time() - st.st_mtime
if age >= max_age:
return None
payload = json.loads(path.read_text())
sessions = payload.get("sessions")
if not isinstance(sessions, dict):
return None
clean = {k: v for k, v in sessions.items()
if isinstance(k, str) and isinstance(v, str) and TTY.fullmatch(k)}
return clean, age
except (OSError, ValueError, TypeError, KeyError):
return None
def _iterm_cache_put(sessions):
"""Publish this enumeration for sibling invocations. Best effort, never raises.
Only SUCCESSFUL enumerations (rc 0) are ever cached -- a blind never writes here,
so the fallback map is always a map iTerm actually returned."""
try:
path = _iterm_cache_path()
fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=".ts-iterm-")
try:
with os.fdopen(fd, "w") as handle:
json.dump({"sessions": sessions}, handle)
os.chmod(tmp, 0o600)
os.replace(tmp, path) # atomic; readers see whole files
except BaseException:
os.unlink(tmp)
raise
except OSError:
pass
def _iterm_blind(reason):
"""A live enumeration failed. Prefer a recent cached map (graceful degrade to a
REAL session map, not the {} that makes `start` paint a confidently-wrong green,
TK-11879). Only when NO usable cache exists is this a genuine blind: record
enum_blind (an honest NOT-MEASURED signal -- the residual wrong-green risk) and
return unavailable. A recovered blind is deliberately NOT recorded, because with
a real fallback map there is no wrong paint to warn about."""
fallback = _iterm_cache_get(_ITERM_STALE_MAX)
if fallback is not None:
if os.environ.get("TERMINAL_STATUS_DEBUG"):
print("iterm-cache: BLIND (%s) -> stale fallback (%.1fs)"
% (reason, fallback[1]), file=sys.stderr)
return fallback[0], "stale"
_record_enum_blind(reason)
return {}, "unavailable"
def iterm_sessions(fresh=False):
# Serve a fresh sibling enumeration from the cross-process cache without touching
# iTerm at all (herd collapse). `fresh=True` forces a live enumeration (testability
# seam + a way to warm the cache deliberately); callers never set it.
if not fresh:
hit = _iterm_cache_get(_ITERM_TTL)
if hit is not None:
if os.environ.get("TERMINAL_STATUS_DEBUG"):
print("iterm-cache: HIT (%.1fs)" % hit[1], file=sys.stderr)
return hit[0], "cached"
# Cold miss. TK-11831 single-flight: collapse a simultaneous herd to ONE
# osascript. A peer already enumerating -> wait briefly for its fresh map,
# else degrade to the SAME stale/blind path an osascript failure would take
# (never a second osascript, never a manufactured fresh answer).
sf_lock = None
if not fresh and _ITERM_SINGLEFLIGHT:
sf_lock = _iterm_singleflight_acquire()
if sf_lock is None:
coalesced = _iterm_wait_for_winner()
if coalesced:
if os.environ.get("TERMINAL_STATUS_DEBUG"):
print("iterm-cache: COALESCED (peer enumerated)", file=sys.stderr)
return coalesced, "cached"
return _iterm_blind(
"single-flight: peer still enumerating after %.1fs" % _ITERM_SF_WAIT)
# sf_lock is an open fd (we won) or "failopen" (degrade to direct enumerate).
script = """
set sep to ASCII character 9
tell application "iTerm2"
set output to ""
repeat with w in windows
repeat with t in tabs of w
repeat with s in sessions of t
set output to output & (tty of s) & sep & (name of s) & linefeed
end repeat
end repeat
end repeat
return output
end tell
"""
try:
# TK-11369 (DTD 2026-09-10, verdict C): was timeout=8. iTerm AppleScript
# enumeration cost scales with tab count (~49 here) and this box has hit
# load 53.78, so an 8s ceiling was trippable. Raised to 30s -- NOT 60s
# like the ps call, because this runs at session start and a longer hang
# would delay startup. A blind here is a genuine blind spot, not a benign
# empty result, so it degrades to a cached map (TK-11879) or, failing that,
# is recorded rather than silently swallowed.
try:
result = subprocess.run(["osascript", "-e", script], capture_output=True,
text=True, timeout=30)
finally:
# Release the single-flight lock the instant enumeration returns so a
# waiting peer can proceed; the cache write below is cheap and racey-safe.
if isinstance(sf_lock, int):
os.close(sf_lock)
sf_lock = None
if result.returncode:
return _iterm_blind("osascript returncode " + str(result.returncode))
except (OSError, subprocess.TimeoutExpired) as exc:
return _iterm_blind(type(exc).__name__ + ": " + str(exc)[:120])
finally:
if isinstance(sf_lock, int): # any early return / exception path
os.close(sf_lock)
sessions = {}
for line in result.stdout.splitlines():
tty, sep, title = line.partition("\t")
tty = tty.removeprefix("/dev/")
if sep and TTY.fullmatch(tty):
sessions[tty] = title
# Only cache a NON-EMPTY map. An rc-0 enumeration that returns zero sessions is
# never legitimate here -- the asking Claude session is itself a session, so an
# empty result is a partial/racey answer (iTerm mid-launch). Caching it would
# serve siblings a `cached {}` for a whole TTL -> the same green fall-through this
# fix exists to kill, and silently (no enum_blind). Leave the cache untouched so
# the next sibling re-enumerates; this call still returns its own honest result.
if sessions:
_iterm_cache_put(sessions)
if os.environ.get("TERMINAL_STATUS_DEBUG"):
print("iterm-cache: MISS -> enumerated %d sessions%s"
% (len(sessions), " (empty, not cached)" if not sessions else ""),
file=sys.stderr)
return sessions, "available"
def scan(store, rows=None, sessions=None):
rows = rows if rows is not None else store.process_provider()
if sessions is None:
sessions, ui_status = iterm_sessions()
else:
ui_status = "fixture"
result = []
for tty, owner in owners(rows).items():
row = store.row(owner, sessions.get(tty, ""))
row["terminal_visible"] = tty in sessions
row["terminal_api"] = ui_status
result.append(row)
# A "stopped" (needs-Steve) tab keeps its specific base colour (TK-11779) but must not
# lose the top-of-list rank the umbrella lightblue used to give it: PRIORITY keys only
# on colour, so a purple+stopped tab would otherwise sort level with a plain gated tab.
# Elevate the stopped group ahead of everything else, then order within it by base
# colour, so a preserved-reason stop still ranks above a same-colour non-stopped tab.
return sorted(result, key=lambda r: (0 if r.get("variant") == "stopped" else 1,
PRIORITY.get(r["color"], len(PRIORITY)), r["tty"]))
def backfill_heartbeat(painted, floored, skipped, live_owners, *, scan_error=None, ts=None):
"""Honest heartbeat verdict for the backfill supervisor (TK-11870).
The verdict is DERIVED from what the cycle actually measured, never hardcoded
PASS -- severity maps to CAPABILITY, not counts (CLAUDE.md TK-11431):
* scan_error set -> WARN + NOT-MEASURED. The `ps` scan (the very input this
supervisor exists to keep warm) failed, so it could not act at ALL this
cycle. `live_owners` is null -- an unmeasured input is NEVER green. The
job still FIRED (fresh mtime), so the who-watches-the-watcher liveness
passes; the FAILURE rides in the verdict, not in the file's absence (the
old path raised before writing, so a ps-timeout produced NO heartbeat and
a byte-identical-to-idle silence).
* live_owners > 0 but painted+floored == 0 (every live tab skipped) -> WARN.
Live tabs were measured but NONE could be asserted this pass -- "could not
act this pass" = WARN (keep-alive doctrine). A live tab may be dark.
* otherwise -> PASS. It ran and could act; and live_owners is always carried
so a measured 0-of-0 (no live tabs) stays distinguishable from 0-of-N (the
false-green rule-1 example) rather than both reading green off a bare count.
"""
if scan_error:
verdict = "WARN"
elif live_owners and painted + floored == 0:
verdict = "WARN"
else:
verdict = "PASS"
hb = {
"ts": ts or dt.datetime.now(dt.timezone.utc).isoformat(),
"verdict": verdict, "status": verdict,
"painted": painted, "floored": floored, "skipped": skipped,
"live_owners": live_owners,
}
if scan_error:
hb["error"] = scan_error
return hb
# --- TK-12168: auto-bind orchestration --------------------------------------
# ticket_binding.discover() is deliberately READ-ONLY ("never create a second
# ticket database" -- see its module docstring), so a plain top-level
# interactive session that never ran `tk` and carries no TK- in its argv sits
# unbound forever ("TK REQUIRED") no matter how much real work it does. This
# closes that gap from the UserPromptSubmit hook (working-state.sh, backgrounded
# so a slow/failed `tk` call never delays the prompt): if the session is still
# unbound after a SUBSTANTIVE prompt, bind an explicit TK- the prompt names, or
# mint one via `tk new` and bind that. Every decision point is injectable
# (tk_new / ticket_set) so the whole flow is testable without a live ledger,
# a real Store, or a subprocess.
_AUTOBIND_RATE_LIMIT = float(os.environ.get("TERMINAL_STATUS_AUTOBIND_RATE_LIMIT", "20"))
def _autobind_state_path(store, owner):
d = store.root / ".autobind"
d.mkdir(parents=True, exist_ok=True, mode=0o700)
return d / (owner.tty + ".json")
@contextlib.contextmanager
def _autobind_lock(path):
"""Non-blocking advisory lock so two near-simultaneous prompts on the same
tty can never both decide to mint a ticket (the "never create duplicates"
guard). A caller that loses the race gets `acquired=False` and does
nothing -- fail open, the next prompt tries again."""
lock_path = path.with_suffix(".lock")
with open(lock_path, "a") as fh:
try:
fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
except (BlockingIOError, OSError):
yield False
return
try:
yield True
finally:
fcntl.flock(fh, fcntl.LOCK_UN)
def _read_autobind_state(path):
try:
return json.loads(path.read_text())
except (OSError, ValueError):
return {}
def _write_autobind_state(path, state):
try:
atomic_write(path, json.dumps(state))
except OSError:
pass # best-effort bookkeeping; never let this raise into the hook
def _default_tk_new(owner, topic, project):
"""Shell out to the canonical `tk` CLI. Raises StatusError/OSError/
TimeoutExpired on any failure; the caller (cmd_auto_bind) always catches."""
tk_bin = Path.home() / "Projects/ticket-system/tk"
agent = "claude-%s" % owner.tty
env = dict(os.environ, TK_AGENT=agent)
out = subprocess.run([str(tk_bin), "new", topic, "-p", project],
capture_output=True, text=True, timeout=15, env=env,
cwd=str(tk_bin.parent))
if out.returncode != 0:
raise StatusError("tk new failed: " + (out.stderr or out.stdout).strip())
m = re.search(r"\bTK-\d+\b", out.stdout)
if not m:
raise StatusError("tk new produced no parseable ticket id: " + out.stdout.strip())
return m.group(0)
def _default_ticket_set(store, owner, ticket_id):
"""Same effect as the `ticket` subcommand: bind explicitly, preserving
whatever color/label the session already has."""
previous, _ = store.read(owner)
store.set(owner, previous["state"] if previous else "none",
label_ticket(previous["label"])[0] if previous else "",
ticket_update=ticket_id, variant=previous.get("variant", "") if previous else "")
# header() is a cosmetic tab-title refresh (adds the "up Nh Mm" suffix) on
# top of the dot that set() already painted -- the ticket is already bound
# in the record at this point, so a pty write hiccup here must not be
# reported as a bind failure (and must never block a headless unit test).
try:
store.header(owner)
except (StatusError, OSError):
pass
def cmd_auto_bind(store, owner, prompt, project, *, now=None, tk_new=None, ticket_set=None):
"""Bind `owner`'s tab to a ticket if it is still unbound after a
substantive prompt. Returns a dict describing what happened; never raises
(every failure mode is caught and reported as an `action`, per the
fail-open requirement -- a `tk` outage must never surface as a hook error).
"""
now = now if now is not None else time.time()
tk_new = tk_new or _default_tk_new
ticket_set = ticket_set or _default_ticket_set
state_path = _autobind_state_path(store, owner)
record, _ = store.read(owner)
already = store.ticket_info(owner, record).get("id", "")
if already:
return {"action": "already-bound", "ticket": already}
with _autobind_lock(state_path) as acquired:
if not acquired:
return {"action": "locked"} # a concurrent prompt is already deciding
state = _read_autobind_state(state_path)
if state.get("ticket"):
# We bound this session before; the record's own ticket field came
# back empty (e.g. an explicit /pinkdot --off or a clear). Re-assert
# our prior binding instead of minting a second ticket.
try:
ticket_set(store, owner, state["ticket"])
except (StatusError, OSError, subprocess.TimeoutExpired):
pass
return {"action": "rebound", "ticket": state["ticket"]}
last_attempt = float(state.get("last_attempt", 0) or 0)
if now - last_attempt < _AUTOBIND_RATE_LIMIT:
return {"action": "rate-limited"}
explicit = tickets.explicit_ticket_in_prompt(prompt)
if explicit:
try:
ticket_set(store, owner, explicit)
except (StatusError, OSError, subprocess.TimeoutExpired) as exc:
_write_autobind_state(state_path, {**state, "last_attempt": now})
return {"action": "explicit-bind-failed", "ticket": explicit, "error": str(exc)}
_write_autobind_state(state_path, {"ticket": explicit, "bound_at": now})
return {"action": "bound-explicit", "ticket": explicit}
if tickets.is_trivial_prompt(prompt):
# No last_attempt write here: the rate limit exists to throttle
# repeated FAILED `tk` attempts, not to punish "ok" / "yes" — the
# very next substantive prompt must still be free to create.
return {"action": "skipped-trivial"}
topic = tickets.short_topic(prompt)
try:
new_id = tk_new(owner, topic, project)
except (StatusError, OSError, subprocess.TimeoutExpired) as exc:
_write_autobind_state(state_path, {**state, "last_attempt": now})
return {"action": "create-failed", "error": str(exc)}
if not new_id:
# Defensive: a well-behaved tk_new raises rather than returning
# empty (see _default_tk_new), but never bind an empty id either
# way -- that would look like a real dot with no ticket at all.
_write_autobind_state(state_path, {**state, "last_attempt": now})
return {"action": "create-failed", "error": "tk_new returned no ticket id"}
if store.known_tickets is not None:
store.known_tickets.add(new_id) # just minted; discover() hasn't re-scanned yet
try:
ticket_set(store, owner, new_id)
except (StatusError, OSError, subprocess.TimeoutExpired) as exc:
_write_autobind_state(state_path, {"ticket": new_id, "bound_at": now})
return {"action": "created-bind-failed", "ticket": new_id, "error": str(exc)}
_write_autobind_state(state_path, {"ticket": new_id, "bound_at": now})
return {"action": "created", "ticket": new_id}
def main(argv=None):
parser = argparse.ArgumentParser(description=__doc__)
sub = parser.add_subparsers(dest="command", required=True)
sv = sub.add_parser("set-variant")
sv.add_argument("variant", nargs="?", default="", choices=["", "monitoring", "stopped", "waiting"])
sv.add_argument("--if-blocked", action="store_true",
help="only apply when the base dot is a NEEDS-STEVE colour (yellow/purple/"
"orange); a cheap no-op otherwise. Used by the Stop hook so the 🔵 "
"umbrella marks only stops that actually need Steve, not every stop.")
sv.add_argument("--tty", default="",
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",
"auto-bind"):
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="")
cmd.add_argument("--all", choices=["claude", "codex"])
cmd.add_argument("--tty", default="",
help="target ANOTHER live session's tty (e.g. ttys018). The owner "
"record must still match the live process table, so a stale "
"writer is refused exactly as for the caller's own tab.")
cmd.add_argument("--monitoring", action="store_true")
cmd.add_argument("--waiting", action="store_true",
help="brighter STATIC green (rgb 0,255,120) for an idle/"
"waiting green session; a green-only tint like --monitoring")
cmd.add_argument("--stopped", action="store_true",
help="keep the base colour dot and place \U0001F535 next to it "
"(any stop that requires Steve's input)")
cmd.add_argument("--deferential", action="store_true",
help="automatic paint: yield to a sticky semantic dot "
"(orange/purple/yellow) instead of overwriting it")
if name == "ticket":
cmd.add_argument("ticket")
if name == "auto-bind":
# TK-12168: called from the UserPromptSubmit hook (working-state.sh),
# backgrounded, so a slow/failed `tk` call never delays the prompt.
cmd.add_argument("--prompt", default="",
help="the user's raw prompt text; read from stdin if omitted")
cmd.add_argument("--quiet", action="store_true")
legacy = sub.add_parser("paint-legacy")
legacy.add_argument("tty")
legacy.add_argument("red", type=int)
legacy.add_argument("green", type=int)
legacy.add_argument("blue", type=int)
legacy.add_argument("title")
legacy.add_argument("off", nargs="?", default="0", choices=["0", "1"])
for name in ("scan", "audit"):
cmd = sub.add_parser(name)
cmd.add_argument("--json", action="store_true")
cmd.add_argument("--tsv", action="store_true")
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)
args = parser.parse_args(argv)
if args.command == "selftest":
return selftest()
store = Store()
if args.command == "backfill":
# TK-11870: the SOLE scheduled process-table PRODUCER + the always-on
# self-heal. One fresh scan here (a) warms the cross-process disk cache
# (_disk_cache_put) so the ~49 interactive sessions read IT instead of each
# racing their own `ps -ax` — collapsing the scan herd that drove the >90s
# timeout that silently dropped dots — and (b) re-asserts every live tab's
# dot from its stored record. No tickets.discover() (that adds a ~4.4s
# `ps eww` env dump we don't need): backfill only REPAINTS existing records,
# it never resolves or creates a ticket label. Runs from a clean launchd env
# (no CLAUDE_CODE_CHILD_SESSION), so the child-session paint rail never trips.
# The `ps` scan is the very input a ps-timeout kills -- catch it so the
# heartbeat CARRIES the failure (WARN + NOT-MEASURED) instead of the old
# path where processes() raised and NO heartbeat was written (silence that
# is byte-identical to an idle box). Still fail loud below (nonzero exit +
# stderr) so launchd + the drift canary also see it.
scan_error = None
try:
rows = processes(fresh=True)
except StatusError as exc:
scan_error = str(exc)
painted = floored = skipped = 0
live_owners = None
else:
painted, floored, skipped = store.backfill(rows)
live_owners = len(owners(rows))
# Liveness heartbeat, written AFTER the side effect (memory
# liveness-artifact-must-follow-the-side-effect): its mtime proves the sole
# producer actually FIRED (not just that launchd reports it "running"), and
# its verdict is derived from what the cycle measured (backfill_heartbeat),
# never a hardcoded PASS. Bounded (overwritten each run). Never breaks the
# paint if the write fails.
try:
hb = store.user_root / "Projects/terminal-status/data/backfill-latest.json"
hb.parent.mkdir(parents=True, exist_ok=True)
atomic_write(hb, json.dumps(
backfill_heartbeat(painted, floored, skipped, live_owners,
scan_error=scan_error), indent=2) + "\n")
except OSError:
pass
if scan_error:
# Fail loud: stderr + nonzero exit, so the failure is not visible ONLY
# in a heartbeat nobody happens to read.
print("backfill: " + scan_error, file=sys.stderr)
return 1
if not getattr(args, "quiet", False):
print("backfill: repainted %d, floored %d, skipped %d"
% (painted, floored, skipped))
return 0
rows = processes()
# TK-11835: tickets.discover() runs TWO `ps` subprocess calls (a `ps -p` argv
# read AND a `ps eww` ENV dump) over every live session PURELY to enrich the
# tab's ticket LABEL for a session that hasn't declared one otherwise. Measured
# ~4.4s under overnight load (the `ps eww` env dump over ~43 pids dominates),
# and it runs on EVERY invocation before command dispatch -- so a cross-tty
# sweep pays it per paint, and N concurrent sweepers saturate the process table
# into the ~30s spiral this ticket reports. A CROSS-TTY paint addresses ONE
# target with an EXPLICIT label/ticket and never needs the argv/session-ledger
# FALLBACK binding, so pass discover its argv/env skip-seam ({} = "already
# resolved, don't scan"). The primary events.jsonl binding + known_tickets
# (explicit-ticket validation) still load fully; only the ps enrichment is
# skipped. A session painting its OWN tab keeps full enrichment (argv/env None).
_cross = bool(getattr(args, "tty", "")) or args.command == "paint-legacy"
store.ticket_evidence, store.known_tickets = tickets.discover(
store.user_root, rows, owners(rows), ancestors,
argv={} if _cross else None, env={} if _cross else None)
if args.command in ("scan", "audit"):
data = scan(store, rows)
if args.tsv:
for r in data:
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:
for r in data:
notes = ", ".join(r.get("warnings", []))
print(f'{r["tty"]} {r["label"]} [{r["status_source"]}] {notes}')
if args.command == "audit" and any(
r.get("warnings") or r["status_source"] in
("invalid", "owner_changed", "legacy_conflict") for r in data):
return 1
return 0
if args.command == "headers":
failures = []
for owner in owners(rows).values():
try:
title = store.header(owner)
if not args.quiet:
print(owner.tty + " " + title)
except (StatusError, OSError) as exc:
failures.append(owner.tty + ": " + str(exc))
if failures:
raise StatusError("; ".join(failures))
return 0
# An EXTERNAL supervisor (the launchd greendot-agent) has no terminal ancestry of its
# own, so current_owner() raises "No owning Claude/Codex terminal in this process
# ancestry". When such a caller addresses a session EXPLICITLY with --tty it never needs
# 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", "auto-bind") or (
args.command == "current" and getattr(args, "paintable", False))
# TK-11835: a CROSS-TTY write (an external supervisor addressing another
# session with --tty, or paint-legacy) must fail FAST on a busy target rather
# than inherit the 65s self-writer lock backstop — a sweep across every tab
# cannot afford to block ~30-65s on each mid-turn session. Give the target a
# short, env-overridable budget and reuse main's already-resolved `rows` so
# the in-lock owner check does not re-scan. A session painting its OWN tab
# (no --tty) keeps the original path unchanged (cross_lock_wait/cross_rows None).
cross_tty = bool(getattr(args, "tty", "")) or args.command == "paint-legacy"
cross_lock_wait = float(
os.environ.get("TERMINAL_STATUS_TTY_LOCK_WAIT", "2.5")) if cross_tty else None
cross_rows = rows if cross_tty else None
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":
# Use the already-resolved caller + the real store instance (the branch shipped calling
# a nonexistent Owner.detect()/STORE, so set-variant crashed in EVERY session — the
# auto-🔵 stopped-marker never fired). Matches every other command's resolution. (2026-09-11)
target = caller
if getattr(args, "tty", ""):
_t = args.tty.removeprefix("/dev/")
if _t not in owners(rows):
raise StatusError("No unique live main agent owns this tty")
target = owners(rows)[_t]
if getattr(args, "if_blocked", False):
prev, _r = store.read(target)
if not prev or prev.get("state") not in ("yellow", "purple", "orange"):
return 0 # not a needs-Steve stop → no 🔵 umbrella (cheap read, no paint)
record = store.set_variant(target, args.variant,
rows=cross_rows, lock_wait=cross_lock_wait)
print(f'/terminal-status → /dev/{target.tty} {record["title"] or "CLEARED"}')
return 0
if args.command == "current":
print(json.dumps(dataclasses.asdict(caller)))
return 0
if args.command == "status":
print(json.dumps(store.row(caller), ensure_ascii=False))
return 0
if args.command == "ticket":
ticket = args.ticket.upper()
ticket = "" if ticket in ("CLEAR", "NONE") else ("TK-" + ticket if ticket.isdigit() else tickets.short(ticket))
if not ticket and args.ticket.upper() not in ("CLEAR", "NONE"):
raise StatusError("Expected a canonical TK number or clear")
previous, _ = store.read(caller)
record = store.set(caller, previous["state"] if previous else "none",
label_ticket(previous["label"])[0] if previous else "",
ticket_update=ticket, variant=previous.get("variant", "") if previous else "")
store.header(caller)
print(f'/terminal-ticket → {caller.tty} {record["ticket"] or "TK REQUIRED"}')
return 0
if args.command == "auto-bind":
prompt = args.prompt
if not prompt and not sys.stdin.isatty():
try:
prompt = sys.stdin.read()
except (OSError, ValueError):
prompt = ""
project = Path.cwd().name
result = cmd_auto_bind(store, caller, prompt, project)
if not getattr(args, "quiet", False):
print(json.dumps(result))
return 0
targets = [caller]
if args.command != "paint-legacy" and getattr(args, "tty", ""):
# External agents (e.g. greendot-agent) repaint another session's dot. This is
# legitimate: assert_owner() still verifies the OWNER RECORD matches the live
# process table, so a dead or reassigned tty is refused — it simply does not
# require the CALLER to be that process, which is the same basis on which
# "set --all" already paints every tab.
_t = args.tty.removeprefix("/dev/")
if _t not in owners(rows):
raise StatusError("No unique live main agent owns this tty")
targets = [owners(rows)[_t]]
if args.command == "paint-legacy":
tty = args.tty.removeprefix("/dev/")
if tty not in owners(rows):
raise StatusError("No unique live main agent owns this tty")
targets = [owners(rows)[tty]]
color = "none" if args.off == "1" else color_of(args.title)
if color == "none" and args.off != "1":
raise StatusError("Unknown legacy semantic color")
if color != "none" and COLORS[color][1] != (args.red, args.green, args.blue):
raise StatusError("Legacy color and RGB disagree")
label = args.title[len(COLORS[color][0]):].strip() if color != "none" else ""
elif args.command == "set":
color, label = args.color, args.label
if args.all:
targets = [o for o in owners(rows).values() if o.runtime == args.all]
elif args.command == "clear":
color, label = "none", ""
for owner in targets:
if args.command == "start":
record, reason = store.load(owner)
if record is not None:
record = store.repaint(owner)
elif reason == "missing":
sessions, api = iterm_sessions()
legacy, legacy_reason = store.legacy_status(owner, sessions.get(owner.tty, ""))
# A cross-process cache HIT ("cached", <30s old) is as trustworthy as a
# live enumeration for the legacy-conflict check, so it restores a sticky
# semantic dot instead of falling through to green (TK-11879). "stale" and
# "unavailable" stay excluded — an older live title is not a safe basis for
# deciding a conflict, so those keep the existing floor/raise behaviour.
if legacy is not None and api in ("available", "cached"):
record = store.set(owner, legacy["state"], legacy["label"])
elif legacy_reason in ("missing", "legacy_stale", "legacy_empty"):
record = store.set(owner, "green")
else:
raise StatusError("Ambiguous legacy state; set an explicit color")
elif reason == "owner_changed":
record = store.set(owner, "green")
else:
raise StatusError("Start found ambiguous status; set an explicit color")
elif args.command == "repaint":
record = store.repaint(owner)
else:
# Variant precedence when more than one flag is somehow passed: stopped (the
# needs-Steve additive marker) wins over the two green tints; between the tints,
# monitoring (teal — an active scheduled deadline) wins over waiting (brighter
# idle green), since a live deadline is the more informative signal.
record = store.set(owner, color, label,
variant=("stopped" if getattr(args, "stopped", False)
else "monitoring" if getattr(args, "monitoring", False)
else "waiting" if getattr(args, "waiting", False) else ""),
deferential=getattr(args, "deferential", False),
rows=cross_rows, lock_wait=cross_lock_wait)
if not getattr(args, "quiet", False):
print(f'/terminal-status → /dev/{owner.tty} {record["title"] or "CLEARED"}')
return 0
if __name__ == "__main__":
try:
sys.exit(main())
except (StatusError, OSError, subprocess.TimeoutExpired) as exc:
print("terminal-status: " + str(exc), file=sys.stderr)
sys.exit(1)