← back to Terminal Status
Add shared session-bound terminal status engine and regression tests
94e4908eeac6f53aab5185949114d0fd70996081 · 2026-09-09 09:57:10 -0700 · Steve Abrams
Files touched
A integrations/dot-floor.shA integrations/dot.shA integrations/paint-lib.shA integrations/paint-tab.shA integrations/repaint-dot.shA integrations/semantic-dot.shA integrations/tab-state.shA integrations/working-state.shA terminal_status.pyA test_terminal_status.py
Diff
commit 94e4908eeac6f53aab5185949114d0fd70996081
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Wed Sep 9 09:57:10 2026 -0700
Add shared session-bound terminal status engine and regression tests
---
integrations/dot-floor.sh | 4 +
integrations/dot.sh | 12 ++
integrations/paint-lib.sh | 12 ++
integrations/paint-tab.sh | 10 +
integrations/repaint-dot.sh | 3 +
integrations/semantic-dot.sh | 7 +
integrations/tab-state.sh | 11 +
integrations/working-state.sh | 10 +
terminal_status.py | 465 ++++++++++++++++++++++++++++++++++++++++++
test_terminal_status.py | 165 +++++++++++++++
10 files changed, 699 insertions(+)
diff --git a/integrations/dot-floor.sh b/integrations/dot-floor.sh
new file mode 100644
index 0000000..1de2cbf
--- /dev/null
+++ b/integrations/dot-floor.sh
@@ -0,0 +1,4 @@
+#!/usr/bin/env bash
+# Stop hooks preserve deliberate status. Unknown is never guessed green.
+cat >/dev/null 2>&1 || true
+python3 "$HOME/Projects/terminal-status/terminal_status.py" repaint --quiet || true
diff --git a/integrations/dot.sh b/integrations/dot.sh
new file mode 100644
index 0000000..b83135a
--- /dev/null
+++ b/integrations/dot.sh
@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+# Shared compatibility entrypoint for the Claude and Codex color skills.
+set -eu
+runtime="${1:?runtime required}"
+color="${2:?color required}"
+shift 2
+engine="$HOME/Projects/terminal-status/terminal_status.py"
+case "${1:-}" in
+ --off) exec python3 "$engine" clear ;;
+ --all) exec python3 "$engine" set "$color" --all "$runtime" ;;
+ *) exec python3 "$engine" set "$color" "${1:-}" ;;
+esac
diff --git a/integrations/paint-lib.sh b/integrations/paint-lib.sh
new file mode 100644
index 0000000..22b60a6
--- /dev/null
+++ b/integrations/paint-lib.sh
@@ -0,0 +1,12 @@
+#!/usr/bin/env bash
+# Compatibility API. Semantic state and terminal writes share one engine.
+TABDOTS_DIR="$HOME/.claude/tab-dots"
+tabdots_is_subagent() { [ -n "${CLAUDE_CODE_CHILD_SESSION:-}" ]; }
+tabdots_resolve_tty() {
+ python3 "$HOME/Projects/terminal-status/terminal_status.py" current |
+ python3 -c 'import json,sys; print("/dev/"+json.load(sys.stdin)["tty"])'
+}
+tabdots_stop_pulse() { :; } # Current painters never create pulse processes.
+tabdots_paint() {
+ python3 "$HOME/Projects/terminal-status/terminal_status.py" paint-legacy "$@"
+}
diff --git a/integrations/paint-tab.sh b/integrations/paint-tab.sh
new file mode 100644
index 0000000..bcf0b11
--- /dev/null
+++ b/integrations/paint-tab.sh
@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+# Legacy one-shot painter routed through semantic status.
+set -eu
+engine="$HOME/Projects/terminal-status/terminal_status.py"
+case "${1:-waiting}" in
+ working|green) exec python3 "$engine" set green WORKING --quiet ;;
+ waiting|purple|solid-purple) exec python3 "$engine" set purple GATED --quiet ;;
+ stop) exec python3 "$engine" repaint --quiet ;;
+ *) echo "Unknown paint-tab state" >&2; exit 2 ;;
+esac
diff --git a/integrations/repaint-dot.sh b/integrations/repaint-dot.sh
new file mode 100644
index 0000000..e9b2cfe
--- /dev/null
+++ b/integrations/repaint-dot.sh
@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+# Rendering only: cannot overwrite a gate, refresh its age, or undo clear.
+exec python3 "$HOME/Projects/terminal-status/terminal_status.py" repaint --quiet
diff --git a/integrations/semantic-dot.sh b/integrations/semantic-dot.sh
new file mode 100644
index 0000000..0229210
--- /dev/null
+++ b/integrations/semantic-dot.sh
@@ -0,0 +1,7 @@
+#!/usr/bin/env bash
+# Existing Codex signature: state R G B emoji default-label [label|flag].
+set -eu
+color="${1:?state required}"
+default_label="${6:?default label required}"
+shift 6
+exec bash "$HOME/Projects/terminal-status/integrations/dot.sh" codex "$color" "${1:-$default_label}"
diff --git a/integrations/tab-state.sh b/integrations/tab-state.sh
new file mode 100644
index 0000000..7836290
--- /dev/null
+++ b/integrations/tab-state.sh
@@ -0,0 +1,11 @@
+#!/usr/bin/env bash
+# Legacy lifecycle entrypoint: one-shot status, no flashers.
+set -eu
+engine="$HOME/Projects/terminal-status/terminal_status.py"
+case "${1:-}" in
+ red) exec python3 "$engine" set pink "Closing session" --quiet ;;
+ flash-red) exec python3 "$engine" set green COMPACTING --quiet ;;
+ purple|flash-purple|waiting) exec python3 "$engine" set purple GATED --quiet ;;
+ green|flash-green|stop) exec python3 "$engine" repaint --quiet ;;
+ *) echo "Unknown tab-state mode" >&2; exit 2 ;;
+esac
diff --git a/integrations/working-state.sh b/integrations/working-state.sh
new file mode 100644
index 0000000..8369e45
--- /dev/null
+++ b/integrations/working-state.sh
@@ -0,0 +1,10 @@
+#!/usr/bin/env bash
+# One status update per user prompt; no heat colors or background title writers.
+set -eu
+engine="$HOME/Projects/terminal-status/terminal_status.py"
+case "${1:-}" in
+ on) exec python3 "$engine" set green WORKING --quiet ;;
+ off) exec python3 "$engine" repaint --quiet ;;
+ _watch|_idle_watch) exit 0 ;;
+ *) echo "usage: working-state.sh {on|off}" >&2; exit 2 ;;
+esac
diff --git a/terminal_status.py b/terminal_status.py
new file mode 100644
index 0000000..e8d53a7
--- /dev/null
+++ b/terminal_status.py
@@ -0,0 +1,465 @@
+#!/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
+
+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"),
+ "none": ("", (0, 0, 0), ""),
+}
+PRIORITY = {name: i for i, name in enumerate(
+ ("orange", "purple", "yellow", "green", "pink", "none"))}
+TTY = re.compile(r"ttys[0-9]+")
+VERSION = 1
+
+
+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)
+
+
+def processes():
+ result = subprocess.run(
+ ["ps", "-axo", "pid=,ppid=,tty=,lstart=,comm="],
+ capture_output=True, text=True, timeout=8,
+ env={**os.environ, "LC_ALL": "C"})
+ 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):
+ if os.environ.get("CLAUDE_CODE_CHILD_SESSION"):
+ raise StatusError("A subagent has no terminal of its own; refusing to paint")
+ 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()
+ raise StatusError("Headless or nested agent; refusing parent-terminal write")
+ raise StatusError("No owning Claude/Codex terminal in this process ancestry")
+
+
+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 = record["title"]
+ 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))
+ badge = base64.b64encode(title.encode()).decode()
+ text += f"\x1b]1;{title}\x07\x1b]2;{title}\x07\x1b]1337;SetBadgeFormat={badge}\x07"
+ return text.encode()
+
+
+def paint(owner, record):
+ 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")
+ payload = osc_payload(record)
+ # One write, no background loop, no terminal input injection.
+ if os.write(fd, payload) != len(payload):
+ raise StatusError("Incomplete terminal paint")
+ finally:
+ os.close(fd)
+
+
+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
+
+ 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):
+ if owners(self.process_provider()).get(owner.tty) != owner:
+ raise StatusError("Terminal owner changed; refusing stale writer")
+
+ @contextlib.contextmanager
+ def lock(self, owner):
+ 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:
+ deadline = time.monotonic() + 3
+ while True:
+ try:
+ fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ break
+ except BlockingIOError:
+ if time.monotonic() >= deadline:
+ raise StatusError("Terminal status is busy; retry")
+ time.sleep(0.02)
+ try:
+ self.assert_owner(owner)
+ 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"
+ expected = (COLORS[r["state"]][0] + " " + r["label"]).strip()
+ 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):
+ issued_ns = issued_ns if issued_ns is not None else time.monotonic_ns()
+ if color not in COLORS or not valid_label(label):
+ raise StatusError("Invalid status or label")
+ label = "" if color == "none" else (label or COLORS[color][2])
+ with self.lock(owner):
+ previous, _ = self.load(owner)
+ if previous and issued_ns <= previous["issued_ns"]:
+ raise StatusError("Obsolete status update; a newer event already won")
+ record = {
+ "version": VERSION, "owner": dataclasses.asdict(owner),
+ "state": color, "label": label,
+ "title": (COLORS[color][0] + " " + label).strip(),
+ "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)
+ atomic_write(self.path(owner), json.dumps(record, indent=2) + "\n")
+ error = None
+ try:
+ self.assert_owner(owner)
+ 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):
+ with self.lock(owner):
+ record, reason = self.read(owner)
+ if record is None:
+ raise StatusError("Cannot repaint unknown status: " + reason)
+ self.assert_owner(owner)
+ self.painter(owner, record)
+ # Never refresh timestamps, rewrite mirrors, or resurrect cleared state.
+ return record
+
+ def row(self, owner, title=""):
+ r, reason = self.read(owner, title)
+ result = {"tty": owner.tty, "pid": str(owner.pid),
+ "runtime": owner.runtime, "live": True,
+ "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="⚪ " + labels.get(reason, "Status not set"))
+ return result
+ result.update(color=r["state"], label=r["title"] or "⚪ Status cleared",
+ 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 iterm_sessions():
+ 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:
+ result = subprocess.run(["osascript", "-e", script], capture_output=True,
+ text=True, timeout=8)
+ if result.returncode:
+ return {}, "unavailable"
+ except (OSError, subprocess.TimeoutExpired):
+ return {}, "unavailable"
+ 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
+ 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)
+ return sorted(result, key=lambda r: (PRIORITY[r["color"]], r["tty"]))
+
+
+def main(argv=None):
+ parser = argparse.ArgumentParser(description=__doc__)
+ sub = parser.add_subparsers(dest="command", required=True)
+ for name in ("set", "clear", "repaint", "current", "start"):
+ cmd = sub.add_parser(name)
+ 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("--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")
+ args = parser.parse_args(argv)
+ store = Store()
+ rows = processes()
+ 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"])))
+ 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}')
+ return 0
+ caller = current_owner(rows)
+ if args.command == "current":
+ print(json.dumps(dataclasses.asdict(caller)))
+ return 0
+ targets = [caller]
+ 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.read(owner)
+ if record is not None:
+ record = store.repaint(owner)
+ elif reason in ("missing", "legacy_stale", "legacy_empty", "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:
+ record = store.set(owner, color, label)
+ 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)
diff --git a/test_terminal_status.py b/test_terminal_status.py
new file mode 100644
index 0000000..fe31f4d
--- /dev/null
+++ b/test_terminal_status.py
@@ -0,0 +1,165 @@
+import dataclasses
+import os
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import terminal_status as ts
+
+
+class StatusTests(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory()
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name)
+ self.process = ts.Process(123, 1, "ttys010",
+ "Wed Sep 9 08:35:08 2026", "/bin/codex")
+ self.rows = {123: self.process}
+ self.owner = self.process.owner()
+ self.paints = []
+ self.store = ts.Store(self.root, lambda: self.rows,
+ lambda owner, record: self.paints.append(record.copy()))
+
+ def legacy(self, runtime, title, stamp=None):
+ path = self.store.legacy[runtime] / (self.owner.tty + ".dot")
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(title)
+ when = stamp if stamp is not None else self.owner.epoch + 10
+ os.utime(path, (when, when))
+ return path
+
+ def test_all_colors_write_canonical_and_identical_mirrors(self):
+ for color in ts.COLORS:
+ r = self.store.set(self.owner, color, "" if color == "none" else "Task")
+ self.assertEqual(self.store.load(self.owner)[0], r)
+ for directory in self.store.legacy.values():
+ self.assertEqual((directory / "ttys010.dot").read_text(), r["title"])
+
+ def test_canonical_green_beats_stale_orange_title_and_cache(self):
+ self.store.set(self.owner, "green", "Working")
+ self.legacy("claude", "🟠 stale gate")
+ row = self.store.row(self.owner, "🟠 stale gate")
+ self.assertEqual(row["color"], "green")
+ self.assertIn("window_title_disagrees", row["warnings"])
+ self.assertIn("claude_mirror_disagrees", row["warnings"])
+
+ def test_repaint_never_changes_record_or_mirror_timestamps(self):
+ self.store.set(self.owner, "orange", "A real paste request")
+ paths = [self.store.path(self.owner)] + [
+ d / "ttys010.dot" for d in self.store.legacy.values()]
+ before = [(p.read_bytes(), p.stat().st_mtime_ns) for p in paths]
+ self.store.repaint(self.owner)
+ self.assertEqual(before, [(p.read_bytes(), p.stat().st_mtime_ns) for p in paths])
+
+ def test_clear_tombstone_prevents_title_cache_and_repaint_resurrection(self):
+ self.store.set(self.owner, "orange", "Old")
+ clear = self.store.set(self.owner, "none")
+ self.legacy("codex", "🟠 old copy returned")
+ self.assertEqual(self.store.row(self.owner, "🟠 old")["color"], "none")
+ self.store.repaint(self.owner)
+ self.assertEqual(self.paints[-1]["state"], "none")
+ self.assertEqual(self.store.load(self.owner)[0], clear)
+
+ def test_old_request_cannot_overwrite_a_newer_clear(self):
+ self.store.set(self.owner, "green", issued_ns=10)
+ clear = self.store.set(self.owner, "none", issued_ns=30)
+ with self.assertRaises(ts.StatusError):
+ self.store.set(self.owner, "orange", issued_ns=20)
+ self.assertEqual(self.store.load(self.owner)[0], clear)
+
+ def test_pid_reuse_and_runtime_change_fail_closed(self):
+ self.store.set(self.owner, "green")
+ 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")
+
+ def test_dead_writer_cannot_modify_new_terminal_owner(self):
+ self.store.set(self.owner, "green")
+ before = self.store.path(self.owner).read_bytes()
+ self.rows = {124: dataclasses.replace(self.process, pid=124)}
+ with self.assertRaises(ts.StatusError):
+ self.store.set(self.owner, "orange")
+ self.assertEqual(self.store.path(self.owner).read_bytes(), before)
+
+ def test_corrupt_record_never_falls_back_to_green(self):
+ self.store.set(self.owner, "green")
+ self.store.path(self.owner).write_text("{broken")
+ self.assertEqual(self.store.row(self.owner, "🟢 stale")["color"], "none")
+ with self.assertRaises(ts.StatusError):
+ self.store.repaint(self.owner)
+
+ def test_control_sequences_cannot_be_injected_through_labels(self):
+ for label in ("bad\nlabel", "\x1b]2;bad\x07", "a" * 513):
+ with self.assertRaises(ts.StatusError):
+ self.store.set(self.owner, "green", label)
+ self.assertFalse(self.store.path(self.owner).exists())
+
+ def test_legacy_is_runtime_scoped_and_stale_records_are_rejected(self):
+ self.legacy("claude", "🟠 other runtime")
+ self.assertIsNone(self.store.read(self.owner)[0])
+ self.legacy("codex", "🟢 previous session", self.owner.epoch - 10)
+ self.assertIsNone(self.store.read(self.owner)[0])
+ self.legacy("codex", "🟢 current")
+ self.assertEqual(self.store.read(self.owner)[0]["state"], "green")
+
+ def test_ambiguous_legacy_gate_never_guesses_green(self):
+ self.legacy("codex", "🟢 legacy")
+ row = self.store.row(self.owner, "🟠 pending paste")
+ self.assertEqual(row["color"], "none")
+ self.assertEqual(row["status_source"], "legacy_conflict")
+ self.assertFalse(self.store.path(self.owner).exists())
+
+ def test_scan_survives_terminal_api_outage_and_drops_dead_sessions(self):
+ self.store.set(self.owner, "purple", "Approval needed")
+ result = ts.scan(self.store, sessions={})
+ self.assertEqual(result[0]["color"], "purple")
+ self.assertFalse(result[0]["terminal_visible"])
+ self.rows = {}
+ self.assertEqual(ts.scan(self.store, sessions={"ttys010": "🟢 stale"}), [])
+
+ def test_runtime_identification_is_exact_not_command_substrings(self):
+ self.rows[456] = ts.Process(456, 1, "ttys011", self.owner.started,
+ "/usr/bin/not-codex")
+ self.rows[457] = ts.Process(457, 123, "ttys010", self.owner.started, "codex")
+ self.assertEqual(ts.owners(self.rows), {"ttys010": self.owner})
+
+ def test_headless_nested_agent_cannot_paint_parent(self):
+ self.rows[456] = ts.Process(456, 123, "??", self.owner.started, "codex")
+ self.rows[os.getpid()] = ts.Process(os.getpid(), 456, "??",
+ self.owner.started, "python3")
+ with self.assertRaises(ts.StatusError):
+ ts.current_owner(self.rows)
+ with patch.dict(os.environ, {"CLAUDE_CODE_CHILD_SESSION": "1"}):
+ with self.assertRaises(ts.StatusError):
+ ts.current_owner(self.rows)
+
+ def test_paint_failure_is_reported_and_auditable(self):
+ def fail(*args):
+ raise OSError("fixture: unavailable terminal")
+ self.store.painter = fail
+ with self.assertRaises(ts.StatusError):
+ self.store.set(self.owner, "green")
+ r, _ = self.store.load(self.owner)
+ self.assertEqual(r["render_status"], "failed")
+ self.assertIn("display_update_failed", self.store.row(self.owner)["warnings"])
+
+ def test_complete_osc_packet_contains_one_consistent_title_and_badge(self):
+ r = self.store.set(self.owner, "yellow", "Choose direction")
+ packet = ts.osc_payload(r)
+ self.assertIn(b"green;brightness;204", packet)
+ self.assertEqual(packet.count(r["title"].encode()), 2)
+ self.assertIn(b"SetBadgeFormat=", packet)
+
+ def test_two_ttys_do_not_share_status(self):
+ second = dataclasses.replace(self.process, pid=124, tty="ttys011")
+ self.rows[124] = second
+ self.store.set(self.owner, "green", "First")
+ self.store.set(second.owner(), "purple", "Second")
+ self.assertEqual(self.store.row(self.owner)["color"], "green")
+ self.assertEqual(self.store.row(second.owner())["color"], "purple")
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
← f49b9f9 Initial terminal status scaffold
·
back to Terminal Status
·
auto-data-snapshot: 2026-09-09T10:04:38 (1 data files) — ver 2e09c09 →