← back to Terminal Status
TK-11317: never blank/green-float a prior needs-Steve colour across owner_changed
b594dbb907285ec9006c17c1981b380c12a16093 · 2026-09-26 08:48:59 -0700 · Steve
Steve's TK-11620 guard ruling (2026-09-26): when a live tty is reused by a new
claude/codex pid (e.g. after /res), Store.load() reports reason owner_changed
and the engine used to blindly floor the new record to green -- silently
dropping whatever purple/orange/yellow/lightblue attention the prior owner
left pending.
Add Store.prior_attention()/prior_owner_info() (read-only: inspect the raw
JSON record at this tty's path with the owner-identity check skipped, plus
the legacy .dot mirror at any mtime, pick the higher-PRIORITY ATTENTION
colour) and Store.settle_owner_change() (write: carry that colour forward
with carried_from provenance, or floor to yellow "New owner - status needed"
with owner_status_unverified only when nothing was pending). Both are new
optional fields on set()'s opt-in-per-call contract, so an ordinary explicit
set() still clears them.
Wire settle_owner_change into backfill()'s and start's owner_changed
branches (missing keeps its plain green floor -- there is no prior state to
preserve there). row() renders the would-be-settled state for the <20s
window before backfill runs, instead of a bare "status not set". The
`ticket` command now explicitly forwards carried_from/owner_status_unverified
so rebinding a ticket id never drops them.
29 new tests (test_owner_change_carry.py) + 4 new selftest cases. Negative
control: 27/29 new tests fail/error when run against the tk11317-preimage
tag (verification/TK-11317-integration/negative-control.txt). Full suite
green after the fix (verification/TK-11317-integration/after-tests.txt):
test_terminal_status.py (75), test_owner_change_carry.py (29), selftest
(18/18), test_stop_verdict.py (40), test_tk_required_fp_metric.py (13).
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UqLpisVAyfYM4zoVP7fFya
Files touched
M terminal_status.pyA test_owner_change_carry.pyM test_terminal_status.pyA verification/TK-11317-integration/after-tests.txtA verification/TK-11317-integration/before/audit-before.jsonA verification/TK-11317-integration/before/state-before.tgzA verification/TK-11317-integration/negative-control.txtA verification/TK-11317-integration/preimage-sha.txtA verification/TK-11317-integration/preimage.sha256A verification/TK-11317-integration/terminal_status.py.preimage
Diff
commit b594dbb907285ec9006c17c1981b380c12a16093
Author: Steve <steve@designerwallcoverings.com>
Date: Sat Sep 26 08:48:59 2026 -0700
TK-11317: never blank/green-float a prior needs-Steve colour across owner_changed
Steve's TK-11620 guard ruling (2026-09-26): when a live tty is reused by a new
claude/codex pid (e.g. after /res), Store.load() reports reason owner_changed
and the engine used to blindly floor the new record to green -- silently
dropping whatever purple/orange/yellow/lightblue attention the prior owner
left pending.
Add Store.prior_attention()/prior_owner_info() (read-only: inspect the raw
JSON record at this tty's path with the owner-identity check skipped, plus
the legacy .dot mirror at any mtime, pick the higher-PRIORITY ATTENTION
colour) and Store.settle_owner_change() (write: carry that colour forward
with carried_from provenance, or floor to yellow "New owner - status needed"
with owner_status_unverified only when nothing was pending). Both are new
optional fields on set()'s opt-in-per-call contract, so an ordinary explicit
set() still clears them.
Wire settle_owner_change into backfill()'s and start's owner_changed
branches (missing keeps its plain green floor -- there is no prior state to
preserve there). row() renders the would-be-settled state for the <20s
window before backfill runs, instead of a bare "status not set". The
`ticket` command now explicitly forwards carried_from/owner_status_unverified
so rebinding a ticket id never drops them.
29 new tests (test_owner_change_carry.py) + 4 new selftest cases. Negative
control: 27/29 new tests fail/error when run against the tk11317-preimage
tag (verification/TK-11317-integration/negative-control.txt). Full suite
green after the fix (verification/TK-11317-integration/after-tests.txt):
test_terminal_status.py (75), test_owner_change_carry.py (29), selftest
(18/18), test_stop_verdict.py (40), test_tk_required_fp_metric.py (13).
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UqLpisVAyfYM4zoVP7fFya
---
terminal_status.py | 249 ++-
test_owner_change_carry.py | 331 ++++
test_terminal_status.py | 9 +-
verification/TK-11317-integration/after-tests.txt | 206 +++
.../TK-11317-integration/before/audit-before.json | 1 +
.../TK-11317-integration/before/state-before.tgz | Bin 0 -> 72779 bytes
.../TK-11317-integration/negative-control.txt | 337 ++++
verification/TK-11317-integration/preimage-sha.txt | 1 +
verification/TK-11317-integration/preimage.sha256 | 1 +
.../terminal_status.py.preimage | 1911 ++++++++++++++++++++
10 files changed, 3019 insertions(+), 27 deletions(-)
diff --git a/terminal_status.py b/terminal_status.py
index 4dde4e2..84ee386 100644
--- a/terminal_status.py
+++ b/terminal_status.py
@@ -62,6 +62,18 @@ COLORS = {
PRIORITY = {name: i for i, name in enumerate(
("lightblue", "orange", "purple", "yellow", "green", "pink", "none"))}
+# TK-11317 (Steve's TK-11620 guard ruling, 2026-09-26): the colours that mean "this tab
+# is asking for Steve's attention" -- the set prior_attention() carries forward across an
+# owner_changed tty reuse so a reused tty can never silently blank or green-floor a prior
+# needs-Steve colour. Deliberately excludes green/pink/none (nothing to carry).
+ATTENTION = frozenset(("lightblue", "orange", "purple", "yellow"))
+
+# The floor settle_owner_change() applies when NEITHER the prior JSON record NOR the
+# legacy .dot mirror carried an ATTENTION colour: a brand-new owner's real state has not
+# been observed yet, so it is not a blind green either -- it is an explicit
+# "unverified" yellow that the session's own hooks refine on its next turn.
+OWNER_STATUS_LABEL = "New owner — status needed"
+
# 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
@@ -659,6 +671,38 @@ def selftest():
case("backfill green-floors the recordless live owner",
f == 1 and green_rec is not None and green_rec["state"] == "green")
+ # 5d/5e. TK-11317 (Steve's TK-11620 guard ruling): backfill's owner_changed
+ # branch must CARRY a prior needs-Steve colour forward across a tty reuse
+ # (never blank/green it), and floor to yellow "New owner - status needed"
+ # ONLY when the prior owner left nothing pending. `missing` (case 5c above)
+ # is unaffected -- it keeps the plain green floor, since there is no prior
+ # state on that tty to preserve.
+ with tempfile.TemporaryDirectory() as oc_dir:
+ old_purple = Process(6060, 1, "ttys093", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
+ new_on_purple_tty = Process(6161, 1, "ttys093", "Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
+ old_green = Process(6262, 1, "ttys094", "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
+ new_on_green_tty = Process(6363, 1, "ttys094", "Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
+ st2 = Store(user_root=oc_dir, process_provider=lambda **_: {},
+ painter=lambda owner, record: None)
+ st2.set(old_purple.owner(), "purple", "TK-11317 gated reason", rows={6060: old_purple})
+ st2.set(old_green.owner(), "green", rows={6262: old_green})
+ oc_rows = {6161: new_on_purple_tty, 6363: new_on_green_tty}
+ p2, f2, s2 = st2.backfill(oc_rows)
+ carried_rec, carried_reason = st2.load(new_on_purple_tty.owner())
+ needed_rec, needed_reason = st2.load(new_on_green_tty.owner())
+ case("backfill carries a prior purple record through owner_changed",
+ carried_reason == "canonical" and carried_rec is not None
+ and carried_rec["state"] == "purple"
+ and carried_rec.get("carried_from", {}).get("pid") == 6060
+ and carried_rec.get("ticket") == "TK-11317")
+ case("backfill floors owner_changed-from-green to yellow status-needed",
+ needed_reason == "canonical" and needed_rec is not None
+ and needed_rec["state"] == "yellow"
+ and needed_rec.get("owner_status_unverified") is True)
+ case("backfill owner_changed floors both new owners (f2 == 2)", f2 == 2)
+ case("backfill missing floor still has no owner_status_unverified flag",
+ green_rec is not None and "owner_status_unverified" not in green_rec)
+
# 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.
@@ -856,31 +900,131 @@ class Store:
finally:
fcntl.flock(lock, fcntl.LOCK_UN)
+ def _validate_record(self, r):
+ """Structural checks load() applies to a raw parsed record, MINUS the
+ owner-identity comparison (TK-11317: prior_attention() below reads a record
+ that fails owner identity BY CONSTRUCTION -- it belongs to a prior owner on a
+ reused tty -- so the identity check is kept as a separate, later step in
+ load() rather than folded in here). `carried_from` / `owner_status_unverified`
+ are optional fields (TK-11317) and intentionally unchecked: any value or
+ absence is accepted, so a record from before this change (neither field) and
+ one from after (either field) both validate identically."""
+ 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 False
+ 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 False
+ if r.get("variant", "") not in ("", "monitoring", "stopped", "waiting"):
+ return False
+ expected = status_title(r["state"], r["label"], ticket, r.get("variant", ""))
+ if r["title"] != expected or (r["state"] == "none" and r["label"]):
+ return False
+ return True
+
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)):
+ if not self._validate_record(r):
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 _read_raw_for_prior(self, owner):
+ """Read + structurally validate the record at this tty's path WITHOUT the
+ owner-identity check load() applies (TK-11317). Used ONLY to inspect what a
+ PRIOR owner left behind on a reused tty before deciding what to carry forward
+ -- never used to authorize a write; every write still goes through
+ set()/assert_owner(), which re-verifies the LIVE owner. Returns the raw dict
+ or None (missing / unreadable / structurally invalid)."""
+ path = self.path(owner)
+ try:
+ r = json.loads(path.read_text())
+ if not self._validate_record(r):
+ return None
+ return r
+ except (OSError, ValueError, KeyError, TypeError):
+ return None
+
+ def prior_attention(self, owner):
+ """TK-11317: what a PRIOR owner (before a tty-reuse owner_changed) left on
+ this tty, if it was an ATTENTION (needs-Steve) colour. Read-only; never
+ writes. Checks BOTH the canonical JSON record at this tty's path (owner
+ mismatch tolerated on purpose -- see _read_raw_for_prior) and the legacy
+ .dot mirror at ANY mtime: at the point this runs the NEW owner has not
+ painted anything yet, so anything sitting in either place can only be the
+ prior owner's. Returns {"color","label","ticket"} for whichever source is
+ HIGHER priority (PRIORITY: lower index wins), or None if neither carried
+ attention. The ticket is carried ONLY when the winning colour came from the
+ JSON record (it is the record that names the specific outstanding ask); a
+ legacy-.dot-only carry never invents a ticket id from its label text."""
+ candidates = []
+ r = self._read_raw_for_prior(owner)
+ if r is not None and r["state"] in ATTENTION:
+ label, _ = label_ticket(r["label"])
+ candidates.append((PRIORITY[r["state"]], r["state"], label, r.get("ticket", "")))
+ legacy_path = self.legacy[owner.runtime] / (owner.tty + ".dot")
+ try:
+ title = legacy_path.read_text().strip()
+ except OSError:
+ title = ""
+ if title and valid_label(title):
+ color = color_of(title)
+ if color in ATTENTION:
+ raw_label = title[len(COLORS[color][0]):].strip()
+ label, _ = label_ticket(raw_label)
+ candidates.append((PRIORITY[color], color, label, ""))
+ if not candidates:
+ return None
+ candidates.sort(key=lambda c: c[0])
+ _, color, label, ticket = candidates[0]
+ return {"color": color, "label": label, "ticket": ticket}
+
+ def prior_owner_info(self, owner):
+ """Best-effort identity of the prior owner whose record occupied this tty,
+ for the settled record's `carried_from` provenance. None when there is no
+ JSON record to read it from (e.g. a legacy-.dot-only carry has no stored
+ pid/started)."""
+ r = self._read_raw_for_prior(owner)
+ if r is not None and isinstance(r.get("owner"), dict):
+ return {"pid": r["owner"].get("pid"), "started": r["owner"].get("started")}
+ return None
+
+ def settle_owner_change(self, owner, *, rows=None, lock_wait=None):
+ """TK-11317 (Steve's TK-11620 guard ruling, 2026-09-26): a tty-reuse
+ owner_changed record must NEVER silently blank or green-floor a prior
+ needs-Steve colour. If the prior owner left attention pending (JSON record
+ or legacy .dot mirror), CARRY it forward onto the new owner's record with
+ provenance (`carried_from`: the prior owner's pid/started, when known).
+ Otherwise floor to yellow "New owner - status needed"
+ (`owner_status_unverified`) rather than a blind green -- a brand-new
+ session's real state has not been observed yet either, so green would be
+ just as unmeasured a claim as silence was."""
+ prior = self.prior_attention(owner)
+ info = self.prior_owner_info(owner)
+ if prior is not None:
+ label = prior["label"] or COLORS[prior["color"]][2]
+ try:
+ return self.set(owner, prior["color"], label,
+ ticket_update=prior["ticket"] or "",
+ rows=rows, lock_wait=lock_wait, carried_from=info)
+ except StatusError:
+ # An unknown/stale carried ticket id must never block restoring the
+ # carried ATTENTION colour itself -- drop the ticket, keep the colour.
+ return self.set(owner, prior["color"], label,
+ ticket_update="", rows=rows, lock_wait=lock_wait,
+ carried_from=info)
+ return self.set(owner, "yellow", OWNER_STATUS_LABEL,
+ rows=rows, lock_wait=lock_wait, owner_status_unverified=True)
+
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")
@@ -907,7 +1051,8 @@ class Store:
return r, reason
def set(self, owner, color, label="", *, issued_ns=None, ticket_update=None,
- variant="", deferential=False, rows=None, lock_wait=None):
+ variant="", deferential=False, rows=None, lock_wait=None,
+ carried_from=None, owner_status_unverified=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")
@@ -990,6 +1135,17 @@ class Store:
"updated_at": dt.datetime.now(dt.timezone.utc).isoformat(),
"render_status": "pending", "mirror_errors": [],
}
+ # TK-11317: both fields are OPT-IN per call, never inherited from
+ # `previous` implicitly -- an ordinary explicit set() (e.g. /greendot)
+ # that doesn't pass them builds a record without them, which is exactly
+ # how "only an explicit non-deferential set clears owner_status_unverified"
+ # is satisfied. A caller that means to PRESERVE either field across a
+ # rebuild (settle_owner_change's carry, the `ticket` command's rebind)
+ # must pass it through explicitly.
+ if carried_from is not None:
+ record["carried_from"] = carried_from
+ if owner_status_unverified:
+ record["owner_status_unverified"] = True
self.assert_owner(owner, rows)
atomic_write(self.path(owner), json.dumps(record, indent=2) + "\n")
error = None
@@ -1028,15 +1184,20 @@ class Store:
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.)
+ 2. RECORDLESS LIVE TAB (`missing` — a live owner with no record of its own at
+ all, e.g. a genuinely first-ever session) -> `set` a GREEN floor (working).
+ 3. OWNER-CHANGED LIVE TAB (`owner_changed` — 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) -> TK-11317
+ (Steve's TK-11620 guard ruling): a reused tty must NEVER silently blank or
+ green-floor a prior needs-Steve colour. `settle_owner_change` CARRIES a
+ prior ATTENTION colour (lightblue/orange/purple/yellow, from either the
+ JSON record or the legacy .dot mirror) forward onto the new owner's record
+ with provenance, or floors to yellow "New owner - status needed"
+ (`owner_status_unverified`) only when the prior owner left no attention
+ pending. Never a blind green floor for this reason — `missing` above is the
+ ONLY reason that still green-floors, because there IS no prior state there
+ to preserve.
Safe by construction: it iterates owners(rows) (the authoritative
process-table -> tty resolution), so it repaints/floors each owner's OWN tty
@@ -1056,9 +1217,12 @@ class Store:
if reason == "canonical":
self.repaint(owner, rows=rows, lock_wait=lock_wait)
painted += 1
- elif reason in ("missing", "owner_changed"):
+ elif reason == "missing":
self.set(owner, "green", rows=rows, lock_wait=lock_wait)
floored += 1
+ elif reason == "owner_changed":
+ self.settle_owner_change(owner, rows=rows, lock_wait=lock_wait)
+ floored += 1
else:
skipped += 1 # invalid / legacy_conflict -> manual refresh
except (StatusError, OSError):
@@ -1109,10 +1273,28 @@ class Store:
"ticket": ticket["id"], "ticket_source": ticket["source"],
"status_source": reason, "owner": dataclasses.asdict(owner)}
if r is None:
+ if reason == "owner_changed":
+ # TK-11317: this is the <20s window before backfill's next cycle
+ # actually settles the record -- render the WOULD-BE settled state
+ # (settle_owner_change's own logic, read-only here) instead of a bare
+ # ⚪ "status not set", so a carried needs-Steve colour is never even
+ # MOMENTARILY invisible.
+ prior = self.prior_attention(owner)
+ if prior is not None:
+ color = prior["color"]
+ label = prior["label"] or COLORS[color][2]
+ tk_id = prior["ticket"] or ticket["id"]
+ else:
+ color, label, tk_id = "yellow", OWNER_STATUS_LABEL, ticket["id"]
+ result.update(
+ color=color,
+ label=status_title(color, label, tk_id or "TK REQUIRED", "")
+ or "⚪ " + (tk_id or "TK REQUIRED") + " · Status cleared",
+ warnings=["owner_status_unverified"])
+ return result
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
@@ -1133,6 +1315,12 @@ class Store:
warnings.append(runtime + "_mirror_missing")
if r.get("render_status") != "applied":
warnings.append("display_update_" + r.get("render_status", "unknown"))
+ # TK-11317: a CANONICAL record can itself still be carrying the
+ # owner_status_unverified flag (settle_owner_change's yellow floor, not yet
+ # confirmed by an explicit set) -- surface that the same way any other
+ # warning is surfaced, regardless of `reason`.
+ if r.get("owner_status_unverified"):
+ warnings.append("owner_status_unverified")
result["warnings"] = warnings
return result
@@ -1816,9 +2004,15 @@ def main(argv=None):
if not ticket and args.ticket.upper() not in ("CLEAR", "NONE"):
raise StatusError("Expected a canonical TK number or clear")
previous, _ = store.read(caller)
+ # TK-11317: the ticket command only REBINDS the ticket id -- it must preserve
+ # a carried needs-Steve colour and the owner_status_unverified flag exactly
+ # as store.set()'s opt-in-per-call contract requires (they are never
+ # inherited implicitly from `previous`).
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 "")
+ ticket_update=ticket, variant=previous.get("variant", "") if previous else "",
+ carried_from=previous.get("carried_from") if previous else None,
+ owner_status_unverified=previous.get("owner_status_unverified") if previous else None)
store.header(caller)
print(f'/terminal-ticket → {caller.tty} {record["ticket"] or "TK REQUIRED"}')
return 0
@@ -1882,7 +2076,10 @@ def main(argv=None):
else:
raise StatusError("Ambiguous legacy state; set an explicit color")
elif reason == "owner_changed":
- record = store.set(owner, "green")
+ # TK-11317: never blindly green -- settle_owner_change carries a
+ # prior needs-Steve colour forward, or floors to yellow
+ # "New owner - status needed" only when nothing was pending.
+ record = store.settle_owner_change(owner)
else:
raise StatusError("Start found ambiguous status; set an explicit color")
elif args.command == "repaint":
diff --git a/test_owner_change_carry.py b/test_owner_change_carry.py
new file mode 100644
index 0000000..ecaaa28
--- /dev/null
+++ b/test_owner_change_carry.py
@@ -0,0 +1,331 @@
+"""TK-11317 (Steve's TK-11620 guard ruling, 2026-09-26): when a live terminal's
+stored record belongs to a PRIOR owner (tty reused by a new claude/codex pid, e.g.
+after /res), the engine must never silently blank or green-floor a prior
+needs-Steve colour. These tests exercise Store.prior_attention,
+Store.settle_owner_change, the backfill/start wiring, row() rendering, and the
+stickiness guarantees (deferential green, the `ticket` command, explicit set)
+against a tempdir Store with a fake in-memory painter -- never a real tty, never
+the shared ~/.local/state/abrams-terminal-status store.
+"""
+import dataclasses
+import tempfile
+import unittest
+from pathlib import Path
+
+import terminal_status as ts
+
+
+class OwnerChangeCarryTests(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name)
+ self.old_process = ts.Process(111, 1, "ttys050",
+ "Wed Sep 9 08:35:08 2026", "/usr/bin/claude")
+ self.new_process = ts.Process(222, 1, "ttys050",
+ "Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
+ self.old_owner = self.old_process.owner()
+ self.new_owner = self.new_process.owner()
+ self.paints = []
+ self.rows = {222: self.new_process}
+ self.store = ts.Store(self.root, lambda **_: self.rows,
+ lambda owner, record: self.paints.append(record.copy()))
+
+ def write_legacy(self, runtime, title):
+ path = self.store.legacy[runtime] / (self.old_owner.tty + ".dot")
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(title)
+ return path
+
+ def make_old_record(self, color, label="reason"):
+ """Write a canonical record for the OLD owner directly (no lock/assert_owner
+ contention with the new owner, since the fixture's process table only
+ contains the NEW owner -- exactly the real-world shape at the moment
+ owner_changed is first observed)."""
+ old_rows = {111: self.old_process}
+ old_store = ts.Store(self.root, lambda **_: old_rows,
+ lambda owner, record: None)
+ return old_store.set(self.old_owner, color, label, rows=old_rows)
+
+ # ---- prior_attention: JSON source ----------------------------------
+
+ def test_prior_purple_record_carried(self):
+ self.make_old_record("purple", "gated memo")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior, {"color": "purple", "label": "gated memo", "ticket": ""})
+
+ def test_prior_orange_record_carried(self):
+ self.make_old_record("orange", "paste waiting")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "orange")
+ self.assertEqual(prior["label"], "paste waiting")
+
+ def test_prior_yellow_record_carried(self):
+ self.make_old_record("yellow", "which fork?")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "yellow")
+ self.assertEqual(prior["label"], "which fork?")
+
+ def test_prior_lightblue_record_carried(self):
+ self.make_old_record("lightblue", "needs Steve")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "lightblue")
+ self.assertEqual(prior["label"], "needs Steve")
+
+ def test_prior_record_ticket_is_carried_from_json_source(self):
+ self.make_old_record("purple", "TK-11317 · gated memo")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["ticket"], "TK-11317")
+ self.assertEqual(prior["label"], "gated memo")
+
+ # ---- prior_attention: legacy .dot source ----------------------------
+
+ def test_legacy_dot_only_attention_carried_when_json_prior_is_green(self):
+ self.make_old_record("green")
+ self.write_legacy("claude", "\U0001F7E3 TK-99999 · legacy gated only")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "purple")
+ self.assertEqual(prior["label"], "legacy gated only")
+ # The ticket is NEVER invented from a legacy-.dot-only carry (rule: the
+ # carried ticket is kept ONLY when the carried colour came from the JSON
+ # record itself).
+ self.assertEqual(prior["ticket"], "")
+
+ def test_legacy_dot_only_attention_carried_when_no_json_record_at_all(self):
+ # No JSON record was ever written for the old owner -- only a legacy mirror
+ # exists (the pre-migration shape this rollout still has to tolerate).
+ self.write_legacy("claude", "\U0001F7E0 waiting on a paste")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "orange")
+ self.assertEqual(prior["label"], "waiting on a paste")
+ self.assertEqual(prior["ticket"], "")
+
+ # ---- priority arbitration between the two sources --------------------
+
+ def test_priority_choice_when_json_and_legacy_disagree(self):
+ # JSON = purple (PRIORITY 2), legacy .dot = orange (PRIORITY 1) -> orange
+ # outranks purple, so orange wins even though it is NOT the JSON source.
+ self.make_old_record("purple", "gated memo")
+ self.write_legacy("claude", "\U0001F7E0 paste waiting")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "orange")
+ self.assertEqual(prior["label"], "paste waiting")
+ self.assertEqual(prior["ticket"], "") # legacy source -> no invented ticket
+
+ def test_priority_choice_lightblue_outranks_everything(self):
+ self.make_old_record("yellow", "which fork?")
+ self.write_legacy("claude", "\U0001F535 needs Steve")
+ prior = self.store.prior_attention(self.new_owner)
+ self.assertEqual(prior["color"], "lightblue")
+
+ # ---- no attention pending -> yellow status-needed --------------------
+
+ def test_prior_green_yields_none_and_settles_to_yellow_status_needed(self):
+ self.make_old_record("green")
+ self.assertIsNone(self.store.prior_attention(self.new_owner))
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(record["state"], "yellow")
+ self.assertEqual(record["label"], ts.OWNER_STATUS_LABEL)
+ self.assertTrue(record["owner_status_unverified"])
+ self.assertNotIn("carried_from", record)
+
+ def test_prior_pink_yields_none_and_settles_to_yellow_status_needed(self):
+ self.make_old_record("pink")
+ self.assertIsNone(self.store.prior_attention(self.new_owner))
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(record["state"], "yellow")
+ self.assertTrue(record["owner_status_unverified"])
+
+ def test_no_prior_record_at_all_settles_to_yellow_status_needed(self):
+ # Reason "missing" on a fresh owner is handled separately (green floor) --
+ # this is prior_attention()/settle_owner_change() called directly on a tty
+ # with NOTHING behind it, which must still never fabricate an attention
+ # colour out of nothing.
+ self.assertIsNone(self.store.prior_attention(self.new_owner))
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(record["state"], "yellow")
+ self.assertTrue(record["owner_status_unverified"])
+
+ # ---- missing (no prior owner at all) keeps its plain green floor -----
+
+ def test_missing_reason_still_green_floors_via_backfill(self):
+ lone = ts.Process(333, 1, "ttys051", "Thu Sep 10 09:00:00 2026", "/usr/bin/claude")
+ rows = {333: lone}
+ store = ts.Store(self.root, lambda **_: rows, lambda owner, record: None)
+ painted, floored, skipped = store.backfill(rows)
+ record, reason = store.load(lone.owner())
+ self.assertEqual(reason, "canonical")
+ self.assertEqual(record["state"], "green")
+ self.assertNotIn("owner_status_unverified", record)
+ self.assertEqual((painted, floored, skipped), (0, 1, 0))
+
+ # ---- carried_from provenance -----------------------------------------
+
+ def test_carried_from_records_prior_owner_identity(self):
+ self.make_old_record("purple", "gated memo")
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(record["carried_from"],
+ {"pid": self.old_owner.pid, "started": self.old_owner.started})
+
+ def test_load_accepts_records_with_and_without_carried_from(self):
+ # load() must validate a record carrying the new optional fields exactly as
+ # readily as one that predates this change (neither field present).
+ self.store.set(self.new_owner, "green", rows=self.rows)
+ record, reason = self.store.load(self.new_owner)
+ self.assertEqual(reason, "canonical")
+ self.assertNotIn("carried_from", record)
+ self.make_old_record("purple", "gated memo")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ record2, reason2 = self.store.load(self.new_owner)
+ self.assertEqual(reason2, "canonical")
+ self.assertIn("carried_from", record2)
+ self.assertEqual(record2, settled)
+
+ # ---- row() rendering during the owner_changed transient window -------
+
+ def test_row_for_owner_changed_shows_carried_colour_not_none(self):
+ self.make_old_record("purple", "gated memo")
+ row = self.store.row(self.new_owner)
+ self.assertEqual(row["status_source"], "owner_changed")
+ self.assertEqual(row["color"], "purple")
+ self.assertIn("gated memo", row["label"])
+ self.assertIn("owner_status_unverified", row["warnings"])
+
+ def test_row_for_owner_changed_with_no_prior_attention_shows_yellow(self):
+ self.make_old_record("green")
+ row = self.store.row(self.new_owner)
+ self.assertEqual(row["status_source"], "owner_changed")
+ self.assertEqual(row["color"], "yellow")
+ self.assertIn(ts.OWNER_STATUS_LABEL, row["label"])
+ self.assertIn("owner_status_unverified", row["warnings"])
+
+ def test_row_never_renders_none_for_owner_changed(self):
+ for color in ("purple", "orange", "yellow", "lightblue", "green", "pink"):
+ with self.subTest(color=color):
+ temp = tempfile.TemporaryDirectory()
+ self.addCleanup(temp.cleanup)
+ root = Path(temp.name)
+ old_rows = {111: self.old_process}
+ old_store = ts.Store(root, lambda **_: old_rows, lambda o, r: None)
+ old_store.set(self.old_owner, color, "some reason", rows=old_rows)
+ store = ts.Store(root, lambda **_: self.rows, lambda o, r: None)
+ row = store.row(self.new_owner)
+ self.assertNotEqual(row["color"], "none")
+
+ def test_row_for_canonical_owner_status_unverified_record_carries_warning(self):
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(record["state"], "yellow")
+ row = self.store.row(self.new_owner)
+ self.assertEqual(row["status_source"], "canonical")
+ self.assertIn("owner_status_unverified", row["warnings"])
+
+ # ---- start command wiring ---------------------------------------------
+
+ def test_start_command_settles_owner_changed_via_carry(self):
+ self.make_old_record("orange", "paste waiting")
+ record, reason = self.store.load(self.new_owner)
+ self.assertIsNone(record)
+ self.assertEqual(reason, "owner_changed")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(settled["state"], "orange")
+
+ # ---- backfill wiring ----------------------------------------------------
+
+ def test_backfill_carries_purple_through_owner_changed(self):
+ self.make_old_record("purple", "gated memo")
+ painted, floored, skipped = self.store.backfill(self.rows)
+ self.assertEqual(floored, 1)
+ record, reason = self.store.load(self.new_owner)
+ self.assertEqual(reason, "canonical")
+ self.assertEqual(record["state"], "purple")
+ self.assertIn("carried_from", record)
+
+ def test_backfill_never_flattens_carried_attention_to_green(self):
+ self.make_old_record("lightblue", "needs Steve")
+ self.store.backfill(self.rows)
+ record, _ = self.store.load(self.new_owner)
+ self.assertNotEqual(record["state"], "green")
+ self.assertEqual(record["state"], "lightblue")
+
+ # ---- stickiness: deferential green preserves the carry -----------------
+
+ def test_deferential_green_keeps_carried_attention(self):
+ self.make_old_record("purple", "gated memo")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(settled["state"], "purple")
+ result = self.store.set(self.new_owner, "green", deferential=True, rows=self.rows)
+ self.assertEqual(result["state"], "purple")
+ self.assertEqual(result["carried_from"], settled["carried_from"])
+ self.assertEqual(result, settled)
+
+ def test_deferential_green_keeps_owner_status_unverified(self):
+ self.make_old_record("green")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertEqual(settled["state"], "yellow")
+ result = self.store.set(self.new_owner, "green", deferential=True, rows=self.rows)
+ self.assertEqual(result["state"], "yellow")
+ self.assertTrue(result["owner_status_unverified"])
+ self.assertEqual(result, settled)
+
+ # ---- stickiness: the `ticket` command's set() call shape ---------------
+
+ def test_ticket_binding_preserves_carried_attention_and_flag(self):
+ self.make_old_record("purple", "gated memo")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ previous, _ = self.store.read(self.new_owner)
+ # This mirrors EXACTLY the `ticket` command handler's store.set() call shape
+ # (terminal_status.py, `if args.command == "ticket":`).
+ record = self.store.set(
+ self.new_owner, previous["state"],
+ ts.label_ticket(previous["label"])[0],
+ ticket_update="TK-11317",
+ variant=previous.get("variant", ""),
+ carried_from=previous.get("carried_from"),
+ owner_status_unverified=previous.get("owner_status_unverified"),
+ rows=self.rows)
+ self.assertEqual(record["state"], "purple")
+ self.assertEqual(record["ticket"], "TK-11317")
+ self.assertEqual(record["carried_from"], settled["carried_from"])
+
+ def test_ticket_binding_preserves_owner_status_unverified(self):
+ self.make_old_record("green")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ previous, _ = self.store.read(self.new_owner)
+ record = self.store.set(
+ self.new_owner, previous["state"],
+ ts.label_ticket(previous["label"])[0],
+ ticket_update="TK-11317",
+ variant=previous.get("variant", ""),
+ carried_from=previous.get("carried_from"),
+ owner_status_unverified=previous.get("owner_status_unverified"),
+ rows=self.rows)
+ self.assertEqual(record["state"], "yellow")
+ self.assertTrue(record["owner_status_unverified"])
+ self.assertEqual(record["ticket"], "TK-11317")
+
+ # ---- explicit non-deferential set CLEARS the flag -----------------------
+
+ def test_explicit_set_green_clears_owner_status_unverified(self):
+ self.make_old_record("green")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertTrue(settled["owner_status_unverified"])
+ record = self.store.set(self.new_owner, "green", rows=self.rows)
+ self.assertNotIn("owner_status_unverified", record)
+
+ def test_explicit_set_green_clears_carried_from(self):
+ self.make_old_record("purple", "gated memo")
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ self.assertIn("carried_from", settled)
+ record = self.store.set(self.new_owner, "green", rows=self.rows)
+ self.assertNotIn("carried_from", record)
+
+ # ---- old owner cannot write (existing owner check, still true) ---------
+
+ def test_old_owner_cannot_write_after_tty_reuse(self):
+ self.make_old_record("purple", "gated memo")
+ with self.assertRaises(ts.StatusError):
+ self.store.set(self.old_owner, "orange", "stale write", rows=self.rows)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/test_terminal_status.py b/test_terminal_status.py
index d5e4fe6..c61dc6f 100644
--- a/test_terminal_status.py
+++ b/test_terminal_status.py
@@ -413,10 +413,17 @@ class StatusTests(unittest.TestCase):
def test_pid_reuse_and_runtime_change_fail_closed(self):
self.store.set(self.owner, "green")
+ # TK-11317: row() no longer renders a bare "none"/"status not set" for
+ # reason owner_changed -- it renders the would-be settle_owner_change()
+ # preview. The prior record here is GREEN (not an ATTENTION colour), so
+ # the preview is the yellow "New owner - status needed" floor, flagged
+ # owner_status_unverified -- never "none" and never a silent green carry.
for new in (dataclasses.replace(self.owner, started="Wed Sep 9 09:35:08 2026"),
dataclasses.replace(self.owner, runtime="claude"),
dataclasses.replace(self.owner, pid=124)):
- self.assertEqual(self.store.row(new, "🟢 stale")["color"], "none")
+ row = self.store.row(new, "🟢 stale")
+ self.assertEqual(row["color"], "yellow")
+ self.assertIn("owner_status_unverified", row["warnings"])
def test_dead_writer_cannot_modify_new_terminal_owner(self):
self.store.set(self.owner, "green")
diff --git a/verification/TK-11317-integration/after-tests.txt b/verification/TK-11317-integration/after-tests.txt
new file mode 100644
index 0000000..8fcd597
--- /dev/null
+++ b/verification/TK-11317-integration/after-tests.txt
@@ -0,0 +1,206 @@
+=== python3 test_terminal_status.py ===
+test_ambiguous_multiple_tickets_in_prompt_falls_through_to_create (__main__.AutoBindTests.test_ambiguous_multiple_tickets_in_prompt_falls_through_to_create) ... ok
+test_concurrent_prompt_on_same_tty_cannot_double_create (__main__.AutoBindTests.test_concurrent_prompt_on_same_tty_cannot_double_create) ... ok
+test_explicit_ticket_in_prompt_binds_without_creating (__main__.AutoBindTests.test_explicit_ticket_in_prompt_binds_without_creating) ... ok
+test_explicit_ticket_not_in_known_ledger_fails_open (__main__.AutoBindTests.test_explicit_ticket_not_in_known_ledger_fails_open) ... ok
+test_fails_open_when_tk_errors_and_rate_limits_the_retry (__main__.AutoBindTests.test_fails_open_when_tk_errors_and_rate_limits_the_retry) ... ok
+test_never_binds_an_empty_ticket_id (__main__.AutoBindTests.test_never_binds_an_empty_ticket_id) ... ok
+test_never_creates_a_duplicate_for_the_same_session (__main__.AutoBindTests.test_never_creates_a_duplicate_for_the_same_session) ... ok
+test_slash_command_with_arguments_is_substantive (__main__.AutoBindTests.test_slash_command_with_arguments_is_substantive) ... ok
+test_trivial_prompt_never_consumes_the_rate_limit_window (__main__.AutoBindTests.test_trivial_prompt_never_consumes_the_rate_limit_window) ... ok
+test_trivial_prompts_never_create_a_ticket (__main__.AutoBindTests.test_trivial_prompts_never_create_a_ticket) ... ok
+test_unbound_plus_substantive_prompt_creates_and_binds (__main__.AutoBindTests.test_unbound_plus_substantive_prompt_creates_and_binds) ... ok
+test_cache_key_is_pid_reuse_safe (__main__.EnrichmentCacheTests.test_cache_key_is_pid_reuse_safe) ... ok
+test_inconclusive_read_is_never_cached_as_a_negative (__main__.EnrichmentCacheTests.test_inconclusive_read_is_never_cached_as_a_negative) ... ok
+test_merge_never_downgrades_a_peer_value (__main__.EnrichmentCacheTests.test_merge_never_downgrades_a_peer_value) ... ok
+test_merge_reads_fresh_and_keeps_a_concurrent_peers_entry (__main__.EnrichmentCacheTests.test_merge_reads_fresh_and_keeps_a_concurrent_peers_entry) ... ok
+test_non_string_cache_values_are_not_served (__main__.EnrichmentCacheTests.test_non_string_cache_values_are_not_served) ... ok
+test_stale_entries_age_out (__main__.EnrichmentCacheTests.test_stale_entries_age_out) ... ok
+test_unreadable_cache_degrades_to_the_uncached_answer (__main__.EnrichmentCacheTests.test_unreadable_cache_degrades_to_the_uncached_answer) ... ok
+test_warm_cache_is_exact_and_shells_out_zero_times (__main__.EnrichmentCacheTests.test_warm_cache_is_exact_and_shells_out_zero_times) ... ok
+test_another_users_cache_file_is_never_trusted (__main__.ItermEnumCacheTests.test_another_users_cache_file_is_never_trusted) ... ok
+test_blind_falls_back_to_the_cached_map_not_green (__main__.ItermEnumCacheTests.test_blind_falls_back_to_the_cached_map_not_green) ... ok
+test_empty_successful_enum_is_not_cached_so_siblings_re_enumerate (__main__.ItermEnumCacheTests.test_empty_successful_enum_is_not_cached_so_siblings_re_enumerate) ... ok
+test_fresh_forces_a_live_enumeration_past_the_cache (__main__.ItermEnumCacheTests.test_fresh_forces_a_live_enumeration_past_the_cache) ... ok
+test_single_flight_peer_holding_lock_coalesces_and_fires_no_osascript (__main__.ItermEnumCacheTests.test_single_flight_peer_holding_lock_coalesces_and_fires_no_osascript) ... ok
+test_single_flight_peer_lock_no_fresh_map_blinds_without_osascript (__main__.ItermEnumCacheTests.test_single_flight_peer_lock_no_fresh_map_blinds_without_osascript) ... ok
+test_stale_map_past_the_bound_is_not_trusted (__main__.ItermEnumCacheTests.test_stale_map_past_the_bound_is_not_trusted) ... ok
+test_success_enumerates_once_then_siblings_hit_the_cache (__main__.ItermEnumCacheTests.test_success_enumerates_once_then_siblings_hit_the_cache) ... ok
+test_truly_blind_with_no_cache_records_enum_blind_and_is_unavailable (__main__.ItermEnumCacheTests.test_truly_blind_with_no_cache_records_enum_blind_and_is_unavailable) ... ok
+test_all_colors_write_canonical_and_identical_mirrors (__main__.StatusTests.test_all_colors_write_canonical_and_identical_mirrors) ... ok
+test_ambiguous_legacy_gate_never_guesses_green (__main__.StatusTests.test_ambiguous_legacy_gate_never_guesses_green) ... ok
+test_busy_target_skips_fast_with_short_lock_wait (__main__.StatusTests.test_busy_target_skips_fast_with_short_lock_wait) ... ok
+test_canonical_green_beats_stale_orange_title_and_cache (__main__.StatusTests.test_canonical_green_beats_stale_orange_title_and_cache) ... ok
+test_clear_tombstone_prevents_title_cache_and_repaint_resurrection (__main__.StatusTests.test_clear_tombstone_prevents_title_cache_and_repaint_resurrection) ... ok
+test_complete_osc_packet_contains_one_consistent_title_and_badge (__main__.StatusTests.test_complete_osc_packet_contains_one_consistent_title_and_badge) ... ok
+test_concurrent_writers_keep_the_newest_event_and_valid_json (__main__.StatusTests.test_concurrent_writers_keep_the_newest_event_and_valid_json) ... ok
+test_control_sequences_cannot_be_injected_through_labels (__main__.StatusTests.test_control_sequences_cannot_be_injected_through_labels) ... ok
+test_corrupt_record_never_falls_back_to_green (__main__.StatusTests.test_corrupt_record_never_falls_back_to_green) ... ok
+test_dead_writer_cannot_modify_new_terminal_owner (__main__.StatusTests.test_dead_writer_cannot_modify_new_terminal_owner) ... ok
+test_declared_agent_binds_driving_ticket_when_slug_embeds_a_second_id (__main__.StatusTests.test_declared_agent_binds_driving_ticket_when_slug_embeds_a_second_id) ... ok
+test_declared_agent_needs_an_export_and_exactly_one_declaration (__main__.StatusTests.test_declared_agent_needs_an_export_and_exactly_one_declaration) ... ok
+test_declared_agent_never_invents_a_ticket_nor_launders_a_bare_session (__main__.StatusTests.test_declared_agent_never_invents_a_ticket_nor_launders_a_bare_session) ... ok
+test_event_clock_is_comparable_between_python_processes (__main__.StatusTests.test_event_clock_is_comparable_between_python_processes) ... ok
+test_explicit_lightblue_reason_wins_over_preserved_base (__main__.StatusTests.test_explicit_lightblue_reason_wins_over_preserved_base) ... ok
+test_headless_nested_agent_cannot_paint_parent (__main__.StatusTests.test_headless_nested_agent_cannot_paint_parent) ... ok
+test_ledger_discovery_rejects_reused_pid_and_nested_agent (__main__.StatusTests.test_ledger_discovery_rejects_reused_pid_and_nested_agent) ... ok
+test_legacy_is_runtime_scoped_and_stale_records_are_rejected (__main__.StatusTests.test_legacy_is_runtime_scoped_and_stale_records_are_rejected) ... ok
+test_lightblue_is_solid_when_no_reason_to_preserve (__main__.StatusTests.test_lightblue_is_solid_when_no_reason_to_preserve) ... ok
+test_lightblue_never_mismatches_old_label_with_new_ticket (__main__.StatusTests.test_lightblue_never_mismatches_old_label_with_new_ticket) ... ok
+test_lightblue_preserves_underlying_needs_steve_reason (__main__.StatusTests.test_lightblue_preserves_underlying_needs_steve_reason) ... ok
+test_monitoring_shade_remains_green_and_survives_repaint (__main__.StatusTests.test_monitoring_shade_remains_green_and_survives_repaint) ... ok
+test_new_assignment_wins_and_generic_hook_preserves_binding (__main__.StatusTests.test_new_assignment_wins_and_generic_hook_preserves_binding) ... ok
+test_old_request_cannot_overwrite_a_newer_clear (__main__.StatusTests.test_old_request_cannot_overwrite_a_newer_clear) ... ok
+test_owner_for_paint_bridge_marker_never_rescues_an_intermediate_agent (__main__.StatusTests.test_owner_for_paint_bridge_marker_never_rescues_an_intermediate_agent) ... ok
+test_owner_for_paint_bridge_marker_never_rescues_undeterminable_ownership (__main__.StatusTests.test_owner_for_paint_bridge_marker_never_rescues_undeterminable_ownership) ... ok
+test_owner_for_paint_bridge_session_paints_when_it_owns_its_tty_directly (__main__.StatusTests.test_owner_for_paint_bridge_session_paints_when_it_owns_its_tty_directly) ... ok
+test_owner_for_paint_rail_paints_top_level_refuses_subagent (__main__.StatusTests.test_owner_for_paint_rail_paints_top_level_refuses_subagent) ... ok
+test_paint_failure_is_reported_and_auditable (__main__.StatusTests.test_paint_failure_is_reported_and_auditable) ... ok
+test_pid_reuse_and_runtime_change_fail_closed (__main__.StatusTests.test_pid_reuse_and_runtime_change_fail_closed) ... ok
+test_pulse_marker_toggles_across_legitimate_repaints (__main__.StatusTests.test_pulse_marker_toggles_across_legitimate_repaints) ... ok
+test_repaint_never_adopts_unbound_legacy_state (__main__.StatusTests.test_repaint_never_adopts_unbound_legacy_state) ... ok
+test_repaint_never_changes_record_or_mirror_timestamps (__main__.StatusTests.test_repaint_never_changes_record_or_mirror_timestamps) ... ok
+test_runtime_identification_is_exact_not_command_substrings (__main__.StatusTests.test_runtime_identification_is_exact_not_command_substrings) ... ok
+test_scan_elevates_stopped_tab_above_same_colour_and_higher_colours (__main__.StatusTests.test_scan_elevates_stopped_tab_above_same_colour_and_higher_colours) ... ok
+test_scan_survives_terminal_api_outage_and_drops_dead_sessions (__main__.StatusTests.test_scan_survives_terminal_api_outage_and_drops_dead_sessions) ... ok
+test_session_ledger_binds_resumed_session_via_term_session_id (__main__.StatusTests.test_session_ledger_binds_resumed_session_via_term_session_id) ... ok
+test_session_ledger_is_non_vacuous_and_never_launders (__main__.StatusTests.test_session_ledger_is_non_vacuous_and_never_launders) ... ok
+test_ticket_header_is_read_only_and_does_not_resurrect_clear (__main__.StatusTests.test_ticket_header_is_read_only_and_does_not_resurrect_clear) ... ok
+test_ticket_survives_every_generic_color_and_clear (__main__.StatusTests.test_ticket_survives_every_generic_color_and_clear) ... ok
+test_ticket_validation_and_tty_reuse (__main__.StatusTests.test_ticket_validation_and_tty_reuse) ... ok
+test_two_ttys_do_not_share_status (__main__.StatusTests.test_two_ttys_do_not_share_status) ... ok
+test_waiting_shade_is_brighter_green_and_survives_repaint (__main__.StatusTests.test_waiting_shade_is_brighter_green_and_survives_repaint) ... ok
+test_waiting_states_pulse_and_solid_states_stay_solid (__main__.StatusTests.test_waiting_states_pulse_and_solid_states_stay_solid) ... ok
+test_explicit_ticket_extraction (__main__.TicketPromptHelperTests.test_explicit_ticket_extraction) ... ok
+test_short_topic_collapses_and_caps (__main__.TicketPromptHelperTests.test_short_topic_collapses_and_caps) ... ok
+test_trivial_prompt_guard (__main__.TicketPromptHelperTests.test_trivial_prompt_guard) ... ok
+
+----------------------------------------------------------------------
+Ran 75 tests in 4.873s
+
+OK
+
+=== python3 test_owner_change_carry.py ===
+test_backfill_carries_purple_through_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_backfill_carries_purple_through_owner_changed) ... ok
+test_backfill_never_flattens_carried_attention_to_green (test_owner_change_carry.OwnerChangeCarryTests.test_backfill_never_flattens_carried_attention_to_green) ... ok
+test_carried_from_records_prior_owner_identity (test_owner_change_carry.OwnerChangeCarryTests.test_carried_from_records_prior_owner_identity) ... ok
+test_deferential_green_keeps_carried_attention (test_owner_change_carry.OwnerChangeCarryTests.test_deferential_green_keeps_carried_attention) ... ok
+test_deferential_green_keeps_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_deferential_green_keeps_owner_status_unverified) ... ok
+test_explicit_set_green_clears_carried_from (test_owner_change_carry.OwnerChangeCarryTests.test_explicit_set_green_clears_carried_from) ... ok
+test_explicit_set_green_clears_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_explicit_set_green_clears_owner_status_unverified) ... ok
+test_legacy_dot_only_attention_carried_when_json_prior_is_green (test_owner_change_carry.OwnerChangeCarryTests.test_legacy_dot_only_attention_carried_when_json_prior_is_green) ... ok
+test_legacy_dot_only_attention_carried_when_no_json_record_at_all (test_owner_change_carry.OwnerChangeCarryTests.test_legacy_dot_only_attention_carried_when_no_json_record_at_all) ... ok
+test_load_accepts_records_with_and_without_carried_from (test_owner_change_carry.OwnerChangeCarryTests.test_load_accepts_records_with_and_without_carried_from) ... ok
+test_missing_reason_still_green_floors_via_backfill (test_owner_change_carry.OwnerChangeCarryTests.test_missing_reason_still_green_floors_via_backfill) ... ok
+test_no_prior_record_at_all_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_no_prior_record_at_all_settles_to_yellow_status_needed) ... ok
+test_old_owner_cannot_write_after_tty_reuse (test_owner_change_carry.OwnerChangeCarryTests.test_old_owner_cannot_write_after_tty_reuse) ... ok
+test_prior_green_yields_none_and_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_prior_green_yields_none_and_settles_to_yellow_status_needed) ... ok
+test_prior_lightblue_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_lightblue_record_carried) ... ok
+test_prior_orange_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_orange_record_carried) ... ok
+test_prior_pink_yields_none_and_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_prior_pink_yields_none_and_settles_to_yellow_status_needed) ... ok
+test_prior_purple_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_purple_record_carried) ... ok
+test_prior_record_ticket_is_carried_from_json_source (test_owner_change_carry.OwnerChangeCarryTests.test_prior_record_ticket_is_carried_from_json_source) ... ok
+test_prior_yellow_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_yellow_record_carried) ... ok
+test_priority_choice_lightblue_outranks_everything (test_owner_change_carry.OwnerChangeCarryTests.test_priority_choice_lightblue_outranks_everything) ... ok
+test_priority_choice_when_json_and_legacy_disagree (test_owner_change_carry.OwnerChangeCarryTests.test_priority_choice_when_json_and_legacy_disagree) ... ok
+test_row_for_canonical_owner_status_unverified_record_carries_warning (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_canonical_owner_status_unverified_record_carries_warning) ... ok
+test_row_for_owner_changed_shows_carried_colour_not_none (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_owner_changed_shows_carried_colour_not_none) ... ok
+test_row_for_owner_changed_with_no_prior_attention_shows_yellow (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_owner_changed_with_no_prior_attention_shows_yellow) ... ok
+test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) ... ok
+test_start_command_settles_owner_changed_via_carry (test_owner_change_carry.OwnerChangeCarryTests.test_start_command_settles_owner_changed_via_carry) ... ok
+test_ticket_binding_preserves_carried_attention_and_flag (test_owner_change_carry.OwnerChangeCarryTests.test_ticket_binding_preserves_carried_attention_and_flag) ... ok
+test_ticket_binding_preserves_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_ticket_binding_preserves_owner_status_unverified) ... ok
+
+----------------------------------------------------------------------
+Ran 29 tests in 0.914s
+
+OK
+
+=== python3 terminal_status.py selftest ===
+ PASS top-level interactive session PAINTS
+ PASS subagent REFUSES
+ PASS session-hook --boot PAINTS
+ PASS --force opt-in PAINTS
+ PASS bridge session owning its tty directly PAINTS
+ PASS intermediate claude process REFUSES
+ PASS undeterminable ownership REFUSES
+ PASS backfill repaints the owner WITH a record
+ PASS backfill NEVER flattens a valid gated dot to green
+ PASS backfill green-floors the recordless live owner
+ PASS backfill carries a prior purple record through owner_changed
+ PASS backfill floors owner_changed-from-green to yellow status-needed
+ PASS backfill owner_changed floors both new owners (f2 == 2)
+ PASS backfill missing floor still has no owner_status_unverified flag
+ PASS heartbeat WARN + NOT-MEASURED when the ps scan failed
+ PASS heartbeat WARN when live tabs exist but NONE could be asserted
+ PASS heartbeat PASS when it acted on the live tabs
+ PASS heartbeat PASS on a measured 0-of-0 (no live tabs), still carrying the count
+selftest: PASS (18/18)
+
+=== python3 test_stop_verdict.py ===
+test_fixture_flag_with_base_override_never_touches_engine (__main__.CliTests.test_fixture_flag_with_base_override_never_touches_engine) ... ok
+test_injected_fault_goes_red (__main__.CliTests.test_injected_fault_goes_red) ... ok
+test_selftest_passes (__main__.CliTests.test_selftest_passes) ... ok
+test_stdin_missing_transcript_exits_zero (__main__.CliTests.test_stdin_missing_transcript_exits_zero) ... ok
+test_done_beats_silent_spinning_task (__main__.DecideTests.test_done_beats_silent_spinning_task) ... ok
+test_done_labels (__main__.DecideTests.test_done_labels) ... ok
+test_explicit_park_beats_waiting_on_steve (__main__.DecideTests.test_explicit_park_beats_waiting_on_steve) ... ok
+test_idle_pink_toggle (__main__.DecideTests.test_idle_pink_toggle) ... ok
+test_labels_respect_engine_valid_label (__main__.DecideTests.test_labels_respect_engine_valid_label) ... ok
+test_needs_steve_base_only_gets_stopped_variant (__main__.DecideTests.test_needs_steve_base_only_gets_stopped_variant) ... ok
+test_next_check_beats_done (__main__.DecideTests.test_next_check_beats_done) ... ok
+test_order_gated_beats_question (__main__.DecideTests.test_order_gated_beats_question) ... ok
+test_order_paste_beats_everything (__main__.DecideTests.test_order_paste_beats_everything) ... ok
+test_order_question_beats_waiting (__main__.DecideTests.test_order_question_beats_waiting) ... ok
+test_order_waiting_on_steve_beats_monitoring_and_done (__main__.DecideTests.test_order_waiting_on_steve_beats_monitoring_and_done) ... ok
+test_output_contract_keys (__main__.DecideTests.test_output_contract_keys) ... ok
+test_parked_base_only_repaints (__main__.DecideTests.test_parked_base_only_repaints) ... ok
+test_parked_beats_hold_and_done (__main__.DecideTests.test_parked_beats_hold_and_done) ... ok
+test_s3_keeps_base_label_when_no_time (__main__.DecideTests.test_s3_keeps_base_label_when_no_time) ... ok
+test_silent_spinning_task_alone_stays_green (__main__.DecideTests.test_silent_spinning_task_alone_stays_green) ... ok
+test_unknown_base_is_evaluated_like_green (__main__.DecideTests.test_unknown_base_is_evaluated_like_green) ... ok
+test_clean_view_strips_fences_code_tables_rules_emphasis (__main__.MatchTests.test_clean_view_strips_fences_code_tables_rules_emphasis) ... ok
+test_d1_captures_time (__main__.MatchTests.test_d1_captures_time) ... ok
+test_done_anywhere_in_closing_message (__main__.MatchTests.test_done_anywhere_in_closing_message) ... ok
+test_question_is_last_line_only (__main__.MatchTests.test_question_is_last_line_only) ... ok
+test_s2_time_is_the_next_check_not_the_done_time (__main__.MatchTests.test_s2_time_is_the_next_check_not_the_done_time) ... ok
+test_s3_comes_from_base_variant (__main__.MatchTests.test_s3_comes_from_base_variant) ... ok
+test_signal_table (__main__.MatchTests.test_signal_table) ... ok
+test_weekly_limit_matches_nothing (__main__.MatchTests.test_weekly_limit_matches_nothing) ... ok
+test_null_text_block_is_skipped (__main__.NullTextBlockTest.test_null_text_block_is_skipped) ... ok
+test_done_with_time_beats_parked_word (__main__.StdinFirstTests.test_done_with_time_beats_parked_word) ... ok
+test_live_background_task_keeps_green (__main__.StdinFirstTests.test_live_background_task_keeps_green) ... ok
+test_message_from_stdin_beats_missing_transcript (__main__.StdinFirstTests.test_message_from_stdin_beats_missing_transcript) ... ok
+test_no_message_returns_none (__main__.StdinFirstTests.test_no_message_returns_none) ... ok
+test_closing_text_is_after_last_user_line (__main__.TranscriptTests.test_closing_text_is_after_last_user_line) ... ok
+test_malformed_lines_are_skipped_and_no_text_is_none (__main__.TranscriptTests.test_malformed_lines_are_skipped_and_no_text_is_none) ... ok
+test_missing_transcript_is_benign (__main__.TranscriptTests.test_missing_transcript_is_benign) ... ok
+test_sidechain_and_noise_records_are_skipped (__main__.TranscriptTests.test_sidechain_and_noise_records_are_skipped) ... ok
+test_spinning_pairs_bg_start_with_notification (__main__.TranscriptTests.test_spinning_pairs_bg_start_with_notification) ... ok
+test_tail_lines_reads_only_the_end (__main__.TranscriptTests.test_tail_lines_reads_only_the_end) ... ok
+
+----------------------------------------------------------------------
+Ran 40 tests in 0.203s
+
+OK
+
+=== python3 test_tk_required_fp_metric.py ===
+test_argv_channel_also_contradicts (__main__.MetricTests.test_argv_channel_also_contradicts) ... ok
+test_clean_bound_session_passes (__main__.MetricTests.test_clean_bound_session_passes) ... ok
+test_contradicted_row_goes_red (__main__.MetricTests.test_contradicted_row_goes_red) ... ok
+test_empty_enumeration_is_warn_not_pass (__main__.MetricTests.test_empty_enumeration_is_warn_not_pass) ... ok
+test_launch_grace_defers_young_sessions_to_pending (__main__.MetricTests.test_launch_grace_defers_young_sessions_to_pending) ... ok
+test_ledger_ticket_must_be_known (__main__.MetricTests.test_ledger_ticket_must_be_known) ... ok
+test_not_measured_is_never_an_accusation_and_does_not_fail (__main__.MetricTests.test_not_measured_is_never_an_accusation_and_does_not_fail) ... ok
+test_only_green_sessions_are_scored (__main__.MetricTests.test_only_green_sessions_are_scored) ... ok
+test_proc_ages_resolve_for_a_live_pid (__main__.MetricTests.test_proc_ages_resolve_for_a_live_pid) ... ok
+test_read_events_are_noise_not_tracking (__main__.MetricTests.test_read_events_are_noise_not_tracking) ... ok
+test_self_test_entrypoint_runs (__main__.MetricTests.test_self_test_entrypoint_runs) ... ok
+test_unknown_age_is_not_hidden_as_young (__main__.MetricTests.test_unknown_age_is_not_hidden_as_young) ... ok
+test_unmeasured_input_is_warn_not_pass (__main__.MetricTests.test_unmeasured_input_is_warn_not_pass) ... ok
+
+----------------------------------------------------------------------
+Ran 13 tests in 0.008s
+
+OK
+self-test OK: contradicted→FAIL, clean→PASS, not_measured→PASS, young→pending
diff --git a/verification/TK-11317-integration/before/audit-before.json b/verification/TK-11317-integration/before/audit-before.json
new file mode 100644
index 0000000..63e3d85
--- /dev/null
+++ b/verification/TK-11317-integration/before/audit-before.json
@@ -0,0 +1 @@
+[{"tty": "ttys041", "pid": "35956", "runtime": "claude", "live": true, "ticket": "TK-12323", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys041", "pid": 35956, "runtime": "claude", "started": "Sat Sep 26 08:35:55 2026"}, "color": "orange", "label": "🟠🛑✧ TK-12323 · PASTE waiting", "variant": "stopped", "updated_at": "2026-09-26T15:38:06.326210+00:00", "revision": 4, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys000", "pid": "91884", "runtime": "claude", "live": true, "ticket": "TK-12172", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys000", "pid": 91884, "runtime": "claude", "started": "Sat Sep 26 07:43:02 2026"}, "color": "lightblue", "label": "🔵✦ TK-12172 · WAITING ON STEVE ·auto", "variant": "", "updated_at": "2026-09-26T15:38:29.356357+00:00", "revision": 113, "warnings": ["window_title_disagrees"], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys006", "pid": "9492", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys006", "pid": 9492, "runtime": "claude", "started": "Fri Sep 25 17:14:40 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T00:14:47.166210+00:00", "revision": 1, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys008", "pid": "10907", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys008", "pid": 10907, "runtime": "claude", "started": "Fri Sep 25 17:14:49 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T00:15:07.796593+00:00", "revision": 1, "warnings": [], "terminal_visible": false, "terminal_api": "cached"}, {"tty": "ttys010", "pid": "27829", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys010", "pid": 27829, "runtime": "claude", "started": "Sat Sep 26 07:54:29 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T14:54:46.413740+00:00", "revision": 1, "warnings": [], "terminal_visible": false, "terminal_api": "cached"}, {"tty": "ttys011", "pid": "66349", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys011", "pid": 66349, "runtime": "claude", "started": "Sat Sep 26 08:30:05 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T15:30:33.944353+00:00", "revision": 1, "warnings": [], "terminal_visible": false, "terminal_api": "cached"}, {"tty": "ttys021", "pid": "82558", "runtime": "claude", "live": true, "ticket": "TK-12164", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys021", "pid": 82558, "runtime": "claude", "started": "Sat Sep 26 08:17:14 2026"}, "color": "green", "label": "🟢 TK-12164 · WORKING", "variant": "", "updated_at": "2026-09-26T15:37:28.851322+00:00", "revision": 25, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys022", "pid": "68394", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys022", "pid": 68394, "runtime": "claude", "started": "Sat Sep 26 07:49:41 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T14:49:48.023514+00:00", "revision": 1, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys027", "pid": "48941", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys027", "pid": 48941, "runtime": "claude", "started": "Sat Sep 26 07:55:54 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T14:56:13.035115+00:00", "revision": 1, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys030", "pid": "46322", "runtime": "claude", "live": true, "ticket": "", "ticket_source": "unbound", "status_source": "canonical", "owner": {"tty": "ttys030", "pid": 46322, "runtime": "claude", "started": "Sat Sep 26 08:04:52 2026"}, "color": "green", "label": "🟢 TK REQUIRED · WORKING", "variant": "", "updated_at": "2026-09-26T15:05:13.512782+00:00", "revision": 1, "warnings": [], "terminal_visible": false, "terminal_api": "cached"}, {"tty": "ttys032", "pid": "92520", "runtime": "claude", "live": true, "ticket": "TK-12013", "ticket_source": "main_process_argument", "status_source": "canonical", "owner": {"tty": "ttys032", "pid": 92520, "runtime": "claude", "started": "Sat Sep 26 08:18:03 2026"}, "color": "green", "label": "🟢 TK-12013 · WORKING", "variant": "", "updated_at": "2026-09-26T15:34:15.564343+00:00", "revision": 13, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys034", "pid": "98027", "runtime": "claude", "live": true, "ticket": "TK-11860", "ticket_source": "main_process_argument", "status_source": "canonical", "owner": {"tty": "ttys034", "pid": 98027, "runtime": "claude", "started": "Sat Sep 26 08:18:18 2026"}, "color": "green", "label": "🟢 TK-11860 · WORKING", "variant": "", "updated_at": "2026-09-26T15:21:35.657823+00:00", "revision": 2, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys035", "pid": "25474", "runtime": "claude", "live": true, "ticket": "TK-11865", "ticket_source": "session_ledger", "status_source": "canonical", "owner": {"tty": "ttys035", "pid": 25474, "runtime": "claude", "started": "Sat Sep 26 08:19:45 2026"}, "color": "green", "label": "🟢 TK-11865 · WORKING", "variant": "", "updated_at": "2026-09-26T15:38:31.285752+00:00", "revision": 10, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys036", "pid": "39377", "runtime": "claude", "live": true, "ticket": "TK-12206", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys036", "pid": 39377, "runtime": "claude", "started": "Sat Sep 26 08:20:43 2026"}, "color": "green", "label": "🟢 TK-12206 · WORKING", "variant": "", "updated_at": "2026-09-26T15:34:14.815715+00:00", "revision": 17, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys039", "pid": "35954", "runtime": "claude", "live": true, "ticket": "TK-12322", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys039", "pid": 35954, "runtime": "claude", "started": "Sat Sep 26 08:35:55 2026"}, "color": "green", "label": "🟢 TK-12322 · WORKING", "variant": "", "updated_at": "2026-09-26T15:36:07.920219+00:00", "revision": 2, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys040", "pid": "35955", "runtime": "claude", "live": true, "ticket": "TK-12321", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys040", "pid": 35955, "runtime": "claude", "started": "Sat Sep 26 08:35:55 2026"}, "color": "green", "label": "🟢 TK-12321 · WORKING", "variant": "", "updated_at": "2026-09-26T15:36:07.487952+00:00", "revision": 2, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys042", "pid": "63126", "runtime": "claude", "live": true, "ticket": "TK-12324", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys042", "pid": 63126, "runtime": "claude", "started": "Sat Sep 26 08:37:51 2026"}, "color": "green", "label": "🟢 TK-12324 · WORKING", "variant": "", "updated_at": "2026-09-26T15:37:58.126901+00:00", "revision": 2, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys004", "pid": "6208", "runtime": "claude", "live": true, "ticket": "TK-12251", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys004", "pid": 6208, "runtime": "claude", "started": "Fri Sep 25 17:14:16 2026"}, "color": "pink", "label": "🩷 TK-12251 · PARKED", "variant": "", "updated_at": "2026-09-26T14:39:27.971023+00:00", "revision": 5, "warnings": [], "terminal_visible": false, "terminal_api": "cached"}, {"tty": "ttys033", "pid": "93125", "runtime": "claude", "live": true, "ticket": "TK-12306", "ticket_source": "explicit", "status_source": "canonical", "owner": {"tty": "ttys033", "pid": 93125, "runtime": "claude", "started": "Sat Sep 26 08:32:14 2026"}, "color": "pink", "label": "🩷 TK-12306 · PARKED", "variant": "", "updated_at": "2026-09-26T15:34:04.063119+00:00", "revision": 3, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}, {"tty": "ttys037", "pid": "58536", "runtime": "claude", "live": true, "ticket": "TK-12308", "ticket_source": "main_process_argument", "status_source": "canonical", "owner": {"tty": "ttys037", "pid": 58536, "runtime": "claude", "started": "Sat Sep 26 08:21:59 2026"}, "color": "pink", "label": "🩷 TK-12308 · PARKED", "variant": "", "updated_at": "2026-09-26T15:30:46.550215+00:00", "revision": 12, "warnings": [], "terminal_visible": true, "terminal_api": "cached"}]
diff --git a/verification/TK-11317-integration/before/state-before.tgz b/verification/TK-11317-integration/before/state-before.tgz
new file mode 100644
index 0000000..c770af7
Binary files /dev/null and b/verification/TK-11317-integration/before/state-before.tgz differ
diff --git a/verification/TK-11317-integration/negative-control.txt b/verification/TK-11317-integration/negative-control.txt
new file mode 100644
index 0000000..d089b52
--- /dev/null
+++ b/verification/TK-11317-integration/negative-control.txt
@@ -0,0 +1,337 @@
+test_backfill_carries_purple_through_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_backfill_carries_purple_through_owner_changed) ... FAIL
+test_backfill_never_flattens_carried_attention_to_green (test_owner_change_carry.OwnerChangeCarryTests.test_backfill_never_flattens_carried_attention_to_green) ... FAIL
+test_carried_from_records_prior_owner_identity (test_owner_change_carry.OwnerChangeCarryTests.test_carried_from_records_prior_owner_identity) ... ERROR
+test_deferential_green_keeps_carried_attention (test_owner_change_carry.OwnerChangeCarryTests.test_deferential_green_keeps_carried_attention) ... ERROR
+test_deferential_green_keeps_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_deferential_green_keeps_owner_status_unverified) ... ERROR
+test_explicit_set_green_clears_carried_from (test_owner_change_carry.OwnerChangeCarryTests.test_explicit_set_green_clears_carried_from) ... ERROR
+test_explicit_set_green_clears_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_explicit_set_green_clears_owner_status_unverified) ... ERROR
+test_legacy_dot_only_attention_carried_when_json_prior_is_green (test_owner_change_carry.OwnerChangeCarryTests.test_legacy_dot_only_attention_carried_when_json_prior_is_green) ... ERROR
+test_legacy_dot_only_attention_carried_when_no_json_record_at_all (test_owner_change_carry.OwnerChangeCarryTests.test_legacy_dot_only_attention_carried_when_no_json_record_at_all) ... ERROR
+test_load_accepts_records_with_and_without_carried_from (test_owner_change_carry.OwnerChangeCarryTests.test_load_accepts_records_with_and_without_carried_from) ... ERROR
+test_missing_reason_still_green_floors_via_backfill (test_owner_change_carry.OwnerChangeCarryTests.test_missing_reason_still_green_floors_via_backfill) ... ok
+test_no_prior_record_at_all_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_no_prior_record_at_all_settles_to_yellow_status_needed) ... ERROR
+test_old_owner_cannot_write_after_tty_reuse (test_owner_change_carry.OwnerChangeCarryTests.test_old_owner_cannot_write_after_tty_reuse) ... ok
+test_prior_green_yields_none_and_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_prior_green_yields_none_and_settles_to_yellow_status_needed) ... ERROR
+test_prior_lightblue_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_lightblue_record_carried) ... ERROR
+test_prior_orange_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_orange_record_carried) ... ERROR
+test_prior_pink_yields_none_and_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_prior_pink_yields_none_and_settles_to_yellow_status_needed) ... ERROR
+test_prior_purple_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_purple_record_carried) ... ERROR
+test_prior_record_ticket_is_carried_from_json_source (test_owner_change_carry.OwnerChangeCarryTests.test_prior_record_ticket_is_carried_from_json_source) ... ERROR
+test_prior_yellow_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_yellow_record_carried) ... ERROR
+test_priority_choice_lightblue_outranks_everything (test_owner_change_carry.OwnerChangeCarryTests.test_priority_choice_lightblue_outranks_everything) ... ERROR
+test_priority_choice_when_json_and_legacy_disagree (test_owner_change_carry.OwnerChangeCarryTests.test_priority_choice_when_json_and_legacy_disagree) ... ERROR
+test_row_for_canonical_owner_status_unverified_record_carries_warning (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_canonical_owner_status_unverified_record_carries_warning) ... ERROR
+test_row_for_owner_changed_shows_carried_colour_not_none (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_owner_changed_shows_carried_colour_not_none) ... FAIL
+test_row_for_owner_changed_with_no_prior_attention_shows_yellow (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_owner_changed_with_no_prior_attention_shows_yellow) ... FAIL
+test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) ...
+ test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='purple') ... FAIL
+ test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='orange') ... FAIL
+ test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='yellow') ... FAIL
+ test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='lightblue') ... FAIL
+ test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='green') ... FAIL
+ test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='pink') ... FAIL
+test_start_command_settles_owner_changed_via_carry (test_owner_change_carry.OwnerChangeCarryTests.test_start_command_settles_owner_changed_via_carry) ... ERROR
+test_ticket_binding_preserves_carried_attention_and_flag (test_owner_change_carry.OwnerChangeCarryTests.test_ticket_binding_preserves_carried_attention_and_flag) ... ERROR
+test_ticket_binding_preserves_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_ticket_binding_preserves_owner_status_unverified) ... ERROR
+
+======================================================================
+ERROR: test_carried_from_records_prior_owner_identity (test_owner_change_carry.OwnerChangeCarryTests.test_carried_from_records_prior_owner_identity)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 166, in test_carried_from_records_prior_owner_identity
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_deferential_green_keeps_carried_attention (test_owner_change_carry.OwnerChangeCarryTests.test_deferential_green_keeps_carried_attention)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 254, in test_deferential_green_keeps_carried_attention
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_deferential_green_keeps_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_deferential_green_keeps_owner_status_unverified)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 263, in test_deferential_green_keeps_owner_status_unverified
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_explicit_set_green_clears_carried_from (test_owner_change_carry.OwnerChangeCarryTests.test_explicit_set_green_clears_carried_from)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 317, in test_explicit_set_green_clears_carried_from
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_explicit_set_green_clears_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_explicit_set_green_clears_owner_status_unverified)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 310, in test_explicit_set_green_clears_owner_status_unverified
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_legacy_dot_only_attention_carried_when_json_prior_is_green (test_owner_change_carry.OwnerChangeCarryTests.test_legacy_dot_only_attention_carried_when_json_prior_is_green)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 86, in test_legacy_dot_only_attention_carried_when_json_prior_is_green
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_legacy_dot_only_attention_carried_when_no_json_record_at_all (test_owner_change_carry.OwnerChangeCarryTests.test_legacy_dot_only_attention_carried_when_no_json_record_at_all)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 98, in test_legacy_dot_only_attention_carried_when_no_json_record_at_all
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_load_accepts_records_with_and_without_carried_from (test_owner_change_carry.OwnerChangeCarryTests.test_load_accepts_records_with_and_without_carried_from)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 178, in test_load_accepts_records_with_and_without_carried_from
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_no_prior_record_at_all_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_no_prior_record_at_all_settles_to_yellow_status_needed)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 144, in test_no_prior_record_at_all_settles_to_yellow_status_needed
+ self.assertIsNone(self.store.prior_attention(self.new_owner))
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_green_yields_none_and_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_prior_green_yields_none_and_settles_to_yellow_status_needed)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 125, in test_prior_green_yields_none_and_settles_to_yellow_status_needed
+ self.assertIsNone(self.store.prior_attention(self.new_owner))
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_lightblue_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_lightblue_record_carried)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 71, in test_prior_lightblue_record_carried
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_orange_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_orange_record_carried)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 59, in test_prior_orange_record_carried
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_pink_yields_none_and_settles_to_yellow_status_needed (test_owner_change_carry.OwnerChangeCarryTests.test_prior_pink_yields_none_and_settles_to_yellow_status_needed)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 134, in test_prior_pink_yields_none_and_settles_to_yellow_status_needed
+ self.assertIsNone(self.store.prior_attention(self.new_owner))
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_purple_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_purple_record_carried)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 54, in test_prior_purple_record_carried
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_record_ticket_is_carried_from_json_source (test_owner_change_carry.OwnerChangeCarryTests.test_prior_record_ticket_is_carried_from_json_source)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 77, in test_prior_record_ticket_is_carried_from_json_source
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_prior_yellow_record_carried (test_owner_change_carry.OwnerChangeCarryTests.test_prior_yellow_record_carried)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 65, in test_prior_yellow_record_carried
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_priority_choice_lightblue_outranks_everything (test_owner_change_carry.OwnerChangeCarryTests.test_priority_choice_lightblue_outranks_everything)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 118, in test_priority_choice_lightblue_outranks_everything
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_priority_choice_when_json_and_legacy_disagree (test_owner_change_carry.OwnerChangeCarryTests.test_priority_choice_when_json_and_legacy_disagree)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 110, in test_priority_choice_when_json_and_legacy_disagree
+ prior = self.store.prior_attention(self.new_owner)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'prior_attention'
+
+======================================================================
+ERROR: test_row_for_canonical_owner_status_unverified_record_carries_warning (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_canonical_owner_status_unverified_record_carries_warning)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 216, in test_row_for_canonical_owner_status_unverified_record_carries_warning
+ record = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_start_command_settles_owner_changed_via_carry (test_owner_change_carry.OwnerChangeCarryTests.test_start_command_settles_owner_changed_via_carry)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 229, in test_start_command_settles_owner_changed_via_carry
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_ticket_binding_preserves_carried_attention_and_flag (test_owner_change_carry.OwnerChangeCarryTests.test_ticket_binding_preserves_carried_attention_and_flag)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 274, in test_ticket_binding_preserves_carried_attention_and_flag
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+ERROR: test_ticket_binding_preserves_owner_status_unverified (test_owner_change_carry.OwnerChangeCarryTests.test_ticket_binding_preserves_owner_status_unverified)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 292, in test_ticket_binding_preserves_owner_status_unverified
+ settled = self.store.settle_owner_change(self.new_owner, rows=self.rows)
+ ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AttributeError: 'Store' object has no attribute 'settle_owner_change'
+
+======================================================================
+FAIL: test_backfill_carries_purple_through_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_backfill_carries_purple_through_owner_changed)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 240, in test_backfill_carries_purple_through_owner_changed
+ self.assertEqual(record["state"], "purple")
+ ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'green' != 'purple'
+- green
++ purple
+
+
+======================================================================
+FAIL: test_backfill_never_flattens_carried_attention_to_green (test_owner_change_carry.OwnerChangeCarryTests.test_backfill_never_flattens_carried_attention_to_green)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 247, in test_backfill_never_flattens_carried_attention_to_green
+ self.assertNotEqual(record["state"], "green")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'green' == 'green'
+
+======================================================================
+FAIL: test_row_for_owner_changed_shows_carried_colour_not_none (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_owner_changed_shows_carried_colour_not_none)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 190, in test_row_for_owner_changed_shows_carried_colour_not_none
+ self.assertEqual(row["color"], "purple")
+ ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' != 'purple'
+- none
++ purple
+
+
+======================================================================
+FAIL: test_row_for_owner_changed_with_no_prior_attention_shows_yellow (test_owner_change_carry.OwnerChangeCarryTests.test_row_for_owner_changed_with_no_prior_attention_shows_yellow)
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 198, in test_row_for_owner_changed_with_no_prior_attention_shows_yellow
+ self.assertEqual(row["color"], "yellow")
+ ~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' != 'yellow'
+- none
++ yellow
+
+
+======================================================================
+FAIL: test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='purple')
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 213, in test_row_never_renders_none_for_owner_changed
+ self.assertNotEqual(row["color"], "none")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' == 'none'
+
+======================================================================
+FAIL: test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='orange')
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 213, in test_row_never_renders_none_for_owner_changed
+ self.assertNotEqual(row["color"], "none")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' == 'none'
+
+======================================================================
+FAIL: test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='yellow')
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 213, in test_row_never_renders_none_for_owner_changed
+ self.assertNotEqual(row["color"], "none")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' == 'none'
+
+======================================================================
+FAIL: test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='lightblue')
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 213, in test_row_never_renders_none_for_owner_changed
+ self.assertNotEqual(row["color"], "none")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' == 'none'
+
+======================================================================
+FAIL: test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='green')
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 213, in test_row_never_renders_none_for_owner_changed
+ self.assertNotEqual(row["color"], "none")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' == 'none'
+
+======================================================================
+FAIL: test_row_never_renders_none_for_owner_changed (test_owner_change_carry.OwnerChangeCarryTests.test_row_never_renders_none_for_owner_changed) (color='pink')
+----------------------------------------------------------------------
+Traceback (most recent call last):
+ File "/private/tmp/claude-501/-Users-macstudio3/81614591-265b-48ed-97b7-6518070fa586/scratchpad/preimage-check/test_owner_change_carry.py", line 213, in test_row_never_renders_none_for_owner_changed
+ self.assertNotEqual(row["color"], "none")
+ ~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^
+AssertionError: 'none' == 'none'
+
+----------------------------------------------------------------------
+Ran 29 tests in 1.425s
+
+FAILED (failures=10, errors=22)
diff --git a/verification/TK-11317-integration/preimage-sha.txt b/verification/TK-11317-integration/preimage-sha.txt
new file mode 100644
index 0000000..5604921
--- /dev/null
+++ b/verification/TK-11317-integration/preimage-sha.txt
@@ -0,0 +1 @@
+6d687763a62a1f3a7ba45e7f80f0436c38863cab
diff --git a/verification/TK-11317-integration/preimage.sha256 b/verification/TK-11317-integration/preimage.sha256
new file mode 100644
index 0000000..d16ae7e
--- /dev/null
+++ b/verification/TK-11317-integration/preimage.sha256
@@ -0,0 +1 @@
+fbd616f144e47d83cad582821bcfa469ef2b3905ba0c9940321de3362016a203 terminal_status.py
diff --git a/verification/TK-11317-integration/terminal_status.py.preimage b/verification/TK-11317-integration/terminal_status.py.preimage
new file mode 100644
index 0000000..4dde4e2
--- /dev/null
+++ b/verification/TK-11317-integration/terminal_status.py.preimage
@@ -0,0 +1,1911 @@
+#!/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)
← 6d68776 Auto-bind a ticket for unbound sessions from the UserPromptS
·
back to Terminal Status
·
TK-11317: live acceptance evidence (Claude+Codex readback, s 76859a5 →