← back to Terminal Status
tk_required_fp_metric.py
376 lines
#!/usr/bin/env python3
"""TK-11666 — measure the TK-REQUIRED FALSE-POSITIVE RATE (the missing denominator).
The monitoring gap that made the "tabs stuck on TK REQUIRED" pattern chronic was
not a missing alert — it was a missing DENOMINATOR. terminal-status-drift-canary
watches STATUS-RECORD freshness; NOTHING answered "of the live green sessions
currently labelled TK REQUIRED, how many are PROVABLY TRACKED (i.e. FALSE
POSITIVES)?" Because nothing computed it, three masterdot passes re-confirmed the
same symptom, inherited the same wrong cause, and opened ~24 spurious tickets.
This is a READ-ONLY measurement. It NEVER writes a ticket, never repaints a tab,
never touches Shopify/prod. Three honest buckets (CLAUDE.md TK-11431 rule 1):
bound — the pane's DISPLAYED label already carries a resolved ticket.
not_measured — labelled "TK REQUIRED" AND no ticket resolvable by ANY channel
(argv TK_AGENT/id, ledger claude@TERM_SESSION_ID). NOT an
accusation; NEVER counted green; the honest unbindable state.
contradicted — labelled "TK REQUIRED" YET independently PROVABLY TRACKED.
THE FALSE POSITIVE — the number to drive to zero.
INDEPENDENCE (TK-11431: a check must not measure the wrong thing): the LABEL is
read from the PERSISTED live dot state (allcolordots --json — what Steve actually
sees), while the resolvability probe re-scans the RAW ledger + RAW argv itself. It
does NOT call the binder it is grading, so a binder bug that leaves a tracked
session reading "TK REQUIRED" is visible as `contradicted` instead of being
laundered into agreement. Population is carried beside observed so "0 of 0" is
distinguishable from "0 of 60". Sessions are keyed on PID, never tty (a tty is a
reusable slot — memory tty-is-a-reusable-slot-not-a-session-identity).
Verdict (fleet-health-rollup PASS/WARN/FAIL vocabulary):
FAIL — contradicted > 0 (a provable false positive exists; drive to zero).
WARN — could NOT observe an input (allcolordots/ledger/ps unavailable) or zero
live sessions enumerated — an unmeasured input is NEVER a green PASS.
PASS — >=1 live session observed AND contradicted == 0 (not_measured is fine).
"""
import datetime as dt
import json
import os
import re
import subprocess
import sys
from pathlib import Path
HOME = Path.home()
LEDGER = HOME / ".claude/tickets/events.jsonl"
ALLCOLORDOTS = HOME / ".claude/skills/allcolordots/allcolordots.sh"
# A session younger than this grace has not necessarily had its first ledger
# event / 60s repaint yet, so a TK-REQUIRED-shaped newborn is reported as
# `pending`, NOT `contradicted` — this is the launch-latency noise that produced
# ~24 spurious tickets. Override with TK_FP_MIN_AGE_S.
MIN_AGE_S = float(os.environ.get("TK_FP_MIN_AGE_S", "180"))
WORK_TYPES = ("assign", "create", "action", "comment") # `read` is board-viewing noise
SHORT = re.compile(r"^TK-\d+(?=$|-)", re.I)
# Same shapes ticket_binding resolves, reimplemented here so the metric is
# INDEPENDENT of the binder (it must be able to disagree with it).
AGENT_EXPORT = re.compile(
r"\bexport TK_AGENT=(?:claude-run|codex-run|local-qwen-27b-run"
r"|local-qwen-14b-run|local-qwen-14b-mac1-run)-(\d+)\.", re.I)
TK_BARE = re.compile(r"\bTK-\d+\b", re.I)
TK_REQUIRED = "TK REQUIRED"
def short(value):
m = SHORT.match(str(value or ""))
return m.group().upper() if m else ""
def classify(sessions, ledger_by_tsid, argv_by_pid, known, min_age_s=MIN_AGE_S):
"""Pure, testable core. Inputs are already-gathered so tests inject fixtures.
sessions : list of {tty, pid, color, label, tsid, age_s}
ledger_by_tsid: {tsid: [(ts_epoch, ticket_short)]} work events only
argv_by_pid : {pid: command_string}
known : set of created ticket short-ids (resolvability safety)
"""
buckets = {"bound": [], "not_measured": [], "contradicted": [], "pending": []}
for s in sessions:
label = s.get("label", "") or ""
# Only GREEN sessions count toward the false-positive question (the ticket
# frames it as "live GREEN sessions currently labelled TK REQUIRED").
if s.get("color") != "green":
continue
if TK_REQUIRED not in label:
buckets["bound"].append(s)
continue
# Labelled TK REQUIRED — probe resolvability INDEPENDENTLY.
via = _resolve(s, ledger_by_tsid, argv_by_pid, known)
if not via:
buckets["not_measured"].append(dict(s, resolvable=False))
continue
entry = dict(s, resolvable=True, resolved_via=via["via"], resolved_ticket=via["id"])
age = s.get("age_s")
# Defer ONLY a session KNOWN to be younger than the launch grace. An
# UNKNOWN age must NOT be excused as "young" — that would hide a real false
# positive (measuring the wrong thing). Unknown age → eligible/contradicted.
if age is not None and age < min_age_s:
buckets["pending"].append(entry) # tracked but too young to blame
else:
buckets["contradicted"].append(entry) # THE false positive
return buckets
def _resolve(session, ledger_by_tsid, argv_by_pid, known):
"""Independent proof-of-tracked for a pane. Ledger first (strongest: the
session actually LOGGED work), then argv. Never invents a ticket: the
resolved id must be in `known` (a ticket we have seen created)."""
tsid = session.get("tsid")
events = ledger_by_tsid.get(tsid) if tsid else None
if events:
# most recent work event under claude@<tsid>
ts_epoch, ticket = max(events, key=lambda e: e[0])
if short(ticket) in known:
return {"via": "ledger", "id": short(ticket)}
command = argv_by_pid.get(session.get("pid"), "") or ""
declared = set(AGENT_EXPORT.findall(command))
if len(declared) == 1 and "TK-" + next(iter(declared)) in known:
return {"via": "argv_agent", "id": "TK-" + next(iter(declared))}
ids = set(m.upper() for m in TK_BARE.findall(command))
if len(ids) == 1 and next(iter(ids)) in known:
return {"via": "argv_id", "id": next(iter(ids))}
return None
def verdict_of(buckets, observed_ok):
contradicted = len(buckets["contradicted"])
green = sum(len(buckets[b]) for b in ("bound", "not_measured", "contradicted", "pending"))
if contradicted > 0:
return "FAIL"
if not observed_ok or green == 0:
return "WARN" # unmeasured input / empty enumeration is NEVER a green PASS
return "PASS"
# ---------- real-world data gathering (production path; no test fixtures) ----------
def gather_sessions():
"""LIVE sessions from allcolordots --json (never from ~/.claude/tab-dots/*.dot,
which is 33% stale phantoms). Returns (sessions, ok)."""
try:
out = subprocess.run(["bash", str(ALLCOLORDOTS), "--json"],
capture_output=True, text=True, timeout=90)
data = json.loads(out.stdout)
except Exception:
return [], False
ages = _proc_ages([d.get("pid") for d in data])
tsids = _term_session_ids([d.get("pid") for d in data])
sessions = []
for d in data:
if not d.get("live"):
continue
pid = _int(d.get("pid"))
sessions.append({"tty": d.get("tty"), "pid": pid, "color": d.get("color"),
"label": d.get("label", ""), "tsid": tsids.get(pid),
"age_s": ages.get(pid)})
return sessions, True
def _int(v):
try:
return int(v)
except (TypeError, ValueError):
return None
def _proc_ages(pids):
"""PID -> age in seconds. macOS `ps` has NO `etimes`; use `lstart` (start
wall-clock) and subtract now, parsed with the same format terminal_status
uses for Owner.epoch. A pid whose age can't be parsed is simply absent (age
None downstream → treated as PAST the grace, never hidden as young)."""
pids = [p for p in (_int(x) for x in pids) if p]
ages = {}
if not pids:
return ages
now = dt.datetime.now().timestamp()
try:
out = subprocess.run(["ps", "-o", "pid=,lstart=", "-p", ",".join(map(str, pids))],
capture_output=True, text=True, timeout=90)
for line in out.stdout.splitlines():
parts = line.strip().split(None, 1)
if len(parts) != 2:
continue
try:
start = dt.datetime.strptime(parts[1].strip(), "%a %b %d %H:%M:%S %Y").timestamp()
ages[int(parts[0])] = max(0.0, now - start)
except (ValueError, OverflowError):
continue
except Exception:
pass
return ages
def _term_session_ids(pids):
"""PID -> TERM_SESSION_ID from process ENV. `ps eww` prints ARGS AND ENV
concatenated, so strip the clean args (from `-o args=`) before matching —
otherwise a TERM_SESSION_ID sitting in a prompt argument is a false hit."""
pids = [p for p in (_int(x) for x in pids) if p]
out_ids = {}
if not pids:
return out_ids
argv = {}
try:
a = subprocess.run(["ps", "-o", "pid=,args=", "-p", ",".join(map(str, pids))],
capture_output=True, text=True, timeout=90)
for line in a.stdout.splitlines():
p = line.strip().split(None, 1)
if len(p) == 2:
argv[int(p[0])] = p[1]
e = subprocess.run(["ps", "eww", "-o", "pid=,command=", "-p", ",".join(map(str, pids))],
capture_output=True, text=True, timeout=90)
for line in e.stdout.splitlines():
p = line.strip().split(None, 1)
if len(p) != 2:
continue
pid, full = int(p[0]), p[1]
args = argv.get(pid, "")
rem = full[len(args):] if args and full.startswith(args) else ""
m = re.search(r"TERM_SESSION_ID=(\S+)", rem)
if m:
out_ids[pid] = m.group(1)
except Exception:
pass
return out_ids
def gather_argv(pids):
argv = {}
pids = [p for p in (_int(x) for x in pids) if p]
if not pids:
return argv, True
try:
out = subprocess.run(["ps", "-o", "pid=,args=", "-p", ",".join(map(str, pids))],
capture_output=True, text=True, timeout=90)
for line in out.stdout.splitlines():
p = line.strip().split(None, 1)
if len(p) == 2:
argv[int(p[0])] = p[1]
return argv, True
except Exception:
return argv, False
def gather_ledger():
"""Raw ledger scan (INDEPENDENT of the binder). Returns
({tsid: [(ts_epoch, ticket_short)]}, known_short_ids, ok)."""
by_tsid, known = {}, set()
try:
stream = LEDGER.open()
except OSError:
return by_tsid, known, False
with stream:
for line in stream:
try:
ev = json.loads(line)
except ValueError:
continue
tk = short(ev.get("id", ""))
if not tk:
continue
if ev.get("type") == "create":
known.add(tk)
agent = ev.get("agent", "")
if agent.startswith("claude@") and ev.get("type") in WORK_TYPES:
tsid = agent[len("claude@"):]
by_tsid.setdefault(tsid, []).append((_ts(ev.get("ts")), tk))
return by_tsid, known, True
def _ts(value):
try:
return dt.datetime.fromisoformat(str(value).replace("Z", "+00:00")).timestamp()
except (ValueError, AttributeError):
return 0.0
def measure():
sessions, sess_ok = gather_sessions()
ledger_by_tsid, known, ledger_ok = gather_ledger()
argv_by_pid, argv_ok = gather_argv([s["pid"] for s in sessions])
observed_ok = sess_ok and ledger_ok and argv_ok and len(sessions) > 0
buckets = classify(sessions, ledger_by_tsid, argv_by_pid, known)
v = verdict_of(buckets, observed_ok)
green = sum(len(buckets[b]) for b in ("bound", "not_measured", "contradicted", "pending"))
tk_required_green = green - len(buckets["bound"])
fp_rate = (len(buckets["contradicted"]) / tk_required_green) if tk_required_green else 0.0
detail = lambda b: [{"tty": s["tty"], "pid": s["pid"], "tsid": s.get("tsid"),
"via": s.get("resolved_via"), "ticket": s.get("resolved_ticket"),
"age_s": s.get("age_s")} for s in buckets[b]]
return {
"skill": "tk-required-fp-metric",
"ts": dt.datetime.now(dt.timezone.utc).isoformat(),
"verdict": v, "status": v,
"measured": observed_ok,
"population": {"live_sessions": len(sessions),
"green_sessions": green,
"green_tk_required": tk_required_green},
"observed": {"bound": len(buckets["bound"]),
"not_measured": len(buckets["not_measured"]),
"contradicted": len(buckets["contradicted"]),
"pending": len(buckets["pending"])},
"false_positive_rate": round(fp_rate, 4),
"contradicted_detail": detail("contradicted"),
"not_measured_detail": detail("not_measured"),
"inputs_ok": {"allcolordots": sess_ok, "ledger": ledger_ok, "argv": argv_ok},
"note": ("contradicted = live green tab reads 'TK REQUIRED' yet is provably "
"tracked (the false positive to drive to 0); not_measured is honest "
"and NOT an accusation; pending = tracked but younger than the "
f"{int(MIN_AGE_S)}s launch grace."),
}
def data_dir():
return Path(os.environ.get(
"TK_FP_DATA_DIR",
str(HOME / ".claude/skills/tk-required-fp-metric/data")))
def write_latest(result):
d = data_dir()
d.mkdir(parents=True, exist_ok=True)
(d / "latest.json").write_text(json.dumps(result, indent=2))
return d / "latest.json"
def _self_test():
"""TK-11431 rule 3: prove the metric goes RED on an injected `contradicted`
row and is non-vacuous (clean set → PASS). Guarded behind an explicit flag a
scheduled run NEVER passes, so a plist can never measure this fixture."""
known = {"TK-11666", "TK-11317"}
# A live GREEN pane reading "TK REQUIRED" whose TERM_SESSION_ID HAS logged work
# on a known ticket = the false positive.
contradicted_session = {"tty": "ttys099", "pid": 4242, "color": "green",
"label": "🟢 TK REQUIRED · WORKING", "tsid": "wXtYpZ:FP",
"age_s": 9000}
ledger = {"wXtYpZ:FP": [(1000.0, "TK-11666-fix-x")]}
hot = classify([contradicted_session], ledger, {}, known)
assert len(hot["contradicted"]) == 1, hot
assert verdict_of(hot, True) == "FAIL", "must FAIL on a contradicted row"
# Remove the fault → clean PASS (a bound session, no false positive).
clean = classify([{"tty": "ttys098", "pid": 4243, "color": "green",
"label": "🟢 TK-11317 · WORKING", "tsid": "wAtBpC:OK",
"age_s": 9000}], {}, {}, known)
assert len(clean["contradicted"]) == 0 and verdict_of(clean, True) == "PASS", clean
# A genuinely ticketless pane is not_measured (never green, never accused).
nm = classify([{"tty": "ttys097", "pid": 4244, "color": "green",
"label": "🟢 TK REQUIRED · WORKING", "tsid": "wPpQ:NONE", "age_s": 9000}],
{}, {}, known)
assert len(nm["not_measured"]) == 1 and len(nm["contradicted"]) == 0, nm
assert verdict_of(nm, True) == "PASS", "not_measured alone does not fail"
# Same shape but too young → pending, not contradicted (launch-latency guard).
young = classify([dict(contradicted_session, age_s=5)], ledger, {}, known)
assert len(young["pending"]) == 1 and len(young["contradicted"]) == 0, young
print("self-test OK: contradicted→FAIL, clean→PASS, not_measured→PASS, young→pending")
def main(argv):
if "--self-test" in argv:
_self_test()
return 0
result = measure()
path = write_latest(result)
if "--json" in argv:
print(json.dumps(result, indent=2))
else:
o = result["observed"]
p = result["population"]
print(f"{result['verdict']} tk-required-fp-metric "
f"live={p['live_sessions']} green={p['green_sessions']} "
f"bound={o['bound']} not_measured={o['not_measured']} "
f"contradicted={o['contradicted']} pending={o['pending']} "
f"fp_rate={result['false_positive_rate']} → {path}")
return 0
if __name__ == "__main__":
sys.exit(main(sys.argv[1:]))