← back to Terminal Status
departed_asks.py
227 lines
"""TK-12326: recover a DEPARTED session's outstanding asks when its tty is reused.
TK-11317 carries a prior owner's attention COLOUR + label onto the new owner's record
so a reused tty never silently blanks a needs-Steve dot. That carry is one-shot and
shallow: the moment the new owner sets its own colour the signal is gone, and it
never says WHICH asks were outstanding. This module is the durable other half:
* record() -- called by Store.settle_owner_change when it carries attention. Appends
one line per departure to <state>/departed-asks.jsonl (tty, prior
owner, colour, label, every TK id mentioned). Append-only, so a later
repaint by the new owner cannot erase it.
* resolve() -- joins each entry to the evidence of whether the ask is still open:
open memos in ~/.claude/yolo-queue/pending-approval/ (top level only;
_done/ etc. are filed) that name the ticket, and the ticket's status.
* ack() -- explicit human/agent dismissal of an entry.
Three states per entry, never two (CLAUDE.md TK-11431 amendment 1):
OUTSTANDING -- an open pending-approval memo NAMES the ticket (filename or header).
RESOLVED -- every ticket is done/archived and no open memo names or mentions it,
or the entry was acked.
UNVERIFIED -- nothing proves it either way (no ticket id, unknown ticket, or a ticket
still open with no memo -- a paste/question leaves no artifact). This is
NOT-MEASURED and is listed, never silently dropped.
"""
from __future__ import annotations
import datetime as dt
import fcntl
import json
import re
from pathlib import Path
TK_RE = re.compile(r"(?<![0-9A-Za-z])TK-(\d+)(?![0-9A-Za-z])", re.I)
DONE_STATES = frozenset(("done", "closed", "cancelled", "canceled", "wontfix"))
def mentions(*texts):
"""Every distinct short ticket id (TK-<digits>) mentioned in the given texts, in
first-seen order."""
seen = []
for text in texts:
for m in TK_RE.finditer(text or ""):
tk = "TK-" + m[1]
if tk not in seen:
seen.append(tk)
return seen
def _now():
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
def _read_lines(path):
try:
text = path.read_text()
except OSError:
return []
out = []
for line in text.splitlines():
try:
out.append(json.loads(line))
except ValueError:
continue
return out
def _append(path, entry):
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, "a") as fh:
fcntl.flock(fh, fcntl.LOCK_EX)
try:
fh.write(json.dumps(entry, ensure_ascii=False) + "\n")
finally:
fcntl.flock(fh, fcntl.LOCK_UN)
def entry_id(tty, prior_pid, prior_started):
return f"{tty}:{prior_pid}:{prior_started}"
def record(path, *, tty, runtime, prior, prior_info, new_pid, raw_texts=()):
"""Append one departure. Idempotent on (tty, prior owner): a second settle for the
same departed owner is a no-op. Returns the entry written, or None if skipped."""
info = prior_info or {}
eid = entry_id(tty, info.get("pid"), info.get("started"))
if info.get("pid") is None:
# legacy-.dot-only carry has no stored owner; key on the colour+label instead so
# the same stale mirror is not re-recorded on every new owner of this tty.
eid = f"{tty}:legacy:{prior['color']}:{prior.get('label', '')}"
for e in _read_lines(path):
if e.get("kind", "departed") == "departed" and e.get("id") == eid:
return None
tickets = mentions(prior.get("ticket", ""), prior.get("label", ""), *raw_texts)
entry = {"kind": "departed", "id": eid, "ts": _now(), "tty": tty,
"runtime": runtime, "prior_pid": info.get("pid"),
"prior_started": info.get("started"), "new_pid": new_pid,
"color": prior["color"], "label": prior.get("label", ""),
"tickets": tickets}
_append(path, entry)
return entry
def ack(path, eid, reason="", agent=""):
known = {e.get("id") for e in _read_lines(path) if e.get("kind", "departed") == "departed"}
if eid not in known:
raise KeyError(eid)
entry = {"kind": "ack", "id": eid, "ts": _now(), "reason": reason, "agent": agent}
_append(path, entry)
return entry
def open_memos(pending_dir):
"""{memo filename: text} for OPEN memos only -- top-level *.md. Sub-directories
(_done, _approved, _archive ...) are filed decisions, not open asks."""
out = {}
try:
paths = sorted(Path(pending_dir).glob("*.md"))
except OSError:
return out
for p in paths:
try:
out[p.name] = p.read_text(errors="replace")
except OSError:
out[p.name] = ""
return out
HEADER_LINES = 15
def memos_naming(ticket, memos):
"""(named, mentioned): memos whose FILENAME or header (first HEADER_LINES lines --
title / Ticket: line / frontmatter) names the ticket -- the memo IS that ask -- vs
memos that only mention it deeper in the body (a cross-reference). Only `named`
proves OUTSTANDING; `mentioned` still blocks RESOLVED (it is unexamined evidence)."""
pat = re.compile(r"(?<![0-9A-Za-z])" + re.escape(ticket) + r"(?![0-9])", re.I)
named, mentioned = [], []
for name, body in sorted(memos.items()):
head = "\n".join(body.splitlines()[:HEADER_LINES])
if pat.search(name) or pat.search(head):
named.append(name)
elif pat.search(body):
mentioned.append(name)
return named, mentioned
def ticket_statuses(tickets_dir, wanted):
"""Latest status per SHORT id (TK-<digits>) for the wanted set, from the canonical
event log. A ticket present only in an events-archive-done-* file is 'done'. A
ticket found nowhere is absent from the result (UNVERIFIED, never assumed done)."""
wanted = set(wanted)
if not wanted:
return {}
root = Path(tickets_dir)
short = re.compile(r"^(TK-\d+)(?:-|$)")
status = {}
def scan(path, archive):
try:
fh = open(path, errors="replace")
except OSError:
return
with fh:
for line in fh:
# cheap pre-filter before json.loads on a ~65MB log; spacing-agnostic
if "TK-" not in line or ('"create"' not in line and '"status"' not in line):
continue
try:
ev = json.loads(line)
except ValueError:
continue
m = short.match(ev.get("id", ""))
if not m or m[1] not in wanted:
continue
if ev.get("type") not in ("create", "status"):
continue
if archive:
status[m[1]] = "done"
elif ev.get("type") == "status" and ev.get("status"):
status[m[1]] = ev["status"]
else:
status.setdefault(m[1], "open")
for archive in sorted(root.glob("events-archive-done-*.jsonl")):
scan(archive, True)
scan(root / "events.jsonl", False)
return status
def resolve(path, pending_dir, tickets_dir, *, statuses=None, memos=None):
"""Every departed entry with its evidence and a three-way state. `statuses` /
`memos` are injection seams for tests (never passed by the CLI)."""
lines = _read_lines(path)
acks = {e["id"]: e for e in lines if e.get("kind") == "ack" and e.get("id")}
entries = [e for e in lines if e.get("kind", "departed") == "departed"]
wanted = {tk for e in entries for tk in e.get("tickets", [])}
if memos is None:
memos = open_memos(pending_dir)
if statuses is None:
statuses = ticket_statuses(tickets_dir, wanted)
out = []
for e in entries:
tks = e.get("tickets", [])
evidence = {}
for tk in tks:
named, mentioned = memos_naming(tk, memos)
evidence[tk] = {"status": statuses.get(tk, "unknown"),
"open_memos": named, "mentioned_in": mentioned}
if e["id"] in acks:
state = "RESOLVED"
elif any(v["open_memos"] for v in evidence.values()):
state = "OUTSTANDING"
elif tks and all(v["status"] in DONE_STATES and not v["mentioned_in"]
for v in evidence.values()):
state = "RESOLVED"
else:
state = "UNVERIFIED"
out.append(dict(e, state=state, evidence=evidence, acked=acks.get(e["id"])))
return out
def summary(resolved):
counts = {"OUTSTANDING": 0, "UNVERIFIED": 0, "RESOLVED": 0}
for r in resolved:
counts[r["state"]] += 1
return counts