← back to Terminal Status
TK-11666: measure the TK-REQUIRED false-positive rate (the missing denominator)
9731e7fc51e20dd9cd99d7c473c90ab08ae22bf1 · 2026-09-14 00:37:26 -0700 · Steve Abrams
Add a READ-ONLY metric answering the question nothing computed — "of the live
GREEN sessions labelled TK REQUIRED, how many are PROVABLY TRACKED (false
positives)?" — the missing correctness metric that let the pattern recur 3x/day
and spawned ~24 spurious tickets.
Three honest buckets (TK-11431 rule 1): bound / not_measured (never green, never
an accusation) / contradicted (the false positive to drive to 0). Carries
population beside observed ("0 of 0" != "0 of 60"), counts LIVE sessions via
allcolordots --json (never the stale tab-dots dir), keys on PID not tty.
Independence: the LABEL is the persisted dot state (what Steve sees) while the
resolvability probe re-scans the RAW ledger (claude@TERM_SESSION_ID work events)
and RAW argv itself — it does NOT call the binder it grades, so a binder gap
that leaves a tracked session reading TK REQUIRED surfaces as contradicted
instead of being laundered into agreement. Coupled to TK-11665: as that binder
improves, contradicted → 0.
Launch-latency guard: a tracked-but-young session is `pending`, not
contradicted; an UNKNOWN age is NEVER excused as young (fail-safe toward
detecting the false positive). Age uses macOS `ps lstart` (there is no etimes).
Verdict: FAIL contradicted>0 · WARN input-unobservable/zero-sessions (never a
green PASS) · PASS >=1 session and contradicted==0. Ships a negative test
(--self-test + 13 unit tests) proven to go RED on an injected contradicted row;
the test seam is the pure classify() function, so a scheduled run never measures
a fixture. Emits verdict+status (canonical PASS/WARN/FAIL) to the skill's
data/latest.json for fleet-health-rollup (no rollup vocabulary change needed).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UVgEj4WxSqB62ieGpgrixR
Files touched
A test_tk_required_fp_metric.pyA tk_required_fp_metric.py
Diff
commit 9731e7fc51e20dd9cd99d7c473c90ab08ae22bf1
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Mon Sep 14 00:37:26 2026 -0700
TK-11666: measure the TK-REQUIRED false-positive rate (the missing denominator)
Add a READ-ONLY metric answering the question nothing computed — "of the live
GREEN sessions labelled TK REQUIRED, how many are PROVABLY TRACKED (false
positives)?" — the missing correctness metric that let the pattern recur 3x/day
and spawned ~24 spurious tickets.
Three honest buckets (TK-11431 rule 1): bound / not_measured (never green, never
an accusation) / contradicted (the false positive to drive to 0). Carries
population beside observed ("0 of 0" != "0 of 60"), counts LIVE sessions via
allcolordots --json (never the stale tab-dots dir), keys on PID not tty.
Independence: the LABEL is the persisted dot state (what Steve sees) while the
resolvability probe re-scans the RAW ledger (claude@TERM_SESSION_ID work events)
and RAW argv itself — it does NOT call the binder it grades, so a binder gap
that leaves a tracked session reading TK REQUIRED surfaces as contradicted
instead of being laundered into agreement. Coupled to TK-11665: as that binder
improves, contradicted → 0.
Launch-latency guard: a tracked-but-young session is `pending`, not
contradicted; an UNKNOWN age is NEVER excused as young (fail-safe toward
detecting the false positive). Age uses macOS `ps lstart` (there is no etimes).
Verdict: FAIL contradicted>0 · WARN input-unobservable/zero-sessions (never a
green PASS) · PASS >=1 session and contradicted==0. Ships a negative test
(--self-test + 13 unit tests) proven to go RED on an injected contradicted row;
the test seam is the pure classify() function, so a scheduled run never measures
a fixture. Emits verdict+status (canonical PASS/WARN/FAIL) to the skill's
data/latest.json for fleet-health-rollup (no rollup vocabulary change needed).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UVgEj4WxSqB62ieGpgrixR
---
test_tk_required_fp_metric.py | 125 ++++++++++++++
tk_required_fp_metric.py | 375 ++++++++++++++++++++++++++++++++++++++++++
2 files changed, 500 insertions(+)
diff --git a/test_tk_required_fp_metric.py b/test_tk_required_fp_metric.py
new file mode 100644
index 0000000..4107662
--- /dev/null
+++ b/test_tk_required_fp_metric.py
@@ -0,0 +1,125 @@
+"""TK-11666 — tests for the TK-REQUIRED false-positive metric.
+
+Includes the mandatory NEGATIVE test (TK-11431 rule 3): the metric must go RED on
+an injected `contradicted` row and be non-vacuous."""
+import unittest
+
+import tk_required_fp_metric as m
+
+
+class MetricTests(unittest.TestCase):
+ KNOWN = {"TK-11666", "TK-11317"}
+
+ def green(self, tty, pid, tsid, label, age_s=9000, color="green"):
+ return {"tty": tty, "pid": pid, "tsid": tsid, "label": label,
+ "age_s": age_s, "color": color}
+
+ def test_contradicted_row_goes_red(self):
+ # A live GREEN pane reading "TK REQUIRED" whose TERM_SESSION_ID HAS logged
+ # work on a KNOWN ticket = the false positive. Metric must FAIL.
+ s = self.green("ttys099", 4242, "wXtYpZ:FP", "🟢 TK REQUIRED · WORKING")
+ ledger = {"wXtYpZ:FP": [(1000.0, "TK-11666-fix-x")]}
+ b = m.classify([s], ledger, {}, self.KNOWN)
+ self.assertEqual(len(b["contradicted"]), 1)
+ self.assertEqual(b["contradicted"][0]["resolved_via"], "ledger")
+ self.assertEqual(m.verdict_of(b, True), "FAIL")
+
+ def test_argv_channel_also_contradicts(self):
+ s = self.green("ttys099", 4242, "wX:none", "🟢 TK REQUIRED · WORKING")
+ argv = {4242: "claude export TK_AGENT=claude-run-11666. You are driving TK-11666-x."}
+ b = m.classify([s], {}, argv, self.KNOWN)
+ self.assertEqual(len(b["contradicted"]), 1)
+ self.assertEqual(b["contradicted"][0]["resolved_via"], "argv_agent")
+ self.assertEqual(m.verdict_of(b, True), "FAIL")
+
+ def test_clean_bound_session_passes(self):
+ s = self.green("ttys098", 4243, "wA:OK", "🟢 TK-11317 · WORKING")
+ b = m.classify([s], {}, {}, self.KNOWN)
+ self.assertEqual(len(b["contradicted"]), 0)
+ self.assertEqual(len(b["bound"]), 1)
+ self.assertEqual(m.verdict_of(b, True), "PASS")
+
+ def test_not_measured_is_never_an_accusation_and_does_not_fail(self):
+ # TK REQUIRED with NO resolvable channel = honest not_measured, not a false
+ # positive, and it does NOT block PASS (rule 1: never an accusation).
+ s = self.green("ttys097", 4244, "wP:NONE", "🟢 TK REQUIRED · WORKING")
+ b = m.classify([s], {}, {}, self.KNOWN)
+ self.assertEqual(len(b["not_measured"]), 1)
+ self.assertEqual(len(b["contradicted"]), 0)
+ self.assertEqual(m.verdict_of(b, True), "PASS")
+
+ def test_ledger_ticket_must_be_known(self):
+ # A ledger event under this session for a ticket NEVER created cannot prove
+ # tracking (never invents a ticket) → not_measured, not contradicted.
+ s = self.green("ttys096", 4245, "wG:GHOST", "🟢 TK REQUIRED · WORKING")
+ ledger = {"wG:GHOST": [(1000.0, "TK-99999999-ghost")]}
+ b = m.classify([s], ledger, {}, self.KNOWN)
+ self.assertEqual(len(b["contradicted"]), 0)
+ self.assertEqual(len(b["not_measured"]), 1)
+
+ def test_read_events_are_noise_not_tracking(self):
+ # `read` events are excluded upstream in gather_ledger (only WORK_TYPES are
+ # collected), so a session that only READ the board has no ledger entry and
+ # stays not_measured.
+ self.assertNotIn("read", m.WORK_TYPES)
+ s = self.green("ttys095", 4246, "wR:READONLY", "🟢 TK REQUIRED · WORKING")
+ b = m.classify([s], {}, {}, self.KNOWN) # no work events for this tsid
+ self.assertEqual(len(b["not_measured"]), 1)
+ self.assertEqual(len(b["contradicted"]), 0)
+
+ def test_launch_grace_defers_young_sessions_to_pending(self):
+ # A tracked-but-young session is `pending`, never `contradicted` — the
+ # launch-latency noise that spawned ~24 spurious tickets.
+ s = self.green("ttys094", 4247, "wY:YOUNG", "🟢 TK REQUIRED · WORKING", age_s=5)
+ ledger = {"wY:YOUNG": [(1000.0, "TK-11666-x")]}
+ b = m.classify([s], ledger, {}, self.KNOWN)
+ self.assertEqual(len(b["pending"]), 1)
+ self.assertEqual(len(b["contradicted"]), 0)
+ self.assertEqual(m.verdict_of(b, True), "PASS")
+
+ def test_empty_enumeration_is_warn_not_pass(self):
+ # Rule 1: an enumeration that returned zero rows is WARN, never a green PASS
+ # ("0 of 0" != "0 of 60").
+ b = m.classify([], {}, {}, self.KNOWN)
+ self.assertEqual(m.verdict_of(b, observed_ok=False), "WARN")
+
+ def test_unmeasured_input_is_warn_not_pass(self):
+ # allcolordots/ledger unavailable → observed_ok False → WARN even with rows.
+ s = self.green("ttys093", 4248, "wU:UNSEEN", "🟢 TK-11317 · WORKING")
+ b = m.classify([s], {}, {}, self.KNOWN)
+ self.assertEqual(m.verdict_of(b, observed_ok=False), "WARN")
+
+ def test_only_green_sessions_are_scored(self):
+ # A purple/gated pane reading TK REQUIRED is NOT part of the false-positive
+ # question (the ticket scopes it to live GREEN sessions).
+ s = self.green("ttys092", 4249, "wPur:X", "🟣 TK REQUIRED · gated", color="purple")
+ ledger = {"wPur:X": [(1000.0, "TK-11666-x")]}
+ b = m.classify([s], ledger, {}, self.KNOWN)
+ self.assertEqual(len(b["contradicted"]), 0)
+ self.assertEqual(len(b["bound"]) + len(b["not_measured"]) + len(b["pending"]), 0)
+
+ def test_unknown_age_is_not_hidden_as_young(self):
+ # A tracked TK-REQUIRED pane with UNKNOWN age (age_s None — e.g. ps lstart
+ # unparseable) must be counted contradicted, NOT excused as pending. Hiding
+ # it would be the "measuring the wrong thing" false green.
+ s = self.green("ttys091", 4250, "wN:NOAGE", "🟢 TK REQUIRED · WORKING", age_s=None)
+ ledger = {"wN:NOAGE": [(1000.0, "TK-11666-x")]}
+ b = m.classify([s], ledger, {}, self.KNOWN)
+ self.assertEqual(len(b["contradicted"]), 1)
+ self.assertEqual(len(b["pending"]), 0)
+ self.assertEqual(m.verdict_of(b, True), "FAIL")
+
+ def test_proc_ages_resolve_for_a_live_pid(self):
+ # Guard the macOS lstart path: a real live pid must yield a positive age
+ # (the etimes bug returned {} → every false positive hidden as young).
+ import os
+ ages = m._proc_ages([os.getpid()])
+ self.assertIn(os.getpid(), ages)
+ self.assertGreaterEqual(ages[os.getpid()], 0.0)
+
+ def test_self_test_entrypoint_runs(self):
+ m._self_test() # raises on any regression
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tk_required_fp_metric.py b/tk_required_fp_metric.py
new file mode 100644
index 0000000..7f3dc5a
--- /dev/null
+++ b/tk_required_fp_metric.py
@@ -0,0 +1,375 @@
+#!/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:]))
← 4187d75 TK-11665: bind a session's ticket via ledger agent claude@TE
·
back to Terminal Status
·
terminal-status: retry transient EAGAIN in write_terminal so 1286106 →