← back to Answer Cockpit

lib/queue.js

274 lines

'use strict';
// queue.js — scan → needs-Steve items, urgency sort, 4s cache, tty allowlist.
//
//   scan()        python3 ~/Projects/terminal-status/terminal_status.py scan --json
//                 fallback: bash ~/.claude/skills/allcolordots/allcolordots.sh --json
//                 4s cache. Returns {rows, stale, terminalApi, source, error, scannedAt}
//   freshLive()   UNCACHED scan → Map tty→row  (POST guards use this, never the cache)
//   build(opts)   filter needs-Steve colors {lightblue,orange,purple,yellow}, live only,
//                 sort variant=="stopped" first then PRIORITY, attach rgb/emoji from
//                 COLORS (terminal_status.py:48-61), merge memos by TK, resolve detail.
const path = require('path');
const crypto = require('crypto');
const { execFile } = require('child_process');
const transcript = require('./transcript');
const memo = require('./memo');
const audit = require('./audit');
const writeback = require('./writeback');
const ticketlane = require('./ticketlane');

const HOME = process.env.HOME || require('os').homedir();
const TS_PY = path.join(HOME, 'Projects/terminal-status/terminal_status.py');
const ALL_SH = path.join(HOME, '.claude/skills/allcolordots/allcolordots.sh');
// 2026-09-16: at 59 live sessions the dot scanner takes ~90s (iTerm2 itself at 150% CPU), so a
// 45s budget made it fail → allcolordots fallback → stale → WARN. Budget it for reality and let
// the last good scan serve requests while a fresh one runs in the background (never a fake PASS:
// stale:true is still surfaced when the LAST scan was the fallback or older than STALE_AFTER_MS).
const CACHE_MS = 15000;
const SCAN_TIMEOUT_MS = 150000;
const STALE_AFTER_MS = 5 * 60 * 1000;

// copied from terminal_status.py:48-63
const COLORS = {
  green: { emoji: '🟢', rgb: [0, 200, 83], meaning: 'WORKING' },
  yellow: { emoji: '🟡', rgb: [255, 204, 0], meaning: 'DIRECTION?' },
  purple: { emoji: '🟣', rgb: [148, 0, 211], meaning: 'GATED' },
  orange: { emoji: '🟠', rgb: [255, 140, 0], meaning: 'PASTE waiting' },
  pink: { emoji: '🩷', rgb: [255, 105, 180], meaning: 'PARKED' },
  lightblue: { emoji: '🔵', rgb: [0, 176, 240], meaning: 'NEEDS STEVE' },
  none: { emoji: '', rgb: [0, 0, 0], meaning: '' },
};
const PRIORITY = { lightblue: 0, orange: 1, purple: 2, yellow: 3, green: 4, pink: 5, none: 6 };
const NEEDS_STEVE = new Set(['lightblue', 'orange', 'purple', 'yellow']);
const TTY_RE = /^ttys\d{3}$/;

let cached = null; // {ts, result}
let inflight = null;
// BOOT_ID changes on every server start (= every code change). The client reloads itself when it
// sees a new one, so a browser tab left open across a fix can never keep typing with STALE JS —
// the likeliest way to reproduce the already-fixed "label typing selects option 1" bug.
const BOOT_ID = `${process.pid}-${Date.now().toString(36)}`;

function run(cmd, args, timeout) {
  return new Promise((resolve) => {
    execFile(cmd, args, { timeout, maxBuffer: 8 << 20, env: { ...process.env, LANG: 'en_US.UTF-8', LC_ALL: 'en_US.UTF-8' } }, (err, stdout, stderr) => {
      if (err) return resolve({ ok: false, err: (stderr || err.message).trim().slice(0, 300), stdout: stdout || '' });
      resolve({ ok: true, stdout: stdout || '', err: null });
    });
  });
}

function normalizeAllcolordotsRow(r) {
  return {
    tty: r.tty, pid: r.pid || '', runtime: 'unknown', live: !!r.live, ticket: (String(r.label || '').match(/TK-\d{2,6}/) || [])[0] || '',
    owner: { tty: r.tty, pid: r.pid ? parseInt(r.pid, 10) : null, runtime: 'unknown', started: null },
    color: r.color, label: r.label || '', variant: '', updated_at: null, revision: null, warnings: ['fallback_scanner'],
    terminal_api: 'unknown',
  };
}

async function rawScan() {
  const scannedAt = new Date().toISOString();
  let r = await run('python3', [TS_PY, 'scan', '--json'], SCAN_TIMEOUT_MS);
  if (r.ok) {
    try {
      const rows = JSON.parse(r.stdout);
      if (Array.isArray(rows)) {
        const terminalApi = rows.length ? (rows.every((x) => x.terminal_api === 'available') ? 'available' : (rows.some((x) => x.terminal_api === 'unavailable') ? 'unavailable' : 'mixed')) : 'unknown';
        return { rows, stale: terminalApi === 'unavailable' || rows.length === 0, terminalApi, source: 'terminal_status.py', error: null, scannedAt };
      }
    } catch (e) { r = { ok: false, err: 'terminal_status.py bad json: ' + e.message }; }
  }
  const primaryErr = r.err;
  const f = await run('bash', [ALL_SH, '--json'], SCAN_TIMEOUT_MS);
  if (f.ok) {
    try {
      const rows = JSON.parse(f.stdout).map(normalizeAllcolordotsRow);
      // a fallback that substitutes a different scanner is NEVER a clean PASS → stale:true
      return { rows, stale: true, terminalApi: 'unknown', source: 'allcolordots.sh (fallback)', error: 'primary scanner failed: ' + primaryErr, scannedAt };
    } catch (e) { return { rows: [], stale: true, terminalApi: 'unknown', source: 'none', error: 'both scanners failed: ' + primaryErr + ' / ' + e.message, scannedAt }; }
  }
  return { rows: [], stale: true, terminalApi: 'unknown', source: 'none', error: 'both scanners failed: ' + primaryErr + ' / ' + f.err, scannedAt };
}

async function scan() {
  if (cached && Date.now() - cached.ts < CACHE_MS) return cached.result;
  // Stale-while-revalidate for the SCAN too: if a scan is running and we have a previous one,
  // serve it (flagged stale when it is genuinely old) instead of waiting ~90s.
  if (cached && !inflight) startScan(); // expired + nothing running → refresh in the background
  if (inflight && cached) {
    const age = Date.now() - cached.ts;
    return { ...cached.result, stale: !!cached.result.stale || age > STALE_AFTER_MS, scanAgeMs: age, refreshing: true };
  }
  if (inflight) return inflight; // very first scan: nothing to serve yet, wait for it
  return startScan();
}
function startScan() {
  if (inflight) return inflight;
  inflight = rawScan().then((res) => {
    cached = { ts: Date.now(), result: res };
    inflight = null;
    res.needsSteve = res.rows.filter((x) => x.live && NEEDS_STEVE.has(x.color)).length;
    audit.heartbeat(res);
    return res;
  }).catch((e) => { inflight = null; const res = { rows: [], stale: true, terminalApi: 'unknown', source: 'none', error: String(e.message), scannedAt: new Date().toISOString() }; audit.heartbeat(res); return res; });
  return inflight;
}

/** freshLive() — UNCACHED. Map tty → row for live rows. Used by every POST guard. */
async function freshLive() {
  const res = await rawScan();
  cached = { ts: Date.now(), result: res };
  res.needsSteve = res.rows.filter((x) => x.live && NEEDS_STEVE.has(x.color)).length;
  audit.heartbeat(res);
  const live = new Map();
  for (const r of res.rows) if (r.live && TTY_RE.test(String(r.tty))) live.set(r.tty, r);
  return { scan: res, live };
}

/**
 * keyOf(row, detail) — `tty|ticket|color|<content digest>`. The digest covers the parsed
 * option labels (or the last text) so a question that CHANGES while tty/ticket/color stay
 * the same still 409s a stale card (Cody hole #3, TK-11793). keyFor(row) resolves detail.
 */
function digestOf(detail) {
  // Digest ONLY the menu (question + option labels): that is the content a click could
  // mis-answer, and it is stable while displayed. A text card's last lines change every
  // second on a thinking session, so hashing them 409'd every correct click.
  if (!detail) return '-';
  const q = detail.question && detail.question.questions && detail.question.questions[0];
  if (!q) return '-';
  // OPTION LABELS ONLY. The question prose differs between the transcript path (the tool_use
  // input) and the pane path (the wrapped on-screen rendering), and a session with transcript
  // saving on can resolve either way from one read to the next — hashing the prose 409'd a
  // correct click in the same second. Labels are identical across both sources.
  const basis = q.options.map((o) => String(o.label || '').replace(/\s+/g, ' ').trim().toLowerCase()).join('|');
  if (!basis) return '-';
  return crypto.createHash('sha1').update(basis).digest('hex').slice(0, 8);
}
function keyOf(row, detail) { return `${row.tty}|${row.ticket || '-'}|${row.color}|${digestOf(detail)}`; }
function keyFor(row) { const s = transcript.resolve(row); return keyOf(row, s.detail); }

function toIso(started) {
  if (!started) return null;
  const t = Date.parse(started);
  return Number.isNaN(t) ? null : new Date(t).toISOString();
}

function kindOf(row, detail, session) {
  const c = row.color;
  if (c === 'purple') return 'gated';
  if (c === 'orange') return 'paste';
  // A queued input in the prompt box means the on-screen menu (if any) is SUPERSEDED —
  // downgrade to text so the option buttons never render (Cody hole #2, TK-11793).
  const liveMenu = !!(detail && detail.question && !detail.queued);
  if (liveMenu) return 'question';
  if (c === 'lightblue') return 'needs-steve';
  return 'text';
}

function buildItem(row, memosByTk, opts) {
  const color = COLORS[row.color] || COLORS.none;
  const session = transcript.resolve(row);
  const detail = session.detail; // null when ambiguous/none
  const warnings = [...(row.warnings || [])];
  if (row.runtime === 'codex') warnings.push('runtime:codex');
  if (session.confidence === 'ambiguous') warnings.push(`transcript ambiguous (${session.candidates.length} candidates) — label only`);
  if (session.confidence === 'none') warnings.push('transcript not found — label only');
  if (session.confidence === 'pane') warnings.push('source: pane contents (transcript saving is off for this session)');
  if (detail && detail.degenerate) warnings.push('pane scrollback is 1-char-per-line (narrow-width collapse) — Focus the tab to redraw');
  if (detail && detail.queued) warnings.push('queued input already in the prompt: ' + String(detail.queued).slice(0, 80));
  if (row.color === 'yellow' && detail && !detail.question && detail.answeredQuestion) warnings.push('stale dot? (question already answered)');
  const memos = row.ticket ? (memosByTk.get(row.ticket) || []) : [];
  const kind = kindOf(row, detail, session);
  return {
    key: keyOf(row, detail),
    tty: row.tty, pid: row.pid, runtime: row.runtime || 'unknown',
    color: row.color, rgb: color.rgb, emoji: color.emoji, meaning: color.meaning,
    ticket: row.ticket || null, label: row.label || '', variant: row.variant || '',
    kind,
    createdAt: toIso(row.owner && row.owner.started),
    updatedAt: row.updated_at || null,
    session: { sessionId: session.sessionId, transcriptPath: session.transcriptPath, cwd: session.cwd, confidence: session.confidence, how: session.how, candidates: session.candidates.length },
    lastText: detail ? detail.lastText : null,
    question: detail ? detail.question : null,
    pasteCmd: detail && detail.pasteCmd ? detail.pasteCmd : (memos[0] && memos[0].pasteLines[0]) || null,
    memo: memos[0] || null,
    memos,
    warnings,
    selfTty: row.tty === writeback.SELF_TTY,
  };
}

/**
 * build({orphans}) → { items, remaining, orphanMemos, scannedAt, cost, stale, source, error }
 */
// Stale-while-revalidate: at 51 live sessions a full build takes many seconds; concurrent
// /api/queue polls (several open tabs × 8s) must not stack behind it. One build runs at a
// time; while it runs, callers get the LAST result immediately with building:true.
const lastBuild = new Map(); // orphans-flag → result
const building = new Map();  // orphans-flag → Promise
async function build(opts = {}) {
  const k = opts.orphans ? 'o' : 'n';
  if (building.has(k)) {
    const prev = lastBuild.get(k);
    if (prev) return { ...prev, building: true };
    return building.get(k);
  }
  const p = buildUncached(opts).then((r) => { lastBuild.set(k, r); building.delete(k); return r; })
    .catch((e) => { building.delete(k); throw e; });
  building.set(k, p);
  return p;
}
async function buildUncached(opts = {}) {
  const res = await scan();
  const memos = memo.list();
  const byTk = memo.byTicket(memos);
  const rows = res.rows.filter((r) => r.live && NEEDS_STEVE.has(r.color) && TTY_RE.test(String(r.tty)) && r.tty !== writeback.SELF_TTY);
  rows.sort((a, b) => (a.variant === 'stopped' ? 0 : 1) - (b.variant === 'stopped' ? 0 : 1) || (PRIORITY[a.color] ?? 9) - (PRIORITY[b.color] ?? 9) || String(a.tty).localeCompare(String(b.tty)));
  transcript.allPaneContents(); // warm ONE pane batch for every resolve below (and the detection pass)
  const items = rows.map((r) => buildItem(r, byTk, opts));
  // Pane-detected questions: a live claude session whose SCREEN shows a real AskUserQuestion
  // menu but whose dot is not a needs-Steve color (none/green/pink — e.g. it could not dot
  // itself because of the CLAUDE_CODE_CHILD_SESSION env leak). Pane truth beats dot truth.
  const panes = transcript.allPaneContents();
  for (const r of res.rows) {
    if (!r.live || !TTY_RE.test(String(r.tty)) || r.tty === writeback.SELF_TTY || NEEDS_STEVE.has(r.color)) continue;
    if (r.runtime === 'codex') continue;
    const text = panes.get(r.tty); if (!text) continue;
    const d = transcript.parsePane(text);
    if (!d.question || d.queued) continue;
    const synth = { ...r, color: 'yellow', dotColor: r.color, label: (r.label || `${r.tty}`) + ' · question on screen (dot not set)' };
    const it = buildItem(synth, byTk, opts);
    it.dotColor = r.color;
    it.warnings.unshift(`pane-detected question — this session's dot is "${r.color || 'none'}", not yellow`);
    items.push(it);
  }
  const linked = new Set(items.map((i) => i.ticket).filter(Boolean));
  const orphanMemos = opts.orphans ? memos.filter((m) => !m.ticket || !linked.has(m.ticket)).map((m) => ({ ...m, excerpt: m.excerpt.slice(0, 600) })) : [];
  // Ticket backlog lane (TK-11793 cycle 3, from night-TK-11793's offer): tickets waiting on
  // Steve (steve_action / external_wait blockers, or status blocked/stopped) whose pane is NOT
  // open — invisible to a tty scan by construction. Display/route only: nothing to type into.
  let ticketBacklog = [], ticketBacklogCount = 0;
  try {
    const all = await ticketlane.loadTickets();
    const lane = all.filter((c) => !linked.has(c.ticket)).map((c) => ({ ...c, urgency: ticketlane.urgency(c), body: String(c.body || '').slice(0, 800) }))
      .sort((a, b) => b.urgency - a.urgency || new Date(a.created) - new Date(b.created));
    ticketBacklogCount = lane.length;
    ticketBacklog = opts.orphans ? lane.slice(0, 80) : [];
  } catch (e) { /* lane is best-effort; a read failure never blocks the live stream */ }
  return { items, remaining: items.length, orphanMemos, orphanCount: memos.filter((m) => !m.ticket || !linked.has(m.ticket)).length, ticketBacklog, ticketBacklogCount, bootId: BOOT_ID, scannedAt: res.scannedAt, cost: '$0 (local)', stale: !!res.stale, source: res.source, error: res.error, terminalApi: res.terminalApi };
}

/** one item, fresh (uncached scan) — for the post-answer confirm. */
async function item(tty) {
  transcript._cache.delete && [...transcript._cache.keys()].forEach((k) => transcript._cache.delete(k));
  const { scan: res, live } = await freshLive();
  const row = live.get(tty);
  if (!row) return { item: null, gone: true, scannedAt: res.scannedAt, stale: !!res.stale };
  const memos = memo.list();
  return { item: buildItem(row, memo.byTicket(memos), {}), gone: false, scannedAt: res.scannedAt, stale: !!res.stale };
}

module.exports = { scan, freshLive, build, item, keyOf, keyFor, COLORS, PRIORITY, NEEDS_STEVE, TTY_RE };