← back to Terminal Status

ticket_binding.py

448 lines

"""Read the existing ticket ledger; never create a second ticket database."""
import datetime as dt
import json
import os
import re
import subprocess
import tempfile
import time

# TK-11505: shared timeout constant so ticket_binding uses the same configurable
# ceiling as terminal_status._scan_processes(). Set TERMINAL_STATUS_PS_TIMEOUT
# to override (default 90s).
_PS_TIMEOUT = float(os.environ.get("TERMINAL_STATUS_PS_TIMEOUT", "90"))

# --- TK-11831: shared, pid-keyed cache for the two `ps` enrichment calls ------
# Both enrichment reads in discover() are IMMUTABLE for the life of a process:
# argv is fixed at exec, and `ps eww` reports the environment the process was
# started with. So a cache keyed on the pid AND its start time is EXACT rather
# than approximate, and including the start time makes it safe against pid
# reuse -- a recycled pid simply misses and is re-read. Only pids MISSING from
# the cache are ps'd, so a steady fleet pays ZERO ps calls per paint and a
# newly-spawned session ps's a one-entry pid list instead of all ~68.
#
# Why this matters (measured, TK-11831 / TK-11832): the pair cost 2.46s at 18
# live owners and ~4.4s under overnight load, and it ran on EVERY self-paint --
# every UserPromptSubmit and every Stop hook -- across ~68 concurrent sessions.
# That is the `ps` storm the watchtower attributed 245% aggregate CPU to, above
# both node (112%) and iTerm2 (57%) at the 09:46 peak.
#
# Why CACHE and not SKIP: TK-11835 removed these calls from the CROSS-TTY path
# via discover's argv={}/env={} skip-seam but deliberately left the SELF path
# alone, and that was right -- measured on this box, full enrichment binds 13
# tickets and the skip path binds 0, so skipping here would silently unbind
# every tab's ticket. Caching keeps the answer byte-identical and only stops us
# recomputing it ~68 times a turn.
#
# Every step FAILS OPEN: any cache fault degrades to the uncached behaviour,
# exactly as a slow ps already degrades the label but never the paint.
_PS_CACHE_PATH = os.environ.get(
    "TERMINAL_STATUS_PS_CACHE",
    os.path.join(tempfile.gettempdir(), "terminal-status-psenrich-%d.json" % os.getuid()))
# The TTL is a HYGIENE bound (stop a days-up box hoarding entries), NOT a
# correctness one: the key already pins identity, so a dead pid's entry is
# simply never looked up again. Note a steady all-hit pass writes nothing and
# therefore does NOT refresh `t` -- entries age out on when they were LEARNED.
# That is fine and deliberate: expiry costs one batched `ps` over the missing
# pids, not one per session.
_PS_CACHE_TTL = float(os.environ.get("TERMINAL_STATUS_PS_CACHE_TTL", "3600"))
_PS_CACHE_MAX = 512


def _enrich_key(owner):
    """pid + start time + runtime: a recycled pid (or a pid that exec'd into a
    different runtime) MISSES instead of serving the previous process's argv."""
    return "%d:%s:%s" % (owner.pid, owner.started, owner.runtime)


def _enrich_cache_read():
    """Best-effort read of the shared cache. Any fault reads as 'empty'."""
    try:
        with open(_PS_CACHE_PATH) as handle:
            blob = json.load(handle)
        entries = blob.get("entries") if isinstance(blob, dict) else None
        if not isinstance(entries, dict):
            return {}
        # Validate the VALUES, not just the container. A hand-edited or partly
        # written file with {"a": null} would otherwise be served straight into
        # the argv map, where the uncached path can only ever put a str -- a
        # cache hit that differs from the uncached answer, which is the one
        # thing this cache may never do. Anything unexpected is simply not a hit.
        floor = time.time() - _PS_CACHE_TTL
        clean = {}
        for key, value in entries.items():
            if not isinstance(value, dict):
                continue
            try:
                if float(value.get("t", 0)) <= floor:
                    continue
            except (TypeError, ValueError):
                continue
            kept = {f: value[f] for f in ("a", "s")
                    if isinstance(value.get(f), str)}
            if kept:
                kept["t"] = value["t"]
                clean[key] = kept
        return clean
    except (OSError, ValueError, TypeError, AttributeError):
        return {}


def _enrich_cache_merge(learned):
    """Publish `learned` without ever clobbering a peer's better answer.

    `os.replace` gives an untearable READ, but it does not serialise a
    read-modify-write across ~68 processes. Merging onto the snapshot this pass
    STARTED with would let a slow writer replace the whole file and silently
    drop -- or worse, blank -- an entry a peer had already learned. A lost key
    only costs an extra `ps`; a blanked one costs a WRONG ticket binding.

    Two properties make that impossible here without a lock. First, re-read the
    file NOW rather than reusing the start-of-pass snapshot. Second, never
    downgrade: a stored non-empty value is never replaced (values are immutable
    per process, so two writers can only ever agree) and an empty value never
    overwrites a non-empty one. Concurrent writes are therefore commutative --
    last-writer-wins is harmless because no writer can lose information.
    """
    entries = _enrich_cache_read()          # FRESH, not the start-of-pass snapshot
    now = time.time()
    for key, value in learned.items():
        entry = dict(entries.get(key, {}))
        for field in ("a", "s"):
            if field not in value:
                continue
            if value[field] == "" and entry.get(field):
                continue                    # never downgrade a peer's real answer
            entry[field] = value[field]
        entry["t"] = now
        entries[key] = entry
    if len(entries) > _PS_CACHE_MAX:  # newest-first; a long-lived box can't grow forever
        entries = dict(sorted(entries.items(),
                              key=lambda kv: -float(kv[1].get("t", 0)))[:_PS_CACHE_MAX])
    tmp = "%s.%d.tmp" % (_PS_CACHE_PATH, os.getpid())
    try:
        fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        with os.fdopen(fd, "w") as handle:  # 0600: argv snippets + TERM_SESSION_ID
            json.dump({"entries": entries}, handle)
        os.replace(tmp, _PS_CACHE_PATH)
    except (OSError, ValueError, TypeError):
        try:
            os.unlink(tmp)
        except OSError:
            pass

SHORT = re.compile(r"^TK-\d+(?=$|-)", re.I)
CID = re.compile(r"^(?:assign|create|action)-[a-z0-9]+-(\d+)-[a-z0-9]+$")

# TK-11631: the LAUNCHER-DECLARED ticket. run-ticket.sh bakes
# `export TK_AGENT=<AGENT_PREFIX>-<IDNUM>.` into the launch prompt, where IDNUM is
# ${ID#TK-} truncated at the first dash -- i.e. the DRIVING ticket's number alone,
# never a referenced one. That makes it unambiguous BY CONSTRUCTION, unlike the
# bare-id scan in discover(): the house slug convention embeds a referenced ticket
# inside the new ticket's own id ("TK-11630-tk-11340-follow-on-..."), so a session
# driving one of those has TWO TK ids in its argv, the len(ids)==1 guard abstains,
# and the tab reads "TK REQUIRED" for that session's entire life even while it logs
# real work on the board (54 of 1783 tickets, 3.0%, carry that shape). The prefixes
# are exactly the AGENT_PREFIX values run-ticket.sh can emit.
#
# TK-11631 follow-up -- the first cut matched a bare `TK_AGENT=` anywhere in argv and
# took re.search's FIRST hit, which is the self-match trap from the memory note
# `detector-argv-substring-self-match`: an agent whose prompt merely NAMES the thing
# it is debugging got bound to it. Measured on four real argv shapes: a prose mention
# ("fix the bug where TK_AGENT=claude-run-11630 fails to bind") bound 11630 with no
# export at all, and the argv of an agent working THIS ticket carried three matches
# and bound the right one only by accident of sentence order. So the match now needs
# BOTH halves, and either one alone is insufficient:
#   1. the literal `export ` prefix and the trailing `.` -- verified against the
#      launcher, not assumed: run-ticket.sh:60 is the first line of the PROMPT
#      heredoc and reads `export TK_AGENT=${AGENT_PREFIX}-${IDNUM}. You are driving`,
#      unconditionally, and 44 of 44 live sessions carry exactly that shape. `ps
#      -o args=` flattens argv into one space-joined string, so the prompt argument's
#      own start cannot be located reliably -- this literal IS the anchor, in place of
#      a positional hack that would only look rigorous.
#   2. exactly ONE DISTINCT declared id (findall, not search) -- mirroring the
#      len(ids)==1 rule this front-runs, so the new source can never be less
#      conservative than the rule it defers to. Two declarations abstain.
# If the launcher's wording ever changes, every binding falls back to "TK REQUIRED",
# which is the honest not-measured state -- never a confident wrong ticket.
AGENT = re.compile(r"\bexport TK_AGENT=(?:claude-run|codex-run|local-qwen-27b-run"
                   r"|local-qwen-14b-run|local-qwen-14b-mac1-run)-(\d+)\.", re.I)


def short(value):
    match = SHORT.match(str(value))
    return match.group().upper() if match else ""


def timestamp(value):
    try:
        return dt.datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
    except (ValueError, AttributeError):
        return 0


def discover(root, rows, live, chain, argv=None, env=None):
    """Attribute events only to a currently live main agent's own MCP process."""
    known, claims, actions = set(), {}, {}
    # TK-11665: the session's OWN ledger identity. When a session sets no TK_AGENT,
    # tk logs it under `claude@<TERM_SESSION_ID>` (the tk CLI's fallback identity --
    # see the agent line in ~/Projects/ticket-system/tk). session_events[tsid] holds
    # the MOST RECENT assign/create/action/comment that session wrote (never `read`,
    # which is board-viewing noise, nor `status`/`blocker`). Bound below to a live
    # session resolved PID->TERM_SESSION_ID, so a bare `claude`/`claude --continue`
    # that carries no ticket in its argv still shows what it is working on.
    session_events = {}
    descendants = {}
    for pid, p in rows.items():
        nearest = next((a for a in chain(pid, rows) if a.runtime), None)
        if nearest and live.get(nearest.tty) == nearest.owner():
            descendants[pid] = (nearest.owner(), timestamp_start(p.started))
    try:
        stream = (root / ".claude/tickets/events.jsonl").open()
    except OSError:
        return {}, known
    with stream:
        for line in stream:
            try:
                event = json.loads(line)
            except ValueError:
                continue
            ticket = short(event.get("id", ""))
            if not ticket:
                continue
            if event.get("type") == "create":
                known.add(ticket)
            # TK-11665: capture the session-identity ledger in this same pass (no
            # second file read). No epoch floor here -- unlike the correlation-id
            # PID match below, TERM_SESSION_ID is a STABLE session identity across a
            # --continue resume in the same pane, so events predating this process's
            # start ARE this session's own prior work and must be kept.
            agent = event.get("agent", "")
            if agent.startswith("claude@") and event.get("type") in (
                    "assign", "create", "action", "comment"):
                tsid = agent[len("claude@"):]
                at = timestamp(event.get("ts"))
                if tsid and at >= session_events.get(tsid, {}).get("at", 0):
                    session_events[tsid] = {"id": ticket, "at": at}
            match = CID.fullmatch(event.get("correlation_id", ""))
            if not match or int(match[1]) not in descendants:
                continue
            owner, born = descendants[int(match[1])]
            at = timestamp(event.get("ts"))
            if at < max(owner.epoch, born):
                continue
            target = claims if event.get("type") in ("assign", "create") else actions
            if at > target.get(owner.tty, {}).get("at", 0):
                target[owner.tty] = {"id": ticket, "at": at,
                                     "source": "ticket_ledger", "correlation_id": match[0]}
    result = {tty: claims.get(tty, actions.get(tty)) for tty in live}
    # TK-11831: read the shared ps-enrichment cache once for both blocks below.
    # Only consulted when a block would actually shell out (argv/env is None) --
    # the cross-tty skip-seam still short-circuits to zero work, as TK-11835 left it.
    _cache = _enrich_cache_read() if (argv is None or env is None) and live else {}
    _learned = {}
    # A unique ticket in the main process launch command is a conservative fallback.
    if argv is None and live:
        # TK-11369: was timeout=8 and FATAL. With ~49 live sessions this pid list
        # is large, and on a loaded box (53.78 today) it blew 8s -- which raised
        # and killed the ENTIRE repaint, so no dot was painted at all. That is
        # backwards: this block is an optional enrichment (see the comment above
        # -- "a conservative fallback" for the ticket LABEL). A slow ps must
        # degrade the label, never the paint. Raised to match the sibling
        # process-table read (_PS_TIMEOUT, now configurable via
        # TERMINAL_STATUS_PS_TIMEOUT, default 90s -- TK-11505), and made non-fatal.
        #
        # TK-11831: serve what the shared cache already knows and ps ONLY the
        # pids it is missing. On a steady fleet `missing` is empty and this
        # block makes no subprocess call at all.
        argv = {}
        missing = {}
        for owner in live.values():
            hit = _cache.get(_enrich_key(owner))
            if hit is not None and "a" in hit:
                argv[owner.pid] = hit["a"]
            else:
                missing[owner.pid] = owner
        try:
            if missing:
                output = subprocess.run(["ps", "-p", ",".join(str(pid) for pid in missing),
                                     "-o", "pid=,args="], capture_output=True, text=True,
                                    timeout=_PS_TIMEOUT)
                for line in output.stdout.splitlines():
                    parts = line.strip().split(None, 1)
                    if len(parts) == 2:
                        argv[int(parts[0])] = parts[1]
                # Cache only a CONCLUSIVE read. A pid absent from the output (it
                # exited mid-call, or ps was truncated) is "not yet known", NOT
                # "known to have no argv" -- storing "" for it would make one
                # transient miss STICKY for the whole TTL, so an unlucky session
                # would sit unbound for an hour. The uncached path retried on the
                # very next paint; the cache must not be worse than what it replaces.
                for pid, owner in missing.items():
                    if argv.get(pid):
                        _learned.setdefault(_enrich_key(owner), {})["a"] = argv[pid]
        except (OSError, subprocess.TimeoutExpired) as exc:
            # Label enrichment unavailable; the dot itself still paints.
            try:
                from terminal_status import _record_enum_blind
                _record_enum_blind("ticket_binding ps -p: " + type(exc).__name__)
            except Exception:
                pass
    # TK-11665: resolve each live session's TERM_SESSION_ID from its process ENV
    # (keyed on PID). `ps eww` prints ARGS AND ENV concatenated, so the env is the
    # tail after the clean args string -- reading TERM_SESSION_ID off the raw output
    # would hit the token sitting in a PROMPT argument (the measurement trap the
    # ticket calls out). We strip the args prefix (already fetched above via
    # `-o args=`) and parse only the remainder; if it does not start with the args
    # we skip that pid rather than risk the trap. Optional enrichment like the argv
    # read: a slow/absent ps degrades the label, never the paint.
    if env is None and live:
        #
        # TK-11831: same shared cache. The parsed TERM_SESSION_ID is stored, not
        # the raw `ps eww` line, so a cache hit never re-runs the prefix-strip
        # (and never re-exposes the raw env to the measurement trap above).
        env = {}
        missing = {}
        for owner in live.values():
            hit = _cache.get(_enrich_key(owner))
            if hit is not None and "s" in hit:
                if hit["s"]:
                    env[owner.pid] = hit["s"]
            else:
                missing[owner.pid] = owner
        try:
            if missing:
                output = subprocess.run(["ps", "eww", "-o", "pid=,command=",
                                     "-p", ",".join(str(pid) for pid in missing)],
                                    capture_output=True, text=True, timeout=_PS_TIMEOUT)
                # `settled` = pids whose env we could actually READ: the pid came
                # back AND its args prefix matched, so the remainder really is the
                # environment. A pid that is absent, or whose prefix did not match
                # (a stale argv), is INCONCLUSIVE -- we learn nothing about it and
                # must not cache a negative. A settled pid with no TERM_SESSION_ID
                # is a genuine answer and IS cached, so those pids stop re-ps'ing.
                settled = set()
                for line in output.stdout.splitlines():
                    parts = line.strip().split(None, 1)
                    if len(parts) != 2:
                        continue
                    pid, full = int(parts[0]), parts[1]
                    args = (argv or {}).get(pid, "")
                    if not (args and full.startswith(args)):
                        continue
                    settled.add(pid)
                    remainder = full[len(args):]
                    m = re.search(r"TERM_SESSION_ID=(\S+)", remainder)
                    if m:
                        env[pid] = m.group(1)
                for pid, owner in missing.items():
                    if pid in settled:
                        _learned.setdefault(_enrich_key(owner), {})["s"] = env.get(pid, "")
        except (OSError, subprocess.TimeoutExpired) as exc:
            try:
                from terminal_status import _record_enum_blind
                _record_enum_blind("ticket_binding ps eww: " + type(exc).__name__)
            except Exception:
                pass
    # TK-11831: publish whatever this pass had to learn. _enrich_cache_merge
    # re-reads the file and refuses to downgrade, so a concurrent peer can
    # neither be dropped nor blanked. Best-effort: if the write fails, the next
    # pass simply re-reads ps, which is exactly today's behaviour.
    if _learned:
        _enrich_cache_merge(_learned)
    for tty, owner in live.items():
        if result[tty]:
            continue
        command = (argv or {}).get(owner.pid, "")
        # The launcher's own declaration is tried FIRST: it names only the driving
        # ticket, so it survives the slug collision that makes the scan below
        # abstain. Same safety conditions as that scan -- exactly one distinct id,
        # and that id must be a ticket we have seen created -- so this can neither
        # invent a ticket nor pick one of several by position.
        declared = set(AGENT.findall(command))
        if len(declared) == 1 and "TK-" + next(iter(declared)) in known:
            result[tty] = {"id": "TK-" + next(iter(declared)), "at": owner.epoch,
                           "source": "main_process_agent"}
            continue
        ids = set(re.findall(r"\bTK-\d+\b", command, re.I))
        if len(ids) == 1 and next(iter(ids)).upper() in known:
            result[tty] = {"id": next(iter(ids)).upper(), "at": owner.epoch,
                           "source": "main_process_argument"}
        # TK-11665: LAST resort -- only when no MCP-correlation and no argv ticket
        # bound. The session's own ledger identity (PID->TERM_SESSION_ID) reflects
        # what it is working on NOW, which is the only signal that survives a
        # --continue resume (argv carries no ticket there). Same `in known` safety
        # as every other source: it can never invent a ticket. Cannot reach a
        # session that logs under a self-chosen TK_AGENT (the ticket's KNOWN CEILING);
        # that stays honest "TK REQUIRED".
        if not result[tty]:
            tsid = (env or {}).get(owner.pid)
            cand = session_events.get(tsid) if tsid else None
            if cand and cand["id"] in known:
                result[tty] = {"id": cand["id"], "at": cand["at"],
                               "source": "session_ledger"}
    return {tty: value for tty, value in result.items() if value}, known


def timestamp_start(value):
    return dt.datetime.strptime(value, "%a %b %d %H:%M:%S %Y").timestamp()


# --- TK-12168: auto-bind pure helpers ---------------------------------------
# A live top-level interactive session (no CLAUDE_CODE_CHILD_SESSION, never ran
# a `tk` command, no TK- in argv) sits unbound forever under discover() above --
# it is a purely READ-ONLY inference and by design never creates a ticket. The
# auto-bind orchestration in terminal_status.py (cmd_auto_bind) closes that gap
# from the UserPromptSubmit hook; these two detectors are the pure, side-effect
# free pieces of that decision so they are unit-testable without a live ledger,
# a Store, or a `tk` subprocess.

EXPLICIT_TICKET = re.compile(r"\bTK-(\d+)\b", re.I)


def explicit_ticket_in_prompt(prompt):
    """The one TK- id the user typed, or "" if none/ambiguous.

    Mirrors the "exactly one distinct id" conservatism used everywhere else in
    this module (AGENT / the bare-id scan in discover()): a prompt naming two
    different tickets abstains rather than guessing which one the user meant.
    """
    ids = {"TK-" + m for m in EXPLICIT_TICKET.findall(prompt or "")}
    return next(iter(ids)) if len(ids) == 1 else ""


# Bare acknowledgements that must never mint a ticket even though they clear
# the word-count floor on their own re-reading (e.g. "ok ok ok"). Kept short
# and literal -- anything not on this list falls through to the word-count
# test below, which is the primary guard.
_ACK_WORDS = {"ok", "okay", "yes", "y", "no", "n", "sure", "thanks", "thank you",
              "thx", "k", "kk", "yep", "yup", "nope", "continue", "go", "proceed",
              "done", "got it", "sounds good", "np", "cool", "great"}


def is_trivial_prompt(prompt):
    """True when a prompt carries no real task signal for auto-bind.

    Per TK-12168: no ticket for trivial prompts ("ok", "yes", slash-commands
    with no args, <3 words). Wait for a substantive one.
    """
    text = (prompt or "").strip()
    if not text:
        return True
    first_line = text.splitlines()[0].strip()
    if first_line.startswith("/") and " " not in first_line and "\t" not in first_line:
        return True  # a bare slash-command with no arguments
    words = text.split()
    if len(words) < 3:
        return True
    bare = re.sub(r"[^a-z ]", "", text.lower()).strip()
    if bare in _ACK_WORDS:
        return True
    return False


def short_topic(prompt, limit=70):
    """Collapse whitespace and cap length for a `tk new` title."""
    text = " ".join((prompt or "").split())
    return text[:limit]