#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
# Copyright (c) 2026 george
"""roost -- every Claude worker and the local infra, on one screen, live.

Runs unchanged on COOPER (Windows, py3.14) and hyrule (macOS, py3.9). Stdlib only,
no claudectl: claudectl ships no Windows build, and everything it reports is
already in Claude Code's own state.

    roost.py            one frame                 (Windows: .PY is in PATHEXT)
    ./roost -w          live, REFRESH_SECONDS apart   (macOS)
    roost.py -w 5       slower refresh for this run
    roost.py --json     records, for piping

In live mode: space repaints now, q quits, Ctrl-C quits. The refresh interval is
the REFRESH_SECONDS constant below -- edit it to change the default everywhere.

i arms interactive mode, off by default. Arming it is what turns on the cursor
and the EXPERIMENTAL tag together -- one key for the whole risky half, so a
stray keypress on a dashboard left running cannot end a session by accident.
Once armed, j/k (or the arrow keys) raise a cursor, which is the only way to
act on a row: x stops the selected session, y copies its sessionId for
`claude --resume`. Both act on the row object that was on screen when the key
was pressed, never on an index re-resolved afterwards -- rows reorder between
frames as sessions go quiet, and an index that outlived its frame would
eventually hit the wrong one.

Sources, all local and all read-only:

  ~/.claude/sessions/<pid>.json    Claude Code workers: pid, sessionId, cwd, name
  ~/.claude/projects/*/<sid>.jsonl Claude transcripts -- model, usage, task text
  ~/.cursor/projects/*/agent-transcripts/*/*.jsonl
                                   Cursor composers (no pid; freshness from mtime)
  127.0.0.1 ports                  ollama / litellm / openwebui

Backends default to both Claude Code and Cursor (ROOST_BACKENDS=claude,cursor).

WORKERS is what each session *is*; INFRA is what it is running against. A
service that has never answered on an untouched default port shows a dim
"off?", not a red DOWN -- hyrule has no local inference stack, and anyone
without a local gateway at all (pure openrouter or the Anthropic API, no
litellm) is in the same boat: nothing is broken, roost just was never told a
port. Red DOWN is reserved for a port you configured, or a service that was up
earlier this run and stopped answering. If your ports differ from the
11434/4000/8080 defaults, override with ROOST_OLLAMA_PORT / ROOST_LITELLM_PORT
/ ROOST_OPENWEBUI_PORT (or the matching --ollama-port / --litellm-port /
--openwebui-port flags, which win per-run).
roost writes nothing to disk on your behalf besides its own run log
(~/.claude/logs/roost.jsonl, off with --no-log) -- no config file, ever.

Context is an estimate: the last assistant turn's input + cache_read +
cache_creation, over the model's window. It tracks what Claude Code shows without
being derived from it.
"""

from __future__ import annotations

import argparse
import glob
import json
import os
import shutil
import signal
import socket
import sqlite3
import re
import subprocess
import sys
import textwrap
import threading
import time
from collections import deque
from datetime import datetime
from pathlib import Path, PureWindowsPath
from urllib.parse import unquote, urlparse

# release-please rewrites the line below on a release PR. The marker is on
# its own line rather than trailing the assignment because release.yml,
# build-deb.sh and check-version-consistency.sh all parse this line with a
# greedy `sed -n 's/^__version__ = "\(.*\)"/\1/p'`, which would swallow a
# trailing comment into the version string.
# x-release-please-start-version
__version__ = "0.15.0"
# x-release-please-end

SCHEMA_SNAPSHOT = "roost.snapshot.v1"

HOME = Path.home()
SESSIONS_DIR = HOME / ".claude" / "sessions"
PROJECTS_DIR = HOME / ".claude" / "projects"
CURSOR_HOME_ENV = "CURSOR_AGENT_HOME"
CURSOR_HOME = Path(os.environ.get(CURSOR_HOME_ENV, str(HOME / ".cursor")))
CURSOR_PROJECTS_DIR = CURSOR_HOME / "projects"
BACKENDS_ENV = "ROOST_BACKENDS"
CURSOR_MAX_IDLE_ENV = "ROOST_CURSOR_MAX_IDLE_SECS"

# The real Anthropic meter -- session/weekly/Fable caps -- has no local source;
# this cache is written by the "claude-usage-scrape" scheduled task (a Claude
# session driving claude-in-chrome against claude.ai/settings/usage), not by
# roost itself. Missing/stale is the normal state on a machine without that
# task, or on hyrule, which has no browser control at all.
USAGE_CACHE = HOME / "claude-usage" / "usage.json"
USAGE_STALE_SECS = 4 * 2 * 3600  # 4x the scrape task's 2h cadence
CREDITS_CACHE = HOME / "claude-usage" / "credits.json"  # written hourly by check-staleness.ps1
USAGE_HISTORY = HOME / "claude-usage" / "history.jsonl"  # one row per distinct scrape, same writer

# ---- config ----------------------------------------------------------------
# Seconds between automatic repaints. Override per-run with `-w N`; space forces
# an immediate repaint regardless.
REFRESH_SECONDS = 1.0

# Only the tail of a transcript matters and they grow to hundreds of MB.
TAIL_BYTES = 262144

# How many past context readings the TREND column spans. Turns, not seconds:
# context only moves when a turn completes, so a time-based window would be
# empty on a session that has been thinking for a minute and misleading on one
# taking a turn a second. Kept small -- it is read out of the transcript tail,
# and a long window would need a longer tail to fill.
HISTORY_TURNS = 8

# A subagent whose parent has exited is still worth seeing for a while -- usually
# it is the run that just finished. Older than this and it is history.
AGENT_RECENT_SECS = 3600

# A subagent counts as working if its transcript was written this recently.
AGENT_ACTIVE_SECS = 30

# Token-flow sparkline on the WORKER table: sample count kept per session, and
# the minimum seconds between samples so a keypress-forced repaint does not
# stuff the history with extra zeros.
SPARK_LEN = 15
SPARK_MIN_STEP = 1.0
# ASCII ramp, dimmest to hottest. Block-drawing characters mojibake in the
# Windows console (same reason bar() is ASCII), so the ramp is punctuation.
# "." is a taken sample with zero flow; a space means no sample yet.
SPARK_RAMP = ".:-=+*#"

# USAGE panel: how far back the tally reaches, and the weekly budget it is
# measured against. There is no local source for the real Anthropic meter, so
# the budget is a number the user sets once after looking at /usage --
# e.g. ROOST_WEEKLY_BUDGET=60M or 850k or a plain integer of tokens.
USAGE_DAYS = 7
USAGE_BUDGET_ENV = "ROOST_WEEKLY_BUDGET"

# GATEWAY panel: where the batch pipeline writes its runs, and where the job
# queue lives. The gateway itself is DB-less, so every activity endpoint 400s --
# the filesystem is the source of truth for what it has been doing.
BATCH_DIR_ENV = "ROOST_BATCH_DIR"
JOBS_DIR_ENV = "JOBS_ROOT"
# A batch run counts as actively writing if its newest output landed within
# ~2x the slower lane's per-item time (~110s for gemma; see the batch README).
BATCH_ACTIVE_SECS = 240
PROXY_LOG_TAIL = 65536

# INFRA panel ports: env var first (so it persists across runs without a flag
# every time), CLI flag overrides per-run. Someone with no local gateway at all
# (pure openrouter/anthropic API, no litellm) just leaves these at the default
# -- the port probe reads DOWN, same as hyrule reads its stack as not running.
OLLAMA_PORT_ENV = "ROOST_OLLAMA_PORT"
LITELLM_PORT_ENV = "ROOST_LITELLM_PORT"
OPENWEBUI_PORT_ENV = "ROOST_OPENWEBUI_PORT"
LITELLM_CONFIG_ENV = "ROOST_LITELLM_CONFIG"
OLLAMA_PORT = int(os.environ.get(OLLAMA_PORT_ENV, 11434))
LITELLM_PORT = int(os.environ.get(LITELLM_PORT_ENV, 4000))
OPENWEBUI_PORT = int(os.environ.get(OPENWEBUI_PORT_ENV, 8080))

# Whether the user told us the port (env var here, CLI flag in main()). A
# configured port that probes closed is DOWN -- they claimed a service lives
# there. An untouched default that has never once answered is "off?" instead:
# most likely there is no such service, or it lives on a port we were never
# told about, and red would send them chasing a crash that never happened.
_PORT_CONFIGURED = {
    "ollama": OLLAMA_PORT_ENV in os.environ,
    "litellm": LITELLM_PORT_ENV in os.environ,
    "openwebui": OPENWEBUI_PORT_ENV in os.environ,
}

# REMOTE panel: ssh aliases come from the environment only, never from file
# contents -- anything writable over the network must not choose ssh targets.
REMOTES_ENV = "ROOST_REMOTES"
REMOTE_CMD_ENV = "ROOST_REMOTE_CMD"
# ssh runs a non-login shell, whose PATH misses Homebrew and per-user bin dirs
# -- so the default widens PATH rather than assuming an install location.
REMOTE_CMD_DEFAULT = (
    'PATH="$PATH:/opt/homebrew/bin:/usr/local/bin:$HOME/Claude/bin" roost --json')
REMOTE_REFRESH_SECS = 30
REMOTE_TIMEOUT_SECS = 15

# Every session roost stops gets one JSON line here. Same shape and the same cap
# as the hook logs next to it, so the same one-liners read all of them.
LOG_PATH = HOME / ".claude" / "logs" / "roost.jsonl"
LOG_MAX_LINES = 5000

# Cursor composers have no pid; show any transcript / header touched within this
# window. composerHeaders.lastUpdatedAt is the preferred freshness signal.
CURSOR_MAX_IDLE_SECS = int(os.environ.get(CURSOR_MAX_IDLE_ENV, "86400"))

# Cursor's global SQLite (composer index + bubble content). Overridable for
# fixtures; default follows the OS app-support layout, not CURSOR_AGENT_HOME.
CURSOR_STATE_DB_ENV = "ROOST_CURSOR_STATE_DB"
CURSOR_WORKSPACE_STORAGE_ENV = "ROOST_CURSOR_WORKSPACE_STORAGE"
# -----------------------------------------------------------------------------


def cursor_state_db():
    """Path to Cursor's global state.vscdb, or None if unset and unknown."""
    override = os.environ.get(CURSOR_STATE_DB_ENV)
    if override:
        return Path(override)
    if sys.platform == "win32":
        return HOME / "AppData" / "Roaming" / "Cursor" / "User" / "globalStorage" / "state.vscdb"
    if sys.platform == "darwin":
        return (HOME / "Library" / "Application Support" / "Cursor" / "User"
                / "globalStorage" / "state.vscdb")
    return HOME / ".config" / "Cursor" / "User" / "globalStorage" / "state.vscdb"


def cursor_workspace_storage():
    """Per-window workspaceStorage root (holds workspace.json → folder URI)."""
    override = os.environ.get(CURSOR_WORKSPACE_STORAGE_ENV)
    if override:
        return Path(override)
    if sys.platform == "win32":
        return HOME / "AppData" / "Roaming" / "Cursor" / "User" / "workspaceStorage"
    if sys.platform == "darwin":
        return (HOME / "Library" / "Application Support" / "Cursor" / "User"
                / "workspaceStorage")
    return HOME / ".config" / "Cursor" / "User" / "workspaceStorage"


def cursor_folder_uri_to_path(uri):
    """Decode a workspace.json folder URI to a local path, or None.

    Accepts file:///... only; vscode-remote:// and other schemes are skipped.
    Windows drive URIs (file:///c%3A/Users/...) decode to a PureWindowsPath
    string on every host -- Path.resolve() on POSIX would otherwise treat
    'C:\\Users\\...' as relative to the runner cwd.
    """
    if not uri or not isinstance(uri, str):
        return None
    if not uri.startswith("file:"):
        return None
    try:
        parsed = urlparse(uri)
        path = unquote(parsed.path or "")
    except ValueError:
        return None
    if not path:
        return None
    # Windows: file:///c%3A/Users/... → /c:/Users/... → c:\Users\...
    if path.startswith("/") and len(path) >= 3 and path[2] == ":":
        return str(PureWindowsPath(path[1:]))
    return str(Path(path))


def _cursor_path_parts(path):
    """Path parts for Cursor's project-slug encoding.

    Windows-style absolute paths (drive letter / backslashes) use
    PureWindowsPath even on POSIX, because workspace.json on a Windows Cursor
    install stores those strings and CI must slug them the same way COOPER does.
    """
    s = str(path)
    if re.match(r"^[A-Za-z]:[\\/]", s) or ("\\" in s and ":" in s[:3]):
        p = PureWindowsPath(s)
        return [p.drive.rstrip(":").lower()] + [x for x in p.parts[1:] if x]
    p = Path(s).resolve()
    if p.drive:
        return [p.drive.rstrip(":").lower()] + [x for x in p.parts[1:] if x]
    return [x for x in p.parts if x and x != "/"]


def _cursor_path_basename(path):
    s = str(path or "")
    if not s:
        return "-"
    if re.match(r"^[A-Za-z]:[\\/]", s) or ("\\" in s and ":" in s[:3]):
        return PureWindowsPath(s).name or "-"
    return Path(s).name or "-"


def read_cursor_workspace_folders(storage_dir=None):
    """workspaceId → absolute cwd from workspaceStorage/*/workspace.json."""
    root = Path(storage_dir) if storage_dir else cursor_workspace_storage()
    out = {}
    if not root.is_dir():
        return out
    try:
        entries = list(root.iterdir())
    except OSError:
        return out
    for d in entries:
        wj = d / "workspace.json"
        if not wj.is_file():
            continue
        try:
            data = json.loads(wj.read_text(encoding="utf-8"))
        except (OSError, ValueError):
            continue
        if not isinstance(data, dict):
            continue
        cwd = cursor_folder_uri_to_path(data.get("folder"))
        if cwd:
            out[d.name] = cwd
    return out

# Set in main(); --no-log turns it off.
LOGGING = True

# agentId -> {description, status, model, type}, harvested from the parent
# transcript. Merged as records arrive; evicted only when the agent itself is
# gone (prune_caches below).
_AGENT_META = {}
_AGENT_LABEL = {}

# parent transcript path -> bytes already harvested for agent meta.
_HARVEST_POS = {}

# Keys touched since the last prune. Anything not re-touched belongs to a
# session or agent that no longer exists; keeping it is a slow leak in a
# dashboard left open for days.
_SEEN_PATHS = set()
_SEEN_AGENTS = set()

# path -> (mtime, parsed result). At a 1s refresh, re-reading 256 KB from every
# transcript each tick is megabytes of disk per second for no new information --
# a transcript that has not been written to cannot have a new model or usage.
_SCAN_CACHE = {}

# path -> first-turn token total (system prompt + tool/skill/MCP schemas,
# before any real work happens). Unlike _SCAN_CACHE this never invalidates --
# the first assistant turn a transcript ever wrote does not change -- so it is
# read once per session, from the head of the file, not the tail.
_START_CACHE = {}

# composer_id -> (transcript_path, project_slug). Rebuilt each frame; cleared in
# prune_caches so a vanished composer does not keep a stale path forever.
_CURSOR_TX_INDEX = None

# True after read_cursor_composer_headers successfully opened state.vscdb.
# Distinguishes "DB present, zero rows" from "DB missing / unreadable", which
# is when the agent-transcripts glob fallback still has to run.
_CURSOR_HEADERS_OK = False

# Live mode may paint the first frame before localhost INFRA probes finish.
_INFRA_ALLOW_DEFER = False

# When collect_snapshot() last ran -- the header's "updated Ns ago" chip. It
# deliberately survives render-only repaints (a resize reflows from cache), so
# the chip reports data age, not paint age. None until the first collect.
_LAST_COLLECT = None

# sessionId -> {"prev": last ctx_tokens, "t": last sample time, "hist": deque}.
# In-memory only: the sparkline shows flow since roost started, nothing older.
_SPARK = {}

# path -> {"mtime", "size", "counts": {(day, model): tokens}}. Incremental: on
# growth only the appended bytes are read, so the full-file pass happens once
# per transcript per roost run.
_USAGE_CACHE = {}

# Nothing in the transcript records which context window a session was opened
# with. Best source: the model name itself -- Anthropic documents each model's
# real window, and unlike usage that is exact regardless of how little of it
# has been used. A claude-fable-5 worker at 177k tokens is ~18% of its real 1M
# window; scored against the wrong 200k tier (usage-only inference) that read
# as an alarming 89% and tripped a false NEAR LIMIT warning.
#
# MODEL_WINDOWS is exact-match first, then longest-matching-prefix, so a dated
# snapshot under a known family (any claude-haiku-4-5-*) resolves without its
# own table row. A model this table has never heard of falls back to the old
# behaviour -- the smallest standard tier the observed usage still fits in --
# and its label is "~"-marked so an inferred window never looks as certain as
# a known one on screen.
MODEL_WINDOWS = {
    "claude-fable-5": 1000000,
    "claude-opus-5": 1000000,
    "claude-sonnet-5": 1000000,
    "claude-haiku-4-5": 200000,
    # Legacy -- may still appear in old transcripts.
    "claude-opus-4-8": 1000000,
    "claude-opus-4-7": 1000000,
    "claude-opus-4-6": 1000000,
    "claude-sonnet-4-6": 1000000,
    "claude-sonnet-4-5-20250929": 200000,
    "claude-opus-4-5-20251101": 200000,
    "claude-opus-4-1-20250805": 200000,
    # Cursor agent models (Task tool_use / modelConfig). Windows measured from
    # composerData.promptTokenBreakdown.maxTokens on COOPER (= 256000), not
    # Anthropic docs. Prefix match covers -fast / -thinking / -medium suffixes.
    "composer-2.5": 256000,
    "composer-2": 256000,
    "gpt-5.6": 256000,
    "gpt-5": 256000,
    "cursor-grok-4.5": 256000,
    "grok-4.5": 256000,
}

WINDOW_TIERS = ((200000, "200k"), (256000, "256k"), (1000000, "1M"))
WINDOW_TIERS_ENV = "ROOST_WINDOW_TIERS"
WINDOW_ENV = "ROOST_WINDOW"


def _parse_size_token(token):
    """Parse 200000, 200k, or 1M into an int. None if it is not a size."""
    if token is None:
        return None
    raw = str(token).strip().replace("_", "").replace(",", "")
    if not raw:
        return None
    try:
        if raw[-1] in "kKmM" and raw[:-1].replace(".", "", 1).isdigit():
            n = float(raw[:-1])
            return int(n * (1000000 if raw[-1] in "mM" else 1000))
        if raw.isdigit():
            return int(raw)
    except ValueError:
        return None
    return None


def parse_window_tiers(raw):
    """Parse ROOST_WINDOW_TIERS (`200000:200k,1000000:1M`).

    Returns (tiers_tuple, error_or_None). On any parse failure the caller keeps
    the built-in list -- a bad env var is a labelled gap, not a crash.
    """
    if raw is None or not str(raw).strip():
        return WINDOW_TIERS, None
    tiers = []
    for part in str(raw).split(","):
        part = part.strip()
        if not part:
            continue
        if ":" not in part:
            return WINDOW_TIERS, "ROOST_WINDOW_TIERS: missing label in %r" % part
        size_s, label = part.split(":", 1)
        size = _parse_size_token(size_s.strip())
        label = label.strip()
        if not size or size <= 0 or not label:
            return WINDOW_TIERS, "ROOST_WINDOW_TIERS: bad entry %r" % part
        tiers.append((size, label))
    if not tiers:
        return WINDOW_TIERS, "ROOST_WINDOW_TIERS is empty"
    tiers.sort(key=lambda t: t[0])
    return tuple(tiers), None


def parse_forced_window(raw):
    """Parse ROOST_WINDOW (`1M`, `200k`, or a token count). None if unset."""
    if raw is None or not str(raw).strip():
        return None, None
    token = str(raw).strip()
    size = _parse_size_token(token)
    if not size or size <= 0:
        return None, "ROOST_WINDOW: not a size: %r" % token
    if len(token) >= 2 and token[-1] in "kKmM" and token[:-1].replace(".", "", 1).isdigit():
        label = token[:-1] + ("M" if token[-1] in "mM" else "k")
    else:
        label = _window_label(size)
    return (size, label), None


def active_window_tiers():
    """Built-in WINDOW_TIERS, or ROOST_WINDOW_TIERS when that parses."""
    tiers, err = parse_window_tiers(os.environ.get(WINDOW_TIERS_ENV))
    return tiers if err is None else WINDOW_TIERS


def window_config_note():
    """One-line WINDOW status for the frame, or None when using silent defaults."""
    forced, ferr = parse_forced_window(os.environ.get(WINDOW_ENV))
    if ferr:
        return ferr + "; using built-in tiers"
    if forced:
        return "override %s (%s)" % (forced[1], WINDOW_ENV)
    raw = os.environ.get(WINDOW_TIERS_ENV)
    tiers, terr = parse_window_tiers(raw)
    if terr:
        return terr + "; using built-in 200k/256k/1M"
    if raw and str(raw).strip():
        return "tiers %s (%s)" % (", ".join(t[1] for t in tiers), WINDOW_TIERS_ENV)
    return None


def model_window(model):
    """Known window size for `model`, or None if it is not in MODEL_WINDOWS --
    exact match first, then the longest matching prefix."""
    if not model:
        return None
    if model in MODEL_WINDOWS:
        return MODEL_WINDOWS[model]
    best_key, best_size = "", None
    for key, size in MODEL_WINDOWS.items():
        if model.startswith(key) and len(key) > len(best_key):
            best_key, best_size = key, size
    return best_size


def _window_label(size):
    if size >= 1000000:
        return "1M"
    if size % 1000 == 0:
        return "%dk" % (size // 1000)
    return str(size)


def window_for(tokens, model=None):
    """(window_size, label) for a session's context window. Known models
    resolve exactly off MODEL_WINDOWS; anything else falls back to the old
    usage-based inference, its label "~"-marked to say so.

    ROOST_WINDOW, when set and parseable, skips inference and the model table
    entirely -- the user stated the window. ROOST_WINDOW_TIERS replaces the
    built-in fallback list. Neither is silent: window_config_note() surfaces
    the assumption on the frame.
    """
    forced, ferr = parse_forced_window(os.environ.get(WINDOW_ENV))
    if ferr is None and forced is not None:
        return forced
    size = model_window(model)
    if size is not None:
        return size, _window_label(size)
    tiers = active_window_tiers()
    tokens = tokens or 0
    for size, label in tiers:
        if tokens <= size:
            return size, "~" + label
    return tiers[-1][0], "~" + tiers[-1][1]


# ---- auto-compact resolution ------------------------------------------------
# autoCompactEnabled is a Claude Code setting -- never written to a session lock
# file or a transcript, so seeing a worker with it off takes walking the same
# settings.json hierarchy Claude Code itself merges: managed, then a project's
# own .claude/settings.local.json, then its .claude/settings.json, then the
# user's ~/.claude/settings.json. Highest scope that sets the key wins; a file
# that exists but never mentions autoCompactEnabled (or its env-var twin,
# DISABLE_AUTO_COMPACT) is transparent, same as Claude Code's own merge.
#
# What this cannot see: a CLI flag the session itself was launched with, or
# DISABLE_AUTO_COMPACT exported in a shell rather than written into a
# settings.json's own "env" block -- roost reads files, not another process's
# environment or argv.
AUTO_COMPACT_KEY = "autoCompactEnabled"
AUTO_COMPACT_ENV_KEY = "DISABLE_AUTO_COMPACT"
MANAGED_SETTINGS_PATH = {
    "win32": Path(r"C:\Program Files\ClaudeCode\managed-settings.json"),
    "darwin": Path("/Library/Application Support/ClaudeCode/managed-settings.json"),
}.get(sys.platform, Path("/etc/claude-code/managed-settings.json"))


def _auto_compact_from_file(path):
    """True/False if this scope's settings.json decides the setting, else None
    -- meaning fall through to the next scope down."""
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return None
    if not isinstance(data, dict):
        return None
    if AUTO_COMPACT_KEY in data:
        return bool(data[AUTO_COMPACT_KEY])
    env = data.get("env")
    if isinstance(env, dict) and AUTO_COMPACT_ENV_KEY in env:
        # DISABLE_AUTO_COMPACT is the env-var mirror of the settings key --
        # truthy disables, so the boolean it decides is inverted.
        return str(env[AUTO_COMPACT_ENV_KEY]).strip().lower() not in ("1", "true", "yes")
    return None


def auto_compact_enabled(cwd):
    """Effective autoCompactEnabled for a session launched from `cwd`. Defaults
    True -- Claude Code's own built-in default -- if nothing in the chain sets
    it anywhere."""
    if not cwd:
        return True
    base = Path(cwd)
    for path in (MANAGED_SETTINGS_PATH,
                 base / ".claude" / "settings.local.json",
                 base / ".claude" / "settings.json",
                 HOME / ".claude" / "settings.json"):
        result = _auto_compact_from_file(path)
        if result is not None:
            return result
    return True


def _services():
    # A function, not a constant tuple, because the ports can change after
    # import: main() applies CLI flags over the env-var defaults above before
    # the first collect_infra() call.
    return (
        ("ollama", OLLAMA_PORT, "/api/ps"),
        ("litellm", LITELLM_PORT, "/health/liveliness"),
        ("openwebui", OPENWEBUI_PORT, "/health"),
    )


RESET = "\033[0m"
BOLD = "\033[1m"
DIM = "\033[2m"
REVERSE = "\033[7m"
RED = "\033[31m"
GREEN = "\033[32m"
YELLOW = "\033[33m"
BLUE = "\033[34m"
MAGENTA = "\033[35m"
CYAN = "\033[36m"

# Selection is the one painted background: black on cyan.
SEL = "\033[30;46m"


def _ident_blue():
    """The identity colour: bright blue, always paired with BOLD by callers.

    Plain ANSI blue (4) is illegible on common dark palettes, so request
    xterm-256 index 12 when the terminal reports 256 colours; elsewhere fall
    back to plain blue and let the mandatory BOLD brighten it on 8-colour
    terminals.
    """
    if ("256" in os.environ.get("TERM", "")
            or os.environ.get("COLORTERM") in ("truecolor", "24bit")
            or os.environ.get("WT_SESSION")):
        return "\033[38;5;12m"
    return BLUE


IDENT_BLUE = _ident_blue()

# Set once in main(), after we know whether the terminal can render escapes.
COLOR = False


def c(text, *codes):
    if not COLOR or not codes:
        return text
    return "".join(codes) + text + RESET


_SGR_RE = re.compile(r"\033\[([0-9;]*)m")


def _blacken_fg(params):
    """Rewrite the foreground colour in one SGR parameter list to black (30),
    keeping every non-colour attribute (bold, dim, reverse, reset) as it was.

    Handles the three foreground forms: 30-37 and 90-97, 38;5;N (256-colour,
    which IDENT_BLUE uses), and 38;2;R;G;B (truecolour).
    """
    out = []
    parts = params.split(";") if params else [""]
    i = 0
    while i < len(parts):
        p = parts[i]
        if p.isdigit() and (30 <= int(p) <= 37 or 90 <= int(p) <= 97):
            out.append("30")
        elif p == "38" and i + 1 < len(parts) and parts[i + 1] == "5":
            out.append("30")
            i += 2
        elif p == "38" and i + 1 < len(parts) and parts[i + 1] == "2":
            out.append("30")
            i += 4
        else:
            out.append(p)
        i += 1
    return ";".join(out)


def highlight(line):
    """Paint a whole line that already contains colour as the selection bar
    (black on cyan -- the one painted background, uniformly).

    Two things would otherwise break the bar. Every per-cell colour ends in
    RESET, which clears the background along with the colour -- so a naive
    wrap highlights only up to the first coloured cell; re-arming after each
    RESET keeps the bar unbroken. And the cells keep their own foregrounds
    (bright-blue WORKER, yellow/red CTX, bucket colours), which on a cyan
    background read as blue-on-cyan and yellow-on-cyan mud; every inner
    foreground is rewritten to black while BOLD/DIM survive, so weight still
    carries the row's emphasis and the bar is one colour pair end to end.
    """
    if not COLOR:
        return line
    inner = _SGR_RE.sub(lambda m: "\033[" + _blacken_fg(m.group(1)) + "m", line)
    return SEL + inner.replace(RESET, RESET + SEL) + RESET


def ascii_safe(s):
    """Drop characters the console cannot render.

    Task text is free-form prose and often carries em dashes and smart quotes;
    the Windows console codepage turns those into replacement blobs mid-table.
    Deliberately dialect-independent: transcript *data* is stripped even in
    the Unicode dialect (it can contain anything, including width-breaking
    characters), while the dialect table's own glyphs are added by the render
    layer around this gate and never pass through it.
    """
    if not s:
        return ""
    return "".join(ch if 32 <= ord(ch) < 127 else "?" for ch in s)


def visible_len(s):
    """Length ignoring ANSI escapes -- what the terminal actually shows."""
    n = 0
    i = 0
    while i < len(s):
        if s[i] == "\033":
            j = s.find("m", i)
            if j == -1:
                break
            i = j + 1
            continue
        n += 1
        i += 1
    return n


def clip_ansi(s, width):
    """Clip to `width` visible columns, keeping escapes intact.

    A naive s[:width] counts escape bytes as columns and truncates mid-sequence,
    which both over-clips the text and leaks raw escape codes onto the screen.
    """
    if visible_len(s) <= width:
        return s
    out = []
    n = 0
    i = 0
    while i < len(s) and n < width:
        if s[i] == "\033":
            j = s.find("m", i)
            if j == -1:
                break
            out.append(s[i:j + 1])
            i = j + 1
            continue
        out.append(s[i])
        n += 1
        i += 1
    if COLOR:
        out.append(RESET)
    return "".join(out)


# ---- glyph dialects ---------------------------------------------------------
# Two dialects, one vocabulary -- chosen by the terminal, not the product
# (docs/design-language.md in leghorn). The Unicode tier is the preferred
# rendering wherever it can display: rounded frames around the panels, liveness
# dots, check/cross marks, real ellipses and middle-dot separators. The ASCII
# tier is the fallback and *always* the dialect of pipe-safe output: pipes,
# --once, and --json keep their historical bytes exactly. Every glyph comes
# from the active table -- a frame must never mix dialects.
#
# The [###---] context bars stay ASCII in both dialects on purpose: they are a
# data texture, not vocabulary, and the hash bar is legible everywhere.

ASCII_ENV = "ROOST_ASCII"

# Marker entries carry their own trailing space so the ASCII entry can be the
# empty string without leaving a stray gap in front of the word it decorates.
_GLYPHS_UNICODE = {
    "ok": "✓ ",        # check mark before a green "up"
    "fail": "✗ ",      # cross before a bold-red DOWN (the word stays)
    "working": "● ",   # liveness dot: working
    "idle": "○ ",      # liveness dot: idle / parked
    "ell": "…",        # elision
    "sep": "·",        # list separator (the QUIET joiner)
    "tl": "╭", "tr": "╮", "bl": "╰", "br": "╯",
    "h": "─", "v": "│",
}
_GLYPHS_ASCII = {
    "ok": "", "fail": "", "working": "", "idle": "",
    "ell": "...", "sep": ".",
    # No frame entries: the ASCII dialect draws bold bare titles, never boxes.
}

# Module default is ASCII, which keeps every import-time caller (tests, --json,
# --once) on the historical byte-identical output. Only main()'s live path may
# switch, once, after probing the terminal.
UNICODE = False
GLYPHS = _GLYPHS_ASCII


def set_dialect(unicode_on):
    """Select the glyph table once for the whole session."""
    global UNICODE, GLYPHS
    UNICODE = bool(unicode_on)
    GLYPHS = _GLYPHS_UNICODE if UNICODE else _GLYPHS_ASCII


def probe_unicode(stdout=None, windows=None, env=None):
    """True when stdout is an interactive UTF-8 terminal that can draw glyphs.

    Three gates, all required: stdout is a tty (a pipe never gets the Unicode
    tier whatever encoding it reports), Python's stdout encoding is UTF-8, and
    -- on Windows only -- WT_SESSION is set. That last gate is the one that
    actually excludes legacy conhost: since PEP 528 (Python 3.6) every Windows
    console reports utf-8 regardless of its code page, so the encoding check
    alone cannot tell Windows Terminal from a conhost window whose raster font
    has no box-drawing glyphs. WT_SESSION is set by Windows Terminal and
    nothing else. On POSIX the tty's encoding is the whole story.
    ROOST_ASCII=1 forces the ASCII dialect regardless -- the escape hatch for
    a terminal that lies.

    `windows` and `env` are injectable for tests; the defaults read the real
    platform and environment.
    """
    env = os.environ if env is None else env
    if env.get(ASCII_ENV):
        return False
    out = stdout if stdout is not None else sys.stdout
    try:
        if not out.isatty():
            return False
    except (AttributeError, ValueError):
        return False
    enc = (getattr(out, "encoding", "") or "").replace("-", "").replace("_", "").lower()
    if enc not in ("utf8", "cp65001"):
        return False
    if windows is None:
        windows = os.name == "nt"
    if windows and not env.get("WT_SESSION"):
        return False
    return True


def choose_dialect(pipe_safe, force_ascii, stdout=None, **probe):
    """The startup decision: pipe-safe modes (--once, --json) and an explicit
    --ascii both pin the ASCII dialect before the terminal is even consulted.
    Extra keywords pass through to probe_unicode (tests inject the platform)."""
    if pipe_safe or force_ascii:
        return False
    return probe_unicode(stdout, **probe)


def frame_panel(title, body, cols=None):
    """Wrap rendered panel lines in a rounded frame -- Unicode dialect only.

    Chrome cyan at dim weight, uppercase title inset two columns and painted
    bold, matching leghorn's Pane.frame. Width hugs the widest body line but
    never reaches the terminal's last column (a glyph in the final cell wraps
    onto the next row and persists). The ASCII dialect never calls this: its
    panels keep their bold bare titles, byte-identical to the historical
    output.
    """
    if cols is None:
        cols = term_size()[0]
    label = " %s " % title
    inner = max([visible_len(ln) for ln in body] + [len(label) + 4])
    inner = min(inner, cols - 3)
    label = label[: max(0, inner - 2)]
    top = (c(GLYPHS["tl"] + GLYPHS["h"], CYAN, DIM)
           + c(label, BOLD, CYAN)
           + c(GLYPHS["h"] * max(0, inner - 1 - len(label)) + GLYPHS["tr"],
               CYAN, DIM))
    edge = c(GLYPHS["v"], CYAN, DIM)
    out = [top]
    for ln in body:
        clipped = clip_ansi(ln, inner)
        out.append(edge + clipped + " " * (inner - visible_len(clipped)) + edge)
    out.append(c(GLYPHS["bl"] + GLYPHS["h"] * inner + GLYPHS["br"], CYAN, DIM))
    return out


def panel(title, body, extra=None):
    """Assemble one titled panel in the active dialect.

    Unicode: a rounded frame with the title inset on the top border. ASCII:
    the historical bold bare title line, byte-identical to what it always
    printed. `extra` is the dim annotation the ASCII title line carries after
    the name (e.g. DETAIL's "esc returns"); in the Unicode dialect the title
    lives on the border, so the annotation becomes the first body line.
    """
    if UNICODE:
        inner = ([c("  " + extra, DIM)] if extra else []) + body
        return [""] + frame_panel(title, inner)
    head = c(title, BOLD)
    if extra:
        head += "  " + c(extra, DIM)
    return ["", head] + body


def alive(pid):
    """True if the process exists. os.kill(pid, 0) is POSIX-only."""
    if os.name == "nt":
        import ctypes

        SYNCHRONIZE = 0x00100000
        h = ctypes.windll.kernel32.OpenProcess(SYNCHRONIZE, False, int(pid))
        if h:
            ctypes.windll.kernel32.CloseHandle(h)
            return True
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def trim_log():
    """Hold the log at LOG_MAX_LINES, checked by size so the common write is one
    append and no read. Records run a few hundred bytes; the slack is deliberate."""
    try:
        if LOG_PATH.stat().st_size < LOG_MAX_LINES * 400:
            return
        kept = LOG_PATH.read_text(encoding="utf-8").splitlines()[-LOG_MAX_LINES:]
        LOG_PATH.write_text("\n".join(kept) + "\n", encoding="utf-8")
    except OSError:
        pass


def log_action(action, worker, ok=True, detail=""):
    """Append one record per action roost takes.

    Actions only -- frames are not logged, and neither is task text. The task is
    free-form prose out of a transcript and would turn an audit trail into a
    copy of what was being worked on; name and sessionId identify the session
    without carrying its contents. The numbers alongside are what make the log
    answer a real question later: how much context a sweep actually reclaimed.

    Never raises. A log that cannot be written is not a reason to lose the UI.
    """
    if not LOGGING:
        return
    rec = {
        "ts": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "action": action,
        "ok": ok,
        "host": socket.gethostname(),
        "name": worker.get("name"),
        "pid": worker.get("pid"),
        "session_id": worker.get("session_id"),
        "model": worker.get("model"),
        "ctx_tokens": worker.get("ctx_tokens"),
        "idle_secs": int(worker["idle_secs"]) if worker.get("idle_secs") else None,
    }
    if detail:
        rec["detail"] = detail
    try:
        LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
        with open(str(LOG_PATH), "a", encoding="utf-8") as fh:
            fh.write(json.dumps(rec) + "\n")
    except OSError:
        return
    trim_log()


def terminate(pid):
    """Stop a session process. Returns an error string, or None on success.

    Windows has no cross-process SIGTERM, so this is TerminateProcess: immediate,
    with no chance for the session to shut down cleanly. Transcripts are written
    a turn at a time, so the most that can be lost is a turn already in flight --
    but it is a kill, not a request, and the man page says so. POSIX gets a real
    SIGTERM and the session exits on its own terms.
    """
    if pid in (os.getpid(), os.getppid()):
        # roost run from inside the session it is pointed at: the cursor lands on
        # the row whose process owns this terminal, and x would take roost with it.
        return "refusing to stop roost's own process tree"
    if os.name == "nt":
        import ctypes

        PROCESS_TERMINATE = 0x0001
        h = ctypes.windll.kernel32.OpenProcess(PROCESS_TERMINATE, False, int(pid))
        if not h:
            return "cannot open pid %d (already gone, or not yours)" % pid
        ok = ctypes.windll.kernel32.TerminateProcess(h, 1)
        ctypes.windll.kernel32.CloseHandle(h)
        return None if ok else "TerminateProcess failed on pid %d" % pid
    try:
        os.kill(pid, signal.SIGTERM)
    except OSError as e:
        return "%s (pid %d)" % (e.strerror or e, pid)
    return None


# xclip is the one that may genuinely be absent; a failed copy is reported, not
# swallowed, because the whole point is pasting the id into a resume command.
CLIP_CMD = {"win32": ["clip"], "darwin": ["pbcopy"]}.get(
    sys.platform, ["xclip", "-selection", "clipboard"])


def to_clipboard(text):
    try:
        p = subprocess.Popen(CLIP_CMD, stdin=subprocess.PIPE)
        p.communicate(text.encode("utf-8"))
        return p.returncode == 0
    except OSError:
        return False


def port_open(port, timeout=0.35):
    s = socket.socket()
    s.settimeout(timeout)
    try:
        s.connect(("127.0.0.1", port))
        return True
    except OSError:
        return False
    finally:
        s.close()


def http_json(port, path, timeout=1.5):
    import urllib.request

    url = "http://127.0.0.1:%d%s" % (port, path)
    try:
        with urllib.request.urlopen(url, timeout=timeout) as r:
            return json.loads(r.read().decode("utf-8", "replace"))
    except Exception:
        return None


def transcript_for(session_id):
    if not session_id:
        return None
    hits = glob.glob(str(PROJECTS_DIR / "*" / (session_id + ".jsonl")))
    return hits[0] if hits else None


def read_tail(path, nbytes=TAIL_BYTES):
    try:
        size = os.path.getsize(path)
        with open(path, "rb") as fh:
            if size > nbytes:
                fh.seek(size - nbytes)
                fh.readline()  # drop the partial line the seek landed in
            return fh.read().decode("utf-8", "replace").splitlines()
    except OSError:
        return []


def scan_transcript(path):
    """Model and context from the newest assistant turn that carries usage.

    Also the last HISTORY_TURNS context totals, oldest first, out of the same
    backward walk. History read from the transcript rather than accumulated
    across frames is populated on the very first frame and survives a restart,
    so it works under --once and --json too -- a ring buffer kept in memory
    would give neither, and at a 1s refresh would sample the same turn dozens
    of times over.
    """
    out = {"model": None, "ctx_tokens": None, "last_write": None,
           "title": None, "prompt": None, "ctx_history": []}
    if not path:
        return out
    _SEEN_PATHS.add(path)
    try:
        out["last_write"] = os.path.getmtime(path)
    except OSError:
        pass

    cached = _SCAN_CACHE.get(path)
    if cached is not None and cached[0] == out["last_write"]:
        return dict(cached[1])

    # Walking backwards, take the newest of each: usage (model + context),
    # customTitle (what Claude Code named the session), lastPrompt (what was last
    # asked). Both title records recur throughout the file, so the tail has them.
    for line in reversed(read_tail(path)):
        line = line.strip()
        if not line:
            continue
        # Cheap pre-filter -- json.loads on every tail line is the expensive part.
        has_usage = '"usage"' in line and '"assistant"' in line
        has_title = '"customTitle"' in line
        has_prompt = '"lastPrompt"' in line
        if not (has_usage or has_title or has_prompt):
            continue
        try:
            d = json.loads(line)
        except ValueError:
            continue

        if out["title"] is None and d.get("customTitle"):
            out["title"] = str(d["customTitle"]).strip()
        if out["prompt"] is None and d.get("lastPrompt"):
            out["prompt"] = " ".join(str(d["lastPrompt"]).split())

        msg = d.get("message") or {}
        usage = msg.get("usage") or {}
        if usage:
            total = (
                (usage.get("input_tokens") or 0)
                + (usage.get("cache_read_input_tokens") or 0)
                + (usage.get("cache_creation_input_tokens") or 0)
            )
            if out["model"] is None:
                out["model"] = msg.get("model")
                out["ctx_tokens"] = total
            # A turn that used tools writes several assistant records carrying
            # the same context total. Only a change is a new data point, or a
            # tool-heavy turn would fill the whole window with one turn's value.
            if len(out["ctx_history"]) < HISTORY_TURNS and (
                    not out["ctx_history"] or out["ctx_history"][-1] != total):
                out["ctx_history"].append(total)

        if (out["model"] and out["title"] and out["prompt"]
                and len(out["ctx_history"]) >= HISTORY_TURNS):
            break

    out["ctx_history"].reverse()  # collected newest-first, reported oldest-first

    if out["last_write"] is not None:
        _SCAN_CACHE[path] = (out["last_write"], dict(out))
    return out


def startup_context(path):
    """Tokens billed on the session's first assistant turn.

    Before any user message is really answered, Claude Code has already
    loaded the system prompt plus every tool, skill listing, and MCP server's
    tool schemas -- that's what shows up as input + cache tokens on turn one.
    A big number here means a heavy skills/MCP set is taxing every session
    from the first token, independent of anything the user has done since.
    Read forward from the head of the file (scan_transcript reads the tail),
    and cache forever once found -- turn one never changes.
    """
    if not path:
        return None
    if path in _START_CACHE:
        return _START_CACHE[path]
    total = None
    try:
        with open(path, "r", encoding="utf-8", errors="replace") as fh:
            for line in fh:
                line = line.strip()
                if not line or '"usage"' not in line or '"assistant"' not in line:
                    continue
                try:
                    d = json.loads(line)
                except ValueError:
                    continue
                usage = (d.get("message") or {}).get("usage") or {}
                if not usage:
                    continue
                total = (
                    (usage.get("input_tokens") or 0)
                    + (usage.get("cache_read_input_tokens") or 0)
                    + (usage.get("cache_creation_input_tokens") or 0)
                )
                break
    except OSError:
        return None
    if total is not None:
        _START_CACHE[path] = total
    return total


def harvest_agent_meta(parent_transcript):
    """Pull subagent description/status out of the parent's tool results.

    A subagent's own transcript never states what it was asked to do in short
    form -- only the parent's `toolUseResult` carries `description`, `status`,
    `resolvedModel` and `agentType`, keyed by agentId. Incremental rather than
    tail-only: a busy parent grows past TAIL_BYTES with its agents' results in
    the half a tail read never sees. The full pass happens once per parent per
    roost run; after that only appended bytes are read, so the every-tick call
    for a still-running agent costs one getsize().
    """
    if not parent_transcript:
        return
    # Keeps the byte position alive across prunes while anything still asks
    # about this parent; without it an orphan's dead parent would be evicted
    # and re-read from byte 0 every tick.
    _SEEN_PATHS.add(parent_transcript)
    pos = _HARVEST_POS.get(parent_transcript, 0)
    try:
        size = os.path.getsize(parent_transcript)
        if size < pos:
            pos = 0  # truncated or replaced -- start over
        if size == pos:
            return
        with open(parent_transcript, "rb") as fh:
            fh.seek(pos)
            data = fh.read()
    except OSError:
        return
    # Consume whole lines only; a half-written trailing line waits for the
    # next pass instead of being parsed as garbage and skipped forever.
    cut = data.rfind(b"\n") + 1
    _HARVEST_POS[parent_transcript] = pos + cut
    for line in data[:cut].decode("utf-8", "replace").splitlines():
        if '"agentId"' not in line:
            continue
        try:
            d = json.loads(line)
        except ValueError:
            continue
        r = d.get("toolUseResult")
        if not isinstance(r, dict):
            continue
        aid = r.get("agentId")
        if not aid:
            continue
        # Marked seen even when the agent has no live row: evicting it here
        # would just force a re-harvest next tick.
        _SEEN_AGENTS.add(aid)
        # Merge rather than first-write-wins: an async launch writes an early
        # record with no agentType, and the completed result that carries it
        # would otherwise be discarded.
        meta = _AGENT_META.setdefault(
            aid, {"description": "", "status": "", "model": "", "type": ""})
        for key, field in (("description", "description"), ("status", "status"),
                           ("model", "resolvedModel"), ("type", "agentType")):
            val = r.get(field)
            if val:
                meta[key] = val


def agent_first_prompt(path, agent_id):
    """Fallback label: the opening words of the task the subagent was given.

    Used when the parent's tool result has scrolled out of the tail. The first
    line of a subagent transcript is immutable, so this is cached outright.
    """
    if agent_id in _AGENT_LABEL:
        return _AGENT_LABEL[agent_id]
    label = ""
    try:
        with open(path, "rb") as fh:
            first = fh.readline().decode("utf-8", "replace")
        d = json.loads(first)
        content = (d.get("message") or {}).get("content")
        if isinstance(content, list):
            content = " ".join(
                p.get("text", "") for p in content if isinstance(p, dict))
        label = " ".join(str(content or "").split())[:60]
    except (OSError, ValueError):
        label = ""
    _AGENT_LABEL[agent_id] = label
    return label


def collect_claude_subagents(live_sids):
    """Subagents run inside their parent process, so they have no pid of their
    own -- but each gets its own transcript at

        projects/<slug>/<parentSessionId>/subagents/agent-<id>.jsonl

    which is one level deeper than the main session transcripts.
    """
    rows = []
    now = time.time()
    pattern = str(PROJECTS_DIR / "*" / "*" / "subagents" / "agent-*.jsonl")
    for path in glob.glob(pattern):
        p = Path(path)
        parent_sid = p.parent.parent.name
        parent_live = parent_sid in live_sids
        # mtime before scan_transcript: a cold board can have hundreds of
        # finished sidechains, and reading every tail is what made the first
        # frame stall for seconds with a blank terminal.
        try:
            mtime = os.path.getmtime(path)
        except OSError:
            continue
        age = now - mtime
        if not parent_live and age > AGENT_RECENT_SECS:
            continue  # finished long ago -- history, not a live worker
        agent_id = p.stem[len("agent-"):] if p.stem.startswith("agent-") else p.stem

        info = scan_transcript(path)
        last = info["last_write"] if info["last_write"] is not None else mtime
        age = (now - last) if last else None

        _SEEN_AGENTS.add(agent_id)
        # Re-harvest while the type is still missing: a running agent's early
        # records carry no agentType; the completed result does.
        if not (_AGENT_META.get(agent_id) or {}).get("type"):
            harvest_agent_meta(transcript_for(parent_sid))
        meta = _AGENT_META.get(agent_id) or {}

        label = ascii_safe(
            meta.get("description") or agent_first_prompt(path, agent_id) or "-")
        model = info["model"] or meta.get("model") or "-"
        pct = None
        win_label = "-"
        if info["ctx_tokens"]:
            window, win_label = window_for(info["ctx_tokens"], model)
            pct = 100.0 * info["ctx_tokens"] / float(window)

        # Parent liveness gates everything: if the parent process is gone, nothing
        # can still be writing to this transcript, however recent the last write.
        if not parent_live:
            state = "orphan"
        elif age is not None and age <= AGENT_ACTIVE_SECS:
            state = "working"
        else:
            state = "idle"

        rows.append({
            "source": "claude",
            "agent_id": agent_id,
            # Sanitised like task text: it comes out of a transcript, and a
            # crafted agentType could otherwise carry escapes into the TUI.
            "agent_type": ascii_safe(meta.get("type") or ""),
            "parent_sid": parent_sid,
            "task": label,
            "model": model,
            "ctx_tokens": info["ctx_tokens"],
            "ctx_pct": pct,
            "window": win_label,
            "idle_secs": age,
            "state": state,
            "parent_live": parent_live,
        })
    return rows


def collect_cursor_subagents(live_sids, db_path=None):
    """Cursor Task/subagent composers from composerHeaders (isSubagent=1).

    Same row shape as Claude subagents so render_subagents() is shared. Parent
    liveness uses the parent composerId against live_sids (Cursor worker
    session_ids are composer UUIDs).
    """
    if "cursor" not in _backends():
        return []
    path = Path(db_path) if db_path else cursor_state_db()
    if path is None or not path.is_file():
        return []
    rows = []
    now = time.time()
    try:
        uri = "file:%s?mode=ro" % path.resolve().as_posix()
        con = sqlite3.connect(uri, uri=True, timeout=0.5)
        try:
            cur = con.execute(
                "SELECT composerId, lastUpdatedAt, createdAt, value "
                "FROM composerHeaders "
                "WHERE COALESCE(isSubagent, 0) = 1 "
                "AND COALESCE(isArchived, 0) = 0")
            for cid, lu, created, val in cur:
                try:
                    h = json.loads(val) if val else {}
                except ValueError:
                    h = {}
                if not isinstance(h, dict):
                    h = {}
                info = h.get("subagentInfo") if isinstance(h.get("subagentInfo"), dict) else {}
                parent = (info.get("parentComposerId")
                          or info.get("rootParentConversationId")
                          or info.get("forkedFromComposerId")
                          or "")
                last = _cursor_ms_to_epoch(h.get("lastUpdatedAt") or lu
                                          or h.get("createdAt") or created)
                age = (now - last) if last is not None else None
                parent_live = bool(parent) and parent in live_sids
                # Keep recently-touched orphans (parent closed) for a while,
                # matching Claude's AGENT_RECENT_SECS behaviour.
                if not parent_live and (age is None or age > AGENT_RECENT_SECS):
                    continue
                if age is not None and age > CURSOR_MAX_IDLE_SECS:
                    continue
                _SEEN_AGENTS.add(cid)
                pct = h.get("contextUsagePercent")
                if isinstance(pct, (int, float)):
                    pct = float(pct)
                else:
                    pct = None
                agent_type = ascii_safe(
                    info.get("subagentTypeName") or info.get("subagentType") or "")
                task = ascii_safe(
                    (h.get("name") or h.get("subtitle") or "").strip() or "-")
                # Headers already carry name + contextUsagePercent. Live Cursor
                # transcripts rarely have usage or message.model (those live in
                # state.vscdb), so scanning every Task composer on first paint
                # was pure stall for no MODEL/TOKENS gain.
                model = "-"
                tok = None
                win_label = "-"
                if pct is not None:
                    window, win_label = window_for(1, "composer-2.5")
                    tok = int(round(pct / 100.0 * window))
                    win_label = _window_label(window)
                if not parent_live:
                    state = "orphan"
                elif age is not None and age <= AGENT_ACTIVE_SECS:
                    state = "working"
                else:
                    state = "idle"
                rows.append({
                    "source": "cursor",
                    "agent_id": cid,
                    "agent_type": str(agent_type),
                    "parent_sid": parent,
                    "task": task,
                    "model": model,
                    "ctx_tokens": tok,
                    "ctx_pct": pct,
                    "window": win_label,
                    "idle_secs": age,
                    "state": state,
                    "parent_live": parent_live,
                })
        finally:
            con.close()
    except (sqlite3.Error, OSError, ValueError):
        return []
    return rows


def collect_subagents(live_sids):
    """Claude sidechains plus Cursor Task composers, same row shape."""
    rows = []
    backends = _backends()
    if "claude" in backends:
        rows.extend(collect_claude_subagents(live_sids))
    if "cursor" in backends:
        rows.extend(collect_cursor_subagents(live_sids))
    rows.sort(key=lambda r: (r["state"] != "working", r["idle_secs"] or 1e9))
    return rows


def _backends():
    raw = os.environ.get(BACKENDS_ENV, "claude,cursor")
    return {b.strip().lower() for b in raw.split(",") if b.strip()}


def cursor_project_slug(path):
    """Cursor's ~/.cursor/projects/<slug> encoding for an absolute path."""
    return "-".join(_cursor_path_parts(path))


def _cursor_task_from_text(text):
    if not text:
        return ""
    m = re.search(r"<user_query>\s*(.*?)\s*</user_query>", text, re.DOTALL)
    if m:
        return " ".join(m.group(1).split())
    return " ".join(str(text).split())


def _cursor_model_from_content(content):
    """Newest Task tool_use model, if any. Parent transcripts rarely carry
    message.model; the model string lives on spawned Task calls instead."""
    if not isinstance(content, list):
        return None
    for part in content:
        if not isinstance(part, dict) or part.get("type") != "tool_use":
            continue
        if part.get("name") != "Task":
            continue
        inp = part.get("input")
        if isinstance(inp, dict) and inp.get("model"):
            return str(inp["model"])
    return None


def scan_cursor_transcript(path):
    """Task text, optional usage, and a best-effort model from Cursor JSONL.

    Live COOPER transcripts (71 files measured) carry *no* usage blocks and no
    message.model -- those live in state.vscdb. This scan still matters for the
    user_query task text and as a fallback when the SQLite index is missing.
    """
    out = {"model": None, "ctx_tokens": None, "last_write": None,
           "title": None, "prompt": None, "ctx_history": []}
    if not path:
        return out
    _SEEN_PATHS.add(path)
    try:
        out["last_write"] = os.path.getmtime(path)
    except OSError:
        pass

    cached = _SCAN_CACHE.get(path)
    if cached is not None and cached[0] == out["last_write"]:
        return dict(cached[1])

    for line in reversed(read_tail(path)):
        line = line.strip()
        if not line:
            continue
        try:
            d = json.loads(line)
        except ValueError:
            continue
        role = d.get("role")
        msg = d.get("message") or {}
        content = msg.get("content")

        if role == "user" and out["prompt"] is None:
            text = ""
            if isinstance(content, list):
                for part in content:
                    if isinstance(part, dict) and part.get("type") == "text":
                        text = part.get("text") or ""
                        break
            elif isinstance(content, str):
                text = content
            out["prompt"] = _cursor_task_from_text(text)

        if role != "assistant":
            continue

        if out["model"] is None:
            out["model"] = (msg.get("model") or d.get("model")
                            or _cursor_model_from_content(content))

        usage = msg.get("usage") or d.get("usage") or {}
        if not usage:
            if out["model"] and out["prompt"]:
                # No usage in the file at all is the common case; stop once we
                # have a task and whatever model the Task tools named.
                pass
            continue
        ctx = (
            (usage.get("input_tokens") or usage.get("prompt_tokens") or 0)
            + (usage.get("cache_read_input_tokens") or 0)
            + (usage.get("cache_creation_input_tokens") or 0)
        )
        if out["ctx_tokens"] is None:
            out["ctx_tokens"] = ctx
        if len(out["ctx_history"]) < HISTORY_TURNS and (
                not out["ctx_history"] or out["ctx_history"][-1] != ctx):
            out["ctx_history"].append(ctx)
        if out["model"] and out["prompt"] and len(out["ctx_history"]) >= HISTORY_TURNS:
            break

    out["ctx_history"].reverse()
    if out["last_write"] is not None:
        _SCAN_CACHE[path] = (out["last_write"], dict(out))
    return out


def _cursor_ms_to_epoch(ms):
    """composerHeaders timestamps are ms since epoch; tolerate seconds too."""
    if not isinstance(ms, (int, float)) or ms <= 0:
        return None
    return ms / 1000.0 if ms > 1e12 else float(ms)


def _cursor_recent_branch(h):
    """Most recently interacted branch across a header's trackedGitRepos.

    Cursor records every branch the composer touched, per repo, each with a
    lastInteractionAt -- the newest one is where the work actually is.
    """
    repos = h.get("trackedGitRepos")
    if not isinstance(repos, list):
        return None
    best, best_t = None, None
    for repo in repos:
        if not isinstance(repo, dict):
            continue
        branches = repo.get("branches")
        if not isinstance(branches, list):
            continue
        for b in branches:
            if not isinstance(b, dict):
                continue
            name = (b.get("branchName") or "").strip()
            if not name:
                continue
            t = b.get("lastInteractionAt")
            t = float(t) if isinstance(t, (int, float)) else 0.0
            if best_t is None or t > best_t:
                best, best_t = name, t
    return best


def read_cursor_composer_headers(db_path=None):
    """Non-archived parent composers from state.vscdb's composerHeaders table.

    Returns a list of dicts with composer_id, name, subtitle, branch, ctx_pct,
    last_write, workspace_id. Empty on any error -- missing DB is normal on a
    machine that has never run Cursor.
    """
    global _CURSOR_HEADERS_OK
    _CURSOR_HEADERS_OK = False
    path = Path(db_path) if db_path else cursor_state_db()
    if path is None or not path.is_file():
        return []
    rows = []
    try:
        # URI mode=ro so we never write; Cursor keeps this file open while
        # running and a write-mode open can fail or corrupt.
        uri = "file:%s?mode=ro" % path.resolve().as_posix()
        con = sqlite3.connect(uri, uri=True, timeout=0.5)
        try:
            cur = con.execute(
                "SELECT composerId, workspaceId, lastUpdatedAt, createdAt, "
                "isSubagent, value FROM composerHeaders "
                "WHERE COALESCE(isArchived, 0) = 0")
            for cid, ws, lu, created, is_sub, val in cur:
                if is_sub:
                    continue
                try:
                    h = json.loads(val) if val else {}
                except ValueError:
                    h = {}
                if not isinstance(h, dict):
                    h = {}
                # Draft empty-state composers are noise on the board.
                if h.get("isDraft") or cid == "empty-state-draft":
                    continue
                if h.get("unifiedMode") == "chat" and h.get("isEphemeral"):
                    continue
                pct = h.get("contextUsagePercent")
                if isinstance(pct, (int, float)):
                    pct = float(pct)
                else:
                    pct = None
                last = _cursor_ms_to_epoch(h.get("lastUpdatedAt") or lu
                                          or h.get("createdAt") or created)
                name = (h.get("name") or "").strip() or None
                subtitle = (h.get("subtitle") or "").strip() or None
                rows.append({
                    "composer_id": cid,
                    "workspace_id": ws or (h.get("workspaceIdentifier") or {}).get("id"),
                    "name": name,
                    "subtitle": subtitle,
                    "branch": _cursor_recent_branch(h),
                    "ctx_pct": pct,
                    "last_write": last,
                    "mode": h.get("unifiedMode") or "",
                })
        finally:
            con.close()
        # Query succeeded even if every row was filtered out -- the SQLite
        # index is authoritative, so collect_cursor_workers must not fall
        # through to a full agent-transcripts scan.
        _CURSOR_HEADERS_OK = True
    except (sqlite3.Error, OSError, ValueError):
        return []
    return rows


def _cursor_transcript_index():
    """One directory walk → composer_id → (path, slug).

    collect_cursor_workers used to glob once per idle composer; with dozens of
    live Task composers that is hundreds of identical tree walks per frame.
    """
    global _CURSOR_TX_INDEX
    if _CURSOR_TX_INDEX is not None:
        return _CURSOR_TX_INDEX
    index = {}
    if not CURSOR_PROJECTS_DIR.is_dir():
        _CURSOR_TX_INDEX = index
        return index
    pattern = str(CURSOR_PROJECTS_DIR / "*" / "agent-transcripts" / "*" / "*")
    by_cid = {}
    for path_str in glob.glob(pattern):
        path = Path(path_str)
        if path.suffix not in (".jsonl", ".txt"):
            continue
        by_cid.setdefault(path.parent.name, []).append(path_str)
    for cid, paths in by_cid.items():
        # Match _cursor_transcript_for's old sorted(glob) first-hit rule.
        path = Path(sorted(paths)[0])
        index[cid] = (str(path), path.parent.parent.parent.name)
    _CURSOR_TX_INDEX = index
    return index


def _cursor_transcript_for(composer_id):
    if not composer_id:
        return None, ""
    hit = _cursor_transcript_index().get(composer_id)
    if hit is None:
        return None, ""
    return hit


def _cursor_worker_row(composer_id, name=None, task=None, task_src="-",
                       model="-", ctx_tokens=None, ctx_pct=None, window="-",
                       ctx_history=None, idle_secs=None, age_secs=None,
                       slug="", flow=None, cwd=""):
    short = composer_id[:8] if len(composer_id) >= 8 else composer_id
    if cwd:
        project = _cursor_path_basename(cwd)
    else:
        project = slug.rsplit("-", 1)[-1] if slug else "-"
    return {
        "source": "cursor",
        "name": name or ("cursor/" + short),
        "pid": None,
        "session_id": composer_id,
        "cwd": cwd or "",
        "project": project,
        "cursor_slug": slug,
        "model": model or "-",
        "ctx_tokens": ctx_tokens,
        "ctx_pct": ctx_pct,
        "ctx_history": ctx_history or [],
        "window": window,
        "task": ascii_safe(task or ""),
        "task_src": task_src,
        "idle_secs": idle_secs,
        "age_secs": age_secs,
        "flow": flow if flow is not None else (" " * SPARK_LEN),
        "auto_compact": True,
    }


def _dedupe_cursor_names(rows):
    """Several composers in one workspace often last touched the *same*
    branch, and identical WORKER names make rows indistinguishable. Suffix
    the composer short-id only on collisions so unique branches stay clean."""
    counts = {}
    for r in rows:
        counts[r["name"]] = counts.get(r["name"], 0) + 1
    for r in rows:
        if counts[r["name"]] > 1 and r.get("session_id"):
            r["name"] = r["name"] + "-" + r["session_id"][:4]
    return rows


def collect_cursor_workers():
    """Cursor composers: prefer composerHeaders (name + CTX%), fall back to
    agent-transcript JSONL when the SQLite index is absent."""
    rows = []
    if "cursor" not in _backends():
        return rows
    now = time.time()
    seen = set()
    folders = read_cursor_workspace_folders()

    for h in read_cursor_composer_headers():
        cid = h["composer_id"]
        last = h["last_write"]
        if last is None:
            continue
        idle = now - last
        if idle > CURSOR_MAX_IDLE_SECS:
            continue
        seen.add(cid)
        tpath, slug = _cursor_transcript_for(cid)
        info = scan_cursor_transcript(tpath) if tpath else {
            "model": None, "ctx_tokens": None, "prompt": None,
            "title": None, "ctx_history": []}
        # Headers already store Cursor's own context %; trust that over any
        # transcript inference (transcripts on COOPER carry no usage at all).
        pct = h["ctx_pct"]
        win_label = "-"
        tok = info.get("ctx_tokens")
        model = info.get("model") or "-"
        if pct is not None and tok is None:
            # Reverse the % against the known Cursor default window so TOKENS
            # is an estimate, marked by the window label without a tilde only
            # when we also know the model family.
            window, win_label = window_for(1, model if model != "-" else "composer-2.5")
            tok = int(round(pct / 100.0 * window))
            win_label = _window_label(window)
        elif tok is not None:
            window, win_label = window_for(tok, model if model != "-" else None)
            if pct is None:
                pct = 100.0 * tok / float(window)

        flow = " " * SPARK_LEN
        st = _SPARK.setdefault(
            cid, {"prev": None, "t": 0.0, "hist": deque(maxlen=SPARK_LEN)})
        if tok is not None and now - st["t"] >= SPARK_MIN_STEP:
            if st["prev"] is not None:
                st["hist"].append(max(0, tok - st["prev"]))
            st["prev"] = tok
            st["t"] = now
        flow = spark(st["hist"])

        title = h["name"]
        prompt = info.get("prompt") or h["subtitle"]
        task = title or prompt or ""
        task_src = "title" if title else ("prompt" if prompt else "-")
        short = cid[:8] if len(cid) >= 8 else cid
        # Prefer the branch the composer last touched -- that is the name a
        # human recognises, same as Claude worktree sessions. The composer id
        # short-hash is the fallback when no repo was ever tracked.
        display = ascii_safe(h.get("branch") or "") or ("cursor/" + short)
        cwd = folders.get(h.get("workspace_id") or "") or ""
        if not slug and cwd:
            slug = cursor_project_slug(cwd)
        # Keep the WORKER column short and stable; the full composer name is
        # the TASK text -- same split Claude Code uses (name vs customTitle).
        rows.append(_cursor_worker_row(
            cid, name=display, task=task, task_src=task_src, model=model,
            ctx_tokens=tok, ctx_pct=pct, window=win_label,
            ctx_history=info.get("ctx_history") or [], idle_secs=idle,
            slug=slug, flow=flow, cwd=cwd))

    # Transcript-only fallback: composers with JSONL but no header row (older
    # Cursor builds, or state.vscdb unavailable). When the SQLite index opened
    # cleanly it is authoritative -- rescanning every agent-transcripts file
    # for composers the idle window already dropped was the multi-second stall.
    if _CURSOR_HEADERS_OK or not CURSOR_PROJECTS_DIR.is_dir():
        return _dedupe_cursor_names(rows)
    pattern = str(CURSOR_PROJECTS_DIR / "*" / "agent-transcripts" / "*" / "*")
    for path_str in glob.glob(pattern):
        path = Path(path_str)
        if path.suffix not in (".jsonl", ".txt"):
            continue
        composer_id = path.parent.name
        if composer_id in seen:
            continue
        info = scan_cursor_transcript(str(path))
        last = info["last_write"]
        if last is None:
            continue
        idle = now - last
        if idle > CURSOR_MAX_IDLE_SECS:
            continue
        seen.add(composer_id)
        slug = path.parent.parent.parent.name
        pct = None
        win_label = "-"
        if info["ctx_tokens"]:
            window, win_label = window_for(info["ctx_tokens"], info["model"])
            pct = 100.0 * info["ctx_tokens"] / float(window)
        flow = " " * SPARK_LEN
        st = _SPARK.setdefault(
            composer_id, {"prev": None, "t": 0.0, "hist": deque(maxlen=SPARK_LEN)})
        tok = info["ctx_tokens"]
        if tok is not None and now - st["t"] >= SPARK_MIN_STEP:
            if st["prev"] is not None:
                st["hist"].append(max(0, tok - st["prev"]))
            st["prev"] = tok
            st["t"] = now
        flow = spark(st["hist"])
        short = composer_id[:8] if len(composer_id) >= 8 else composer_id
        rows.append(_cursor_worker_row(
            composer_id, name="cursor/" + short,
            task=info["title"] or info["prompt"] or "",
            task_src=("title" if info["title"] else
                      ("prompt" if info["prompt"] else "-")),
            model=info["model"] or "-", ctx_tokens=info["ctx_tokens"],
            ctx_pct=pct, window=win_label, ctx_history=info["ctx_history"],
            idle_secs=idle, slug=slug, flow=flow))
    return _dedupe_cursor_names(rows)


def _worker_key(w):
    src = w.get("source") or "claude"
    if w.get("pid") is not None:
        return (src, "pid", w["pid"])
    return (src, "sid", w.get("session_id"))


def collect_claude_workers():
    rows = []
    if not SESSIONS_DIR.is_dir():
        return rows
    now = time.time()
    ac_cache = {}
    for f in sorted(SESSIONS_DIR.glob("*.json")):
        try:
            s = json.loads(f.read_text())
        except (OSError, ValueError):
            continue
        pid = s.get("pid")
        if not isinstance(pid, int) or not alive(pid):
            continue
        sid = s.get("sessionId") or ""
        info = scan_transcript(transcript_for(sid))
        pct = None
        win_label = "-"
        if info["ctx_tokens"]:
            window, win_label = window_for(info["ctx_tokens"], info["model"])
            pct = 100.0 * info["ctx_tokens"] / float(window)
        idle = None
        if info["last_write"]:
            idle = now - info["last_write"]

        # Token-flow sample for the FLOW sparkline: growth of the newest turn's
        # context since the last sample is a cheap throughput proxy. Compaction
        # shrinks the context -- that is not negative flow, so it clamps to 0.
        # Throttled so a keypress-forced repaint cannot stuff the history.
        flow = " " * SPARK_LEN
        if sid:
            st = _SPARK.setdefault(
                sid, {"prev": None, "t": 0.0, "hist": deque(maxlen=SPARK_LEN)})
            tok = info["ctx_tokens"]
            if tok is not None and now - st["t"] >= SPARK_MIN_STEP:
                if st["prev"] is not None:
                    st["hist"].append(max(0, tok - st["prev"]))
                st["prev"] = tok
                st["t"] = now
            flow = spark(st["hist"])

        started = s.get("startedAt")
        age = (now - started / 1000.0) if isinstance(started, (int, float)) else None
        cwd = s.get("cwd") or ""
        if cwd not in ac_cache:
            ac_cache[cwd] = auto_compact_enabled(cwd)
        rows.append({
            "start_tokens": startup_context(transcript_for(sid)),
            "source": "claude",
            "name": s.get("name") or "-",
            "pid": pid,
            "session_id": sid,
            "cwd": cwd,
            "project": Path(cwd or ".").name or "-",
            "model": info["model"] or "-",
            "ctx_tokens": info["ctx_tokens"],
            "ctx_pct": pct,
            # Oldest first, at most HISTORY_TURNS long. Emitted by --json too:
            # the series is more use to a script than the rendered delta is.
            "ctx_history": info["ctx_history"],
            "window": win_label,
            # The title Claude Code gave the session; the last prompt is the
            # fallback for sessions too young to have been named yet.
            # Sanitised at the source: this is transcript text, and it lands in
            # a TUI that steers the cursor with escape sequences. An unescaped
            # a raw ESC in a title could clear the screen or repaint the table.
            "task": ascii_safe(info["title"] or info["prompt"] or ""),
            "task_src": "title" if info["title"] else ("prompt" if info["prompt"] else "-"),
            "idle_secs": idle,
            "age_secs": age,
            "flow": flow,
            "auto_compact": ac_cache[cwd],
        })
    return rows


def collect_workers():
    rows = []
    if "claude" in _backends():
        rows.extend(collect_claude_workers())
    rows.extend(collect_cursor_workers())
    return rows


# Services that have answered at least once this run. Down-after-up is a real
# outage (red DOWN); never-up on an unconfigured default port is "off?".
_INFRA_EVER_UP = set()


def collect_infra():
    out = []
    for name, port, path in _services():
        if not port_open(port):
            unseen = name not in _INFRA_EVER_UP and not _PORT_CONFIGURED.get(name)
            out.append({"name": name, "port": port, "up": False,
                        "unseen": unseen,
                        "detail": "never seen on this port" if unseen else "not running"})
            continue
        _INFRA_EVER_UP.add(name)
        detail = ""
        if name == "ollama":
            ps = http_json(port, path)
            models = (ps or {}).get("models") or []
            if models:
                detail = ", ".join(
                    "%s (%.1f GB)" % (m.get("name", "?"), (m.get("size_vram") or m.get("size") or 0) / 1e9)
                    for m in models
                )
            else:
                detail = "no model resident"
        out.append({"name": name, "port": port, "up": True, "unseen": False,
                    "detail": detail})
    return out


# collect_infra blocks on sockets, and a service that is DOWN is the worst
# case: the connect burns its full timeout (0.35 s measured on Windows, where
# a filtered localhost port times out instead of refusing). Inside the render
# loop that stalls every repaint and keypress, so the loop reads a snapshot
# that a daemon thread keeps fresh instead of probing inline.
INFRA_REFRESH_SECONDS = 3.0
_INFRA_LOCK = threading.Lock()
_INFRA_SNAPSHOT = None
_INFRA_THREAD = None


def _infra_worker():
    global _INFRA_SNAPSHOT
    while True:
        # A raise here would kill the daemon thread silently and freeze the
        # INFRA line at its last snapshot forever -- stale data with no tell.
        # Keeping the old snapshot and retrying next cycle is strictly better
        # than dying quietly; on main the same raise was at least visible.
        try:
            snap = collect_infra()
        except Exception:
            snap = None
        if snap is not None:
            with _INFRA_LOCK:
                _INFRA_SNAPSHOT = snap
        time.sleep(INFRA_REFRESH_SECONDS)


def infra_cached():
    """Snapshot view of collect_infra for the render loop.

    The first call probes synchronously so a one-shot frame is never missing
    the INFRA line, then hands refreshing to a daemon thread. Staleness is
    bounded by INFRA_REFRESH_SECONDS plus one probe.

    Live mode sets _INFRA_ALLOW_DEFER so the first paint can show loading dots instead
    of waiting on three localhost connects (filtered ports time out on Windows).
    """
    global _INFRA_SNAPSHOT, _INFRA_THREAD
    with _INFRA_LOCK:
        snap = _INFRA_SNAPSHOT
    if snap is None:
        if _INFRA_ALLOW_DEFER:
            snap = [
                {"name": name, "port": port, "up": None, "detail": ""}
                for name, port, _path in _services()
            ]
        else:
            snap = collect_infra()
            with _INFRA_LOCK:
                _INFRA_SNAPSHOT = snap
    if _INFRA_THREAD is None:
        _INFRA_THREAD = threading.Thread(
            target=_infra_worker, name="infra-probe", daemon=True)
        _INFRA_THREAD.start()
    return snap


def collect_usage_caps():
    """The real 5-hour/weekly/Fable caps, from the claude-usage-scrape cache.

    Returns None if the cache has never been written (task not yet run, or not
    installed on this machine) -- distinct from a stale-but-present cache, which
    render_usage_caps shows with an age instead of hiding.
    """
    try:
        with open(USAGE_CACHE, "r", encoding="utf-8") as fh:
            data = json.load(fh)
    except (OSError, ValueError):
        return None
    last_success = data.get("last_success_epoch")
    data["age_secs"] = (time.time() - last_success) if last_success else None
    try:
        # utf-8-sig: the writer is PowerShell 5.1 Out-File, which prepends a BOM.
        with open(CREDITS_CACHE, "r", encoding="utf-8-sig") as fh:
            data["openrouter_credits"] = json.load(fh)
    except (OSError, ValueError):
        pass  # credits are an optional extra on the caps line, never load-bearing
    data["burn"] = _burn_rates()
    return data


def _burn_rates(window_hours=48):
    """Trailing burn rate in pct/hour per weekly cap, from history.jsonl.

    The runway view ("~2.1d left") needs this; the even-burn pace delta does
    not. Rows after the most recent RESET (a pct drop) are the only valid
    sample -- mixing across a reset would average in a cliff. Returns {} until
    there are >=3 post-reset points spanning >=6h, so a fresh install or a
    just-reset week quietly falls back to the pace display.
    """
    try:
        with open(USAGE_HISTORY, "r", encoding="utf-8-sig") as fh:
            rows = [json.loads(ln) for ln in fh if ln.strip()]
    except (OSError, ValueError):
        return {}
    cutoff = time.time() - window_hours * 3600
    rows = [r for r in rows if r.get("epoch", 0) >= cutoff]
    out = {}
    for key in ("weekly", "fable"):
        pts = [(r["epoch"], r[key]) for r in rows
               if isinstance(r.get(key), (int, float))]
        # Drop everything before the most recent reset (pct decrease).
        for i in range(len(pts) - 1, 0, -1):
            if pts[i][1] < pts[i - 1][1]:
                pts = pts[i:]
                break
        if len(pts) < 3:
            continue
        span_h = (pts[-1][0] - pts[0][0]) / 3600.0
        if span_h < 6:
            continue
        rate = (pts[-1][1] - pts[0][1]) / span_h
        if rate > 0:
            out[key] = rate
    return out


def _iso_to_epoch(ts):
    """Ollama's expires_at is RFC3339 with an offset, e.g.
    '2026-08-01T15:04:05.123456-07:00'. Any format surprise just drops the
    unload timer -- it is not worth losing the whole panel over."""
    if not ts:
        return None
    try:
        return datetime.fromisoformat(ts).timestamp()
    except Exception:
        return None


def collect_local_models():
    """Installed and resident Ollama models, merged from /api/tags and /api/ps.

    /api/ps alone -- what the INFRA line uses -- only sees what is currently in
    VRAM, so a model that is installed but idle is invisible there. This is the
    fuller picture: everything `ollama list` knows about, with residency and a
    VRAM figure layered on top for whichever of those happen to be loaded.
    """
    if not port_open(OLLAMA_PORT):
        return []
    tags = http_json(OLLAMA_PORT, "/api/tags") or {}
    ps = http_json(OLLAMA_PORT, "/api/ps") or {}
    resident = {m.get("name"): m for m in (ps.get("models") or [])}
    out = []
    for m in tags.get("models") or []:
        name = m.get("name", "?")
        r = resident.get(name)
        expires_secs = None
        if r and r.get("expires_at"):
            epoch = _iso_to_epoch(r["expires_at"])
            if epoch is not None:
                expires_secs = max(0, epoch - time.time())
        out.append({
            "name": name,
            "disk_gb": (m.get("size") or 0) / 1e9,
            "resident": r is not None,
            "vram_gb": ((r.get("size_vram") or r.get("size") or 0) / 1e9) if r else None,
            "expires_secs": expires_secs,
        })
    return out


def _yaml_scalar(value):
    v = (value or "").strip()
    if len(v) >= 2 and v[0] == v[-1] and v[0] in "\"'":
        return v[1:-1]
    return v


def parse_litellm_model_list(text):
    """Pull model_name, model, and api_base out of a LiteLLM config.yaml.

    Everything else -- api_key, drop_params, the rest of litellm_params -- is
    ignored on purpose. Aliases are never invented: an entry without
    model_name is skipped.
    """
    if not text:
        return []
    lines = text.splitlines()
    start = None
    for i, ln in enumerate(lines):
        stripped = ln.split("#", 1)[0].rstrip()
        if stripped == "model_list:":
            start = i + 1
            break
        if stripped.startswith("model_list:"):
            rest = stripped[len("model_list:"):].strip()
            if rest in ("[]", "null", "~"):
                return []
            start = i + 1
            break
    if start is None:
        return []

    items = []
    current = None
    keep = ("model_name", "model", "api_base")

    def flush():
        nonlocal current
        if current and current.get("model_name"):
            items.append({
                "model_name": current["model_name"],
                "model": current.get("model") or "",
                "api_base": current.get("api_base") or "",
            })
        current = None

    for ln in lines[start:]:
        cut = ln.split("#", 1)[0].rstrip()
        if not cut.strip():
            continue
        indent = len(ln) - len(ln.lstrip(" "))
        if indent == 0 and not cut.lstrip().startswith("-"):
            break
        stripped = cut.strip()
        if stripped.startswith("-"):
            flush()
            current = {}
            rest = stripped[1:].strip()
            if rest and ":" in rest:
                key, val = rest.split(":", 1)
                key = key.strip()
                if key in keep:
                    parsed = _yaml_scalar(val)
                    if parsed:
                        current[key] = parsed
            continue
        if current is None or ":" not in stripped:
            continue
        key, val = stripped.split(":", 1)
        key = key.strip()
        if key in keep:
            parsed = _yaml_scalar(val)
            if parsed:
                current[key] = parsed
    flush()
    return items


def _litellm_config_path():
    return Path(os.environ.get(
        LITELLM_CONFIG_ENV, str(HOME / "litellm-server" / "config.yaml")))


def _ollama_backing(model):
    for prefix in ("ollama_chat/", "ollama/"):
        if model.startswith(prefix):
            return model[len(prefix):]
    return None


def _ollama_name_installed(name, installed):
    if not name or not installed:
        return False
    if name in installed:
        return True
    if (name + ":latest") in installed:
        return True
    if name.endswith(":latest") and name[:-7] in installed:
        return True
    return False


def collect_gateway_models(installed_names=None):
    """Configured LiteLLM aliases from config.yaml, plus an Ollama cross-check.

    This is configured intent, not liveness. Missing config is a labelled gap.
    """
    path = _litellm_config_path()
    if not path.is_file():
        return [], "no LiteLLM config at %s" % path
    try:
        text = path.read_text(encoding="utf-8")
    except OSError:
        return [], "LiteLLM config unreadable at %s" % path
    entries = parse_litellm_model_list(text)
    if not entries:
        return [], "no model_list in LiteLLM config"
    rows = []
    for e in entries:
        backing = e["model"]
        ollama = _ollama_backing(backing)
        kind = "local" if ollama else "cloud"
        if kind == "cloud":
            ollama_state = "-"
        elif installed_names is None:
            ollama_state = "unknown"
        elif _ollama_name_installed(ollama, installed_names):
            ollama_state = "installed"
        else:
            ollama_state = "missing"
        rows.append({
            "alias": e["model_name"],
            "model": backing,
            "api_base": e["api_base"],
            "kind": kind,
            "ollama_name": ollama,
            "ollama": ollama_state,
        })
    return rows, None


def _batch_root():
    return Path(os.environ.get(BATCH_DIR_ENV,
                               str(HOME / "litellm-server" / "batch")))


def _jobs_root():
    return Path(os.environ.get(JOBS_DIR_ENV, str(HOME / "jobs")))


def _proxy_log_activity(path):
    """Last-request age and requests/min, read off the tail of proxy.log.

    Best-effort by design: the log format is LiteLLM's to change, so anything
    that fails to parse just drops these two numbers rather than the panel.
    Only timestamped lines mentioning a completions route count as requests.
    """
    lines = read_tail(str(path), PROXY_LOG_TAIL)
    stamp = re.compile(r"(\d{4}-\d{2}-\d{2})[ T](\d{2}:\d{2}:\d{2})")
    now = time.time()
    last = None
    recent = 0
    for line in lines:
        if "completion" not in line and "chat" not in line:
            continue
        m = stamp.search(line)
        if not m:
            continue
        try:
            t = time.mktime(time.strptime(
                m.group(1) + " " + m.group(2), "%Y-%m-%d %H:%M:%S"))
        except ValueError:
            continue
        last = t if last is None else max(last, t)
        if now - t <= 60:
            recent += 1
    return {"last_req_secs": (now - last) if last is not None else None,
            "req_per_min": recent if last is not None else None}


def collect_gateway():
    """LiteLLM liveliness plus batch-run progress, derived from files.

    The gateway is DB-less, so its activity endpoints all 400 -- but the batch
    pipeline is resumable by construction (one output file per finished item),
    which means progress is fully derivable from the filesystem with zero
    cooperation from the running process. _run.json, when extract.py wrote one,
    pins the worklist and model; without it the run still shows, just with less.
    """
    now = time.time()
    out = {"litellm_up": port_open(LITELLM_PORT), "runs": [], "jobs": None,
           "last_req_secs": None, "req_per_min": None,
           "configured": [], "configured_gap": None,
           "probed_at": now}

    configured, gap = collect_gateway_models(None)
    if configured and any(r["kind"] == "local" for r in configured):
        names = {m["name"] for m in collect_local_models() if m.get("name")}
        configured, gap = collect_gateway_models(names)
    out["configured"] = configured
    out["configured_gap"] = gap

    root = _batch_root()
    log = root.parent / "proxy.log"
    if log.exists():
        out.update(_proxy_log_activity(log))

    if root.is_dir():
        for d in sorted(root.iterdir()):
            if not d.is_dir():
                continue
            try:
                outputs = [p for p in d.glob("*.json")
                           if not p.name.startswith("_")]
                failures = d / "_failures.jsonl"
                failed = 0
                if failures.exists():
                    failed = sum(
                        1 for ln in failures.read_text(
                            encoding="utf-8", errors="replace").splitlines()
                        if ln.strip())
                meta = {}
                rj = d / "_run.json"
                if rj.exists():
                    try:
                        meta = json.loads(rj.read_text(encoding="utf-8"))
                    except ValueError:
                        meta = {}
                # A dir of loose JSON is not automatically a run: schemas/ and
                # the like would otherwise show up as one. Without a _run.json
                # only the results-* naming convention identifies a run.
                if not meta and not failed and not d.name.startswith("results"):
                    continue
                if not outputs and not failed and not meta:
                    continue  # empty results dir, nothing to say yet

                mtimes = sorted(p.stat().st_mtime for p in outputs)
                newest = mtimes[-1] if mtimes else None
                # Rate from the spread of the newest outputs rather than from
                # the start time: a resumed run's start says nothing about the
                # pace it is writing at now.
                rate_hr = None
                recent = mtimes[-50:]
                if len(recent) >= 2 and recent[-1] > recent[0]:
                    rate_hr = (len(recent) - 1) / (recent[-1] - recent[0]) * 3600.0
                total = meta.get("total")
                eta_secs = None
                if rate_hr and isinstance(total, int):
                    remaining = total - len(outputs) - failed
                    if remaining > 0:
                        eta_secs = remaining / rate_hr * 3600.0
                worklist = meta.get("worklist")
                out["runs"].append({
                    "name": d.name,
                    "model": meta.get("model"),
                    "worklist": Path(worklist).name if worklist else None,
                    "done": len(outputs),
                    "total": total,
                    "failed": failed,
                    "rate_hr": rate_hr,
                    "eta_secs": eta_secs,
                    "last_write_secs": (now - newest) if newest else None,
                    "active": newest is not None
                              and (now - newest) <= BATCH_ACTIVE_SECS,
                })
            except OSError:
                continue
        out["runs"].sort(key=lambda r: (not r["active"],
                                        r["last_write_secs"] or 1e12))

    jobs = _jobs_root()
    if jobs.is_dir():
        depth = {}
        for state in ("inbox", "running", "done", "failed"):
            sub = jobs / state
            if state in ("done", "failed"):
                depth[state] = sum(1 for p in sub.iterdir()
                                   if p.is_dir()) if sub.is_dir() else 0
            else:
                depth[state] = len(list(sub.glob("*.json"))) if sub.is_dir() else 0
        out["jobs"] = depth
    return out


def render_gateway(gw):
    """LiteLLM plus everything it has been fed, without asking it anything --
    a DB-less gateway keeps no history, so the batch pipeline's own output
    files carry the progress story."""
    probed = gw.get("probed_at")
    extra = ("probed " + time.strftime("%H:%M:%S", time.localtime(probed))
             if probed is not None else None)
    lines = []
    mark = (c(GLYPHS["ok"] + "up", GREEN) if gw["litellm_up"]
            else c(GLYPHS["fail"] + "DOWN", BOLD, RED))
    head = "  litellm %s (127.0.0.1:%d)" % (mark, LITELLM_PORT)
    if gw["last_req_secs"] is not None:
        head += "   last request %s ago" % dur(gw["last_req_secs"])
    if gw["req_per_min"] is not None:
        head += "   %d req/min" % gw["req_per_min"]
    lines.append(head)
    if gw["jobs"] is not None:
        j = gw["jobs"]
        inbox = str(j["inbox"])
        lines.append("  jobs queue: inbox %s  running %s  done %s  failed %s" % (
            c(inbox, BOLD, YELLOW) if j["inbox"] else inbox,
            c(str(j["running"]), GREEN) if j["running"] else j["running"],
            j["done"],
            c(str(j["failed"]), RED) if j["failed"] else j["failed"]))

    gap = gw.get("configured_gap")
    configured = gw.get("configured") or []
    lines.append(c("  configured models (intent, not liveness)", DIM))
    if gap:
        lines.append(c("  " + gap, YELLOW))
    elif configured:
        cols = [
            ("ALIAS", lambda r: r["alias"]),
            ("BACKING", lambda r: r["model"] or "-"),
            ("WHERE", lambda r: r["kind"]),
            ("OLLAMA", lambda r: r["ollama"]),
        ]
        table = [[h for h, _ in cols]] + [[f(r) for _, f in cols] for r in configured]
        w = [max(len(row[i]) for row in table) for i in range(len(cols))]
        lines.append("  " + c("  ".join(
            table[0][i].ljust(w[i]) for i in range(len(cols))), BOLD))
        for row, r in zip(table[1:], configured):
            cells = [row[i].ljust(w[i]) for i in range(len(cols))]
            line = "  " + "  ".join(cells)
            if r["ollama"] == "missing":
                lines.append(c(line, BOLD, YELLOW))
            elif r["kind"] == "cloud":
                lines.append(c(line, CYAN))
            else:
                lines.append(line)
        missing = sum(1 for r in configured if r["ollama"] == "missing")
        lines.append("  " + c("%d alias(es), %d missing locally" % (
            len(configured), missing), DIM))

    if not gw["runs"]:
        lines.append(c("  no batch runs found", DIM))
        return panel("GATEWAY", lines, extra=extra)

    cols = [
        ("BATCH RUN", lambda r: r["name"]),
        ("MODEL", lambda r: r["model"] or "?"),
        ("DONE/TOTAL", lambda r: "%d/%s" % (
            r["done"], r["total"] if r["total"] is not None else "?")),
        ("FAIL", lambda r: str(r["failed"])),
        ("RATE", lambda r: "-" if not r["rate_hr"] else "%d/hr" % round(r["rate_hr"])),
        ("ETA", lambda r: dur(r["eta_secs"]) if r["eta_secs"]
            else ("done" if r["total"] is not None
                  and r["done"] + r["failed"] >= r["total"] else "-")),
        ("LAST WRITE", lambda r: "-" if r["last_write_secs"] is None
            else dur(r["last_write_secs"]) + " ago"),
    ]
    table = [[h for h, _ in cols]] + [[f(r) for _, f in cols] for r in gw["runs"]]
    w = [max(len(row[i]) for row in table) for i in range(len(cols))]
    lines.append("  " + c("  ".join(
        table[0][i].ljust(w[i]) for i in range(len(cols))), BOLD))
    for row, r in zip(table[1:], gw["runs"]):
        cells = [row[i].ljust(w[i]) for i in range(len(cols))]
        # Same colouring rule as LOCAL MODELS: green while actively writing,
        # dim once done or stale.
        line = "  " + "  ".join(cells)
        lines.append(c(line, GREEN) if r["active"] else c(line, DIM))
    active = sum(1 for r in gw["runs"] if r["active"])
    lines.append("  " + c("%d run(s), %d active" % (len(gw["runs"]), active), DIM))
    return panel("GATEWAY", lines, extra=extra)


# host -> {"data", "t", "err", "thread"}. Fetches run on daemon threads so a
# host behind a closed MacBook lid shows as stale rather than hanging the UI.
_REMOTE = {}
_REMOTE_LOCK = threading.Lock()


def _fetch_remote(host):
    cmd = os.environ.get(REMOTE_CMD_ENV, REMOTE_CMD_DEFAULT)
    err, data = None, None
    try:
        p = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=5", host, cmd],
            capture_output=True, text=True, timeout=REMOTE_TIMEOUT_SECS)
        if p.returncode == 0:
            data = json.loads(p.stdout)
        else:
            tail = (p.stderr or "").strip().splitlines()
            err = tail[-1][:60] if tail else "exit %d" % p.returncode
    except subprocess.TimeoutExpired:
        err = "timeout after %ds" % REMOTE_TIMEOUT_SECS
    except (OSError, ValueError) as e:
        err = str(e)[:60]
    with _REMOTE_LOCK:
        st = _REMOTE.setdefault(host, {})
        if data is not None:
            st["data"] = data
            st["t"] = time.time()
            st["err"] = None
        else:
            st["err"] = err  # keep the last good data and its timestamp
        st["thread"] = None


def collect_remote():
    hosts = [h.strip() for h in
             os.environ.get(REMOTES_ENV, "hyrule").split(",") if h.strip()]
    rows = []
    now = time.time()
    for host in hosts:
        with _REMOTE_LOCK:
            st = _REMOTE.setdefault(host, {})
            age = (now - st["t"]) if st.get("t") else None
            if st.get("thread") is None and (age is None
                                             or age >= REMOTE_REFRESH_SECS):
                t = threading.Thread(target=_fetch_remote, args=(host,),
                                     daemon=True)
                st["thread"] = t
                t.start()
                first_try = not st.get("t") and not st.get("err")
            else:
                t, first_try = None, False
        # Only the very first attempt per host gets a blocking grace period --
        # it is what makes --once useful. A host that already failed once (the
        # closed-lid case) never blocks again; its row shows the error instead.
        if t is not None and first_try:
            t.join(REMOTE_TIMEOUT_SECS + 2)
        with _REMOTE_LOCK:
            st = _REMOTE[host]
            age = (time.time() - st["t"]) if st.get("t") else None
            rows.append({"host": host, "data": st.get("data"),
                         "age_secs": age, "err": st.get("err"),
                         "fetching": st.get("thread") is not None})
    return rows


def render_remote(remotes):
    """One summary row per remote host, rendered from that host's own
    `roost --json` over ssh. Data is cached: a host that stops answering keeps
    its last good row with the age saying how old it is."""
    if not remotes:
        return panel("REMOTE", [
            c("  set %s (comma-separated ssh aliases)" % REMOTES_ENV, DIM)])
    lines = []

    cols = [
        ("HOST", lambda r: r["host"]),
        ("WORKERS", lambda r: r["nworkers"]),
        ("WORKING", lambda r: r["working"]),
        ("RESIDENT MODELS", lambda r: r["resident"]),
        ("BATCH", lambda r: r["batch"]),
        ("JOBS", lambda r: r["jobs"]),
        ("AGE", lambda r: r["age"]),
    ]
    view = []
    for r in remotes:
        d = r["data"]
        if d is None:
            view.append({"host": r["host"], "nworkers": "-", "working": "-",
                         "resident": "-", "batch": "-", "jobs": "-",
                         "age": "fetching" + GLYPHS["ell"] if r["fetching"]
                                else (r["err"] or "-"), "stale": True})
            continue
        workers = d.get("workers") or []
        working = sum(1 for w in workers
                      if w.get("idle_secs") is not None and w["idle_secs"] < 60)
        resident = ", ".join(m["name"] for m in (d.get("local_models") or [])
                             if m.get("resident")) or "-"
        gw = d.get("gateway") or {}
        runs = [x for x in (gw.get("runs") or []) if x.get("active")]
        if runs:
            batch = ", ".join("%s %d/%s" % (
                x["name"], x["done"],
                x["total"] if x.get("total") is not None else "?")
                for x in runs)
        else:
            batch = "-"
        j = gw.get("jobs")
        jobs = ("in %d run %d fail %d" % (j["inbox"], j["running"], j["failed"])
                if j else "-")
        age = dur(r["age_secs"]) if r["age_secs"] is not None else "-"
        stale = r["age_secs"] is not None and r["age_secs"] > 3 * REMOTE_REFRESH_SECS
        if r["err"]:
            age += " (%s)" % r["err"]
        elif stale:
            age += " (stale)"
        view.append({"host": r["host"], "nworkers": str(len(workers)),
                     "working": str(working), "resident": resident,
                     "batch": batch, "jobs": jobs, "age": age, "stale": stale})

    table = [[h for h, _ in cols]] + [[f(r) for _, f in cols] for r in view]
    w = [max(len(row[i]) for row in table) for i in range(len(cols))]
    lines.append("  " + c("  ".join(
        table[0][i].ljust(w[i]) for i in range(len(cols))), BOLD))
    for row, r in zip(table[1:], view):
        cells = []
        for i, (header, _) in enumerate(cols):
            txt = row[i].ljust(w[i])
            if header == "HOST":
                cells.append(c(txt, BOLD))
            elif header == "WORKING" and not r["stale"] and row[i] not in ("0", "-"):
                cells.append(c(txt, GREEN))
            elif r["stale"] or header == "AGE":
                cells.append(c(txt, DIM))
            else:
                cells.append(txt)
        lines.append("  " + "  ".join(cells))
    return panel("REMOTE", lines)


def parse_budget(s):
    """'60M', '850k', or a plain token count. None on unset or garbage."""
    if not s:
        return None
    s = s.strip()
    try:
        if s and s[-1] in "kK":
            return int(float(s[:-1]) * 1000)
        if s and s[-1] in "mM":
            return int(float(s[:-1]) * 1000000)
        return int(float(s))
    except ValueError:
        return None


def _tally_lines(lines, counts):
    """Accumulate (day, model) -> input+output tokens from assistant turns.

    Cache reads/creation are deliberately excluded: they are billed and
    rate-limited differently, and counting them would swamp the number with
    re-reads of unchanged context. What is left is closest to "work done".
    """
    for line in lines:
        if '"usage"' not in line or '"assistant"' not in line:
            continue
        try:
            d = json.loads(line)
        except ValueError:
            continue
        msg = d.get("message") or {}
        usage = msg.get("usage") or {}
        if not usage:
            continue
        day = str(d.get("timestamp") or "")[:10]
        if len(day) != 10:
            continue
        # Raw model name kept: the "claude-" prefix is what later separates
        # cloud burn (counts against the plan) from local models (free).
        # Sanitised: it is transcript text headed for the TUI.
        model = ascii_safe(msg.get("model") or "?") or "?"
        tok = (usage.get("input_tokens") or 0) + (usage.get("output_tokens") or 0)
        key = (day, model)
        counts[key] = counts.get(key, 0) + tok


def collect_usage():
    """Observed tokens per day per model over the last USAGE_DAYS.

    Incremental: each transcript is read in full exactly once per roost run
    (only files touched inside the window), then only appended bytes after
    that. Only called while the USAGE panel is open, so the one full pass
    happens on the first `u`, not at launch.
    """
    cutoff = time.time() - USAGE_DAYS * 86400
    paths = set(glob.glob(str(PROJECTS_DIR / "*" / "*.jsonl")))
    paths.update(glob.glob(
        str(PROJECTS_DIR / "*" / "*" / "subagents" / "agent-*.jsonl")))

    # A deleted transcript never reappears in the glob, so its entry would sit
    # in the cache forever -- still counted into the panel, and a slow leak.
    for path in [p for p in _USAGE_CACHE if p not in paths]:
        del _USAGE_CACHE[path]

    for path in paths:
        try:
            mtime = os.path.getmtime(path)
            size = os.path.getsize(path)
        except OSError:
            _USAGE_CACHE.pop(path, None)  # deleted between glob and stat
            continue
        if mtime < cutoff:
            _USAGE_CACHE.pop(path, None)
            continue
        st = _USAGE_CACHE.get(path)
        if st and st["size"] == size and st["mtime"] == mtime:
            continue
        if st and size >= st["size"]:
            pos, counts = st["size"], st["counts"]
        else:
            pos, counts = 0, {}  # new file, or it shrank -- start over
        try:
            with open(path, "rb") as fh:
                fh.seek(pos)
                data = fh.read()
        except OSError:
            continue
        # Consume only whole lines; a half-written trailing line is left for
        # the next pass rather than being parsed as garbage and lost.
        cut = data.rfind(b"\n") + 1
        _tally_lines(data[:cut].decode("utf-8", "replace").splitlines(), counts)
        _USAGE_CACHE[path] = {"mtime": mtime, "size": pos + cut, "counts": counts}

    days = {}
    for st in _USAGE_CACHE.values():
        for (day, model), tok in st["counts"].items():
            byday = days.setdefault(day, {})
            byday[model] = byday.get(model, 0) + tok
    return days


def render_usage(days):
    extra = ("observed transcript tokens (input+output) -- an estimate, "
             "not the Anthropic meter")
    lines = []
    # Day keys come from transcript timestamps, which are UTC -- so the day
    # boundary is UTC too. Keep only the window and newest first.
    recent = sorted(days, reverse=True)[:USAGE_DAYS]
    if not recent:
        lines.append(c("  nothing recorded in the last %d days" % USAGE_DAYS, DIM))
        return panel("USAGE", lines, extra=extra)

    today = time.strftime("%Y-%m-%d", time.gmtime())
    rows = []
    for day in recent:
        by_model = days[day]
        # Cloud burn is what counts against the plan; local Ollama models cost
        # nothing, so they show in the breakdown but not the budget math.
        cloud = sum(t for m, t in by_model.items() if m.startswith("claude-"))
        top = sorted(by_model.items(), key=lambda kv: -kv[1])
        detail = ", ".join(
            "%s %s" % (m.replace("claude-", "") if m.startswith("claude-")
                       else m + " (local)", compact(t))
            for m, t in top[:3])
        if len(top) > 3:
            detail += ", +%d more" % (len(top) - 3)
        rows.append((day, cloud, detail))

    wid = max(len(compact(t)) for _, t, _ in rows)
    for day, cloud, detail in rows:
        mark = c(" <- today", GREEN) if day == today else ""
        lines.append("  %s  %s  %s%s" % (
            c(day, BOLD if day == today else DIM),
            compact(cloud).rjust(wid), c(detail, DIM), mark))

    week_total = sum(t for _, t, _ in rows)
    today_cloud = sum(t for m, t in days.get(today, {}).items()
                      if m.startswith("claude-"))
    summary = "today %s  |  %dd %s cloud" % (
        compact(today_cloud), len(rows), compact(week_total))
    budget = parse_budget(os.environ.get(USAGE_BUDGET_ENV))
    lines.append("")
    if budget:
        pct = 100.0 * week_total / float(budget)
        code = (BOLD, RED) if pct >= 80 else ((YELLOW,) if pct >= 50 else (GREEN,))
        lines.append("  " + c(summary, BOLD) + "  "
                     + c("/ %s budget (%.0f%%)" % (compact(budget), pct), *code))
    else:
        lines.append("  " + c(summary, BOLD))
        lines.append("  " + c("set %s (e.g. 60M) to measure against your plan -- "
                              "calibrate the number from /usage once" % USAGE_BUDGET_ENV,
                              DIM))
    return panel("USAGE", lines, extra=extra)


# ---- advisory thresholds ----------------------------------------------------
# A turn costs whatever the context currently holds, so a fat session is
# expensive on every future turn, not once. These are the lines where that stops
# being worth paying.
EXPENSIVE_TOKENS = 150000   # per-turn cost above which a fresh session is cheaper
PARKED_IDLE_HOURS = 2       # untouched this long and still fat == parked
STALE_IDLE_HOURS = 6        # untouched this long and small == just clutter
NEAR_LIMIT_PCT = 80
ADVICE_TASK_WIDTH = 52  # task text in ADVICE; the detail lives on the next line
TYPICAL_BASELINE = 50000    # measured: what a fresh session starts at here
# -----------------------------------------------------------------------------


def advise(workers):
    """Concrete actions, ordered by how many tokens they save."""
    out = []
    ranked = sorted(workers, key=lambda r: -(r["ctx_tokens"] or 0))

    for r in ranked:
        tok = r["ctx_tokens"] or 0
        idle_h = (r["idle_secs"] or 0) / 3600.0
        pct = r["ctx_pct"] or 0
        # Cursor composers have no pid; the worker name already carries a short
        # composer id (cursor/<8hex>). Formatting pid with %d used to crash the
        # whole TUI the moment ADVICE opened on a mixed fleet.
        pid = r.get("pid")
        tag = ("%s (pid %d)" % (r["name"], pid)) if pid is not None else r["name"]
        # The pid alone does not tell you what you would be closing. The task is
        # what makes the call obvious -- "audit the build scripts" is easy to
        # abandon, "migrate the database" is not.
        task = ascii_safe(r.get("task") or "")
        if len(task) > ADVICE_TASK_WIDTH:
            ell = GLYPHS["ell"]
            task = task[:ADVICE_TASK_WIDTH - len(ell)] + ell
        saving = tok - TYPICAL_BASELINE

        if tok >= EXPENSIVE_TOKENS and idle_h >= PARKED_IDLE_HOURS:
            out.append((saving, c("PARKED+COSTLY", BOLD, RED), tag, task,
                        "idle %.1fh holding %s tokens. Resuming costs that much on the "
                        "FIRST turn. Start a fresh session instead (~%s) and save ~%s per turn."
                        % (idle_h, "{:,}".format(tok), "{:,}".format(TYPICAL_BASELINE),
                           "{:,}".format(saving))))
        elif pct >= NEAR_LIMIT_PCT:
            if r.get("auto_compact", True):
                detail = ("at %.0f%% of its window. Wrap up or /compact before it "
                           "auto-compacts mid-task." % pct)
            else:
                detail = ("at %.0f%% of its window with auto-compact off for this "
                           "session -- there is no safety net here. Wrap up or "
                           "/compact now, or it errors out instead of compacting."
                           % pct)
            out.append((saving, c("NEAR LIMIT", BOLD, YELLOW), tag, task, detail))
        elif tok >= EXPENSIVE_TOKENS:
            out.append((saving, c("EXPENSIVE", YELLOW), tag, task,
                        "every turn now reprocesses %s tokens. Fine to finish the "
                        "current task in; do not start an unrelated one here."
                        % "{:,}".format(tok)))
        elif idle_h >= STALE_IDLE_HOURS:
            out.append((0, c("STALE", DIM), tag, task,
                        "idle %.1fh at %.0f%%. Costs nothing while it sits, but it hides "
                        "the sessions that matter -- close it." % (idle_h, pct)))

    if not out:
        return panel(
            "ADVICE",
            ["  nothing to act on -- no parked, oversized, or stale sessions"])
    lines = []
    for _, label, tag, task, text in sorted(out, key=lambda x: -x[0]):
        head = "  %s  %s" % (label, c(tag, BOLD))
        if task:
            head += "  " + c(task, DIM)
        lines.append(head)
        lines.append("      %s" % text)

    total = sum(r["ctx_tokens"] or 0 for r in workers)
    ideal = TYPICAL_BASELINE * len(workers)
    if total > ideal:
        lines.append("")
        lines.append("  One turn in each of these %d sessions reprocesses %s tokens. The "
                     "same %d sessions started fresh would cost %s -- a %.1fx difference."
                     % (len(workers), c("{:,}".format(total), BOLD), len(workers),
                        "{:,}".format(ideal), total / float(ideal)))
    return panel("ADVICE", lines)


def growth(history):
    """Context added across the retained turns -- the TREND cell.

    A signed total, not a shape. Context inside a session only ever rises: it
    falls solely on /compact. So a sparkline *of context* draws the same
    monotonic ramp for every row and a rise/fall arrow points up on every row,
    while the amount separates a session creeping by 2k a window from one
    adding 21k. FLOW next door is a sparkline of throughput, not of context --
    a different series, which is why it is worth a shape and this is not.
    """
    if not history or len(history) < 2:
        return "-"
    d = history[-1] - history[0]
    if d == 0:
        return "="
    return ("+" if d > 0 else "-") + compact(abs(d))


def dur(secs):
    """One-unit age: 45s, 12m, 3h, 2d (the charter's text convention).

    The largest unit that fits, truncated -- 90 seconds is "1m", 47 hours is
    "1d". Two-unit forms like "47h59m" bought false precision at the cost of
    width and a visible cliff where they gave way to days.
    """
    if secs is None:
        return "-"
    secs = int(secs)
    if secs < 60:
        return "%ds" % secs
    if secs < 3600:
        return "%dm" % (secs // 60)
    if secs < 86400:
        return "%dh" % (secs // 3600)
    return "%dd" % (secs // 86400)


def compact(n):
    """484k, not 484,030 -- exact digits cost width and buy nothing here."""
    if n is None:
        return "-"
    if n >= 1000000:
        return "%.1fM" % (n / 1000000.0)
    if n >= 1000:
        return "%dk" % (n // 1000)
    return str(n)


def spark(hist):
    """ASCII sparkline of recent token flow, newest sample at the right.

    Normalised to the buffer's own max, so it shows the *shape* of activity --
    bursts and quiet stretches -- not absolute volume. Left-padded: history
    grows in from the right as samples arrive.
    """
    if not hist:
        return " " * SPARK_LEN
    mx = max(hist)
    out = []
    for v in hist:
        if v <= 0 or mx <= 0:
            out.append(SPARK_RAMP[0])
        else:
            idx = 1 + int((v / float(mx)) * (len(SPARK_RAMP) - 2))
            out.append(SPARK_RAMP[min(idx, len(SPARK_RAMP) - 1)])
    return "".join(out).rjust(SPARK_LEN)


BAR_WIDTH = 12


def bar(pct):
    """A filled bar for context use.

    Carries what a WIN column used to: a short bar beside a large token count
    reads as "big window, room to spare" without a separate column saying so.
    ASCII in both dialects on purpose: the bar is a data texture, not glyph
    vocabulary, and [###---] is legible everywhere including the
    legacy-codepage Windows console.
    """
    if pct is None:
        return "[" + " " * BAR_WIDTH + "]"
    filled = int(round(min(pct, 100.0) / 100.0 * BAR_WIDTH))
    return "[" + "#" * filled + "-" * (BAR_WIDTH - filled) + "]"


def bucket(w):
    """Which attention group a session belongs in, most actionable first.

    Ordering is by what it costs to ignore, not by size: a session at 85% is
    about to stop working, a fat parked one bills its whole context on the next
    turn, and everything quiet is noise until it is not.
    """
    pct = w["ctx_pct"] or 0
    tok = w["ctx_tokens"] or 0
    idle = w["idle_secs"]
    if w["ctx_tokens"] is None:
        # No usage yet. A brand-new unknown stays STARTING; once it has been
        # idle for a minute it is noise -- Cursor composers often never grow a
        # usage block at all, and without this they flood STARTING forever
        # (never reaching QUIET collapse), burying the ranked board and every
        # panel that paints below it.
        if idle is not None and idle >= 60:
            return 4, "QUIET"
        return 3, "STARTING"
    if pct >= NEAR_LIMIT_PCT:
        return 0, "NEAR LIMIT"
    if tok > EXPENSIVE_TOKENS and (idle or 0) > PARKED_IDLE_HOURS * 3600:
        return 1, "PARKED + COSTLY"
    if idle is not None and idle < 60:
        return 2, "WORKING NOW"
    return 4, "QUIET"


BUCKET_COLORS = {0: RED, 1: YELLOW, 2: GREEN, 3: DIM, 4: DIM}


def arrange(workers, expand_quiet=False):
    """Split workers into table rows and the collapsed QUIET tail.

    Pulled out of render() so the cursor and the screen agree by construction.
    Two orderings computed separately would eventually disagree, and the failure
    mode of that disagreement is stopping the wrong session.

    QUIET expands under the cursor because that group is precisely what the
    sweep is for: a session idle for hours is invisible in the collapsed line,
    and unreachable if the cursor cannot enter it.
    """
    tagged = [(bucket(w), w) for w in workers]
    shown = [(b, w) for (b, w) in tagged if b[0] != 4 or expand_quiet]
    quiet = [] if expand_quiet else [w for (b, w) in tagged if b[0] == 4]
    # Within a group, order by what a turn costs. Percentage buries the
    # expensive sessions: 484k tokens on the 1M window reads as a mild 48%,
    # while 140k on a 200k window looks alarming at 70% and costs a third as much.
    shown.sort(key=lambda t: (t[0][0], -(t[1]["ctx_tokens"] or 0)))
    return shown, quiet


def render(workers, sel=None):
    """Table for `workers`. `sel` is an index into arrange()'s shown rows."""
    lines = []
    if not workers:
        lines.append("no live agent sessions")
        return lines

    mixed = any(w.get("source") not in (None, "claude") for w in workers)
    cols = []
    if mixed:
        cols.append(("SRC", lambda r: (r.get("source") or "claude")[:6]))
    cols.extend([
        ("WORKER", lambda r: r["name"]),
        ("MODEL", lambda r: (r["model"] or "-").replace("claude-", "")),
        ("CONTEXT", lambda r: bar(r["ctx_pct"])),
        ("CTX", lambda r: "-" if r["ctx_pct"] is None else "%.0f%%" % r["ctx_pct"]),
        ("TOKENS", lambda r: "-" if not r["ctx_tokens"] else compact(r["ctx_tokens"])),
        # TREND and FLOW are not the same reading twice. TREND is an amount of
        # context added, read back out of the transcript -- so it is populated
        # on the first frame and survives --once and --json. FLOW is a shape
        # sampled while roost runs and starts empty. How much, versus when.
        ("TREND", lambda r: growth(r.get("ctx_history"))),
        ("FLOW", lambda r: r.get("flow") or " " * SPARK_LEN),
        ("IDLE", lambda r: dur(r["idle_secs"])),
    ])

    shown, quiet = arrange(workers, expand_quiet=sel is not None)

    # Widths span every row that will be printed, so groups stay aligned with
    # each other rather than each group forming its own ragged table.
    body = [[f(w) for _, f in cols] for (_, w) in shown]
    head = [h for h, _ in cols]
    wid = [max([len(head[i])] + [len(row[i]) for row in body])
           for i in range(len(cols))]

    # The group label is a gutter column, not a line of its own: a header row
    # per group cost a screen line each and pushed the table down past a short
    # window. On the first row of each group only, blank on the rest.
    lw = max([len(b[1]) for (b, _) in shown] or [0])

    if shown:
        lines.append(c("  " + " " * lw + "  "
                       + "  ".join(head[i].ljust(wid[i]) for i in range(len(cols)))
                       + "  TASK", BOLD))
    last = None
    for i, ((b, w), row) in enumerate(zip(shown, body)):
        if b[1] != last:
            last = b[1]
            label = c(b[1].ljust(lw), BOLD, BUCKET_COLORS.get(b[0], DIM))
        else:
            label = " " * lw
        cells = [label] + [style_cell(cols[j][0], row[j].ljust(wid[j]), w) for j in range(len(cols))]
        # Last gate before the terminal. Sanitised at collection too, but this
        # is the boundary that matters: every task string reaches the screen here.
        task = ascii_safe(w.get("task") or "")
        if w.get("task_src") == "prompt" and task:
            task = c(task, DIM)  # not yet named -- this is the raw last prompt
        if not w.get("auto_compact", True):
            # The one WORKERS-row marker that is not about token economics --
            # auto-compact off means NEAR LIMIT has no safety net, so it is
            # called out even on a session nowhere near its window yet.
            tag = c("[no-compact]", BOLD, YELLOW)
            task = tag + " " + task if task else tag
        # The marker is printed whether or not colour is on: over SSH, in a pipe,
        # or on a terminal with no reverse video it is the only thing that says
        # which row x would act on.
        mark = "> " if i == sel else "  "
        line = mark + "  ".join(cells) + "  " + task
        lines.append(highlight(line) if i == sel else line)

    if quiet:
        names = (" %s " % GLYPHS["sep"]).join(x["name"] for x in quiet[:12])
        # Attention colour, not dim: a list cut short must say so loudly
        # enough to be seen, or the truncation is a lie.
        tail = (c(" %s +%d" % (GLYPHS["sep"], len(quiet) - 12), YELLOW)
                if len(quiet) > 12 else "")
        lines.append("")
        lines.append(c("QUIET (%d)  " % len(quiet), BOLD, DIM) + c(names, DIM) + tail)

    # Totals, because the per-row numbers do not add up in your head. The one
    # that matters is context held across the fleet: it is what a sweep would
    # reclaim, and until now it was only answerable after the fact, from the
    # stop log. Percentages are deliberately not totalled -- they are fractions
    # of different windows, so their sum means nothing.
    held = sum(r["ctx_tokens"] or 0 for r in workers)
    added = sum(h[-1] - h[0] for h in
                (r.get("ctx_history") or [] for r in workers) if len(h) >= 2)
    near = sum(1 for r in workers if (r["ctx_pct"] or 0) >= NEAR_LIMIT_PCT)

    models = sorted(set(r["model"] for r in workers if r["model"] and r["model"] != "-"))
    summary = "%d worker(s)  |  %s held" % (len(workers), compact(held))
    if added:
        summary += "  |  %s last %d turns" % (compact(added), HISTORY_TURNS)
    if near:
        summary += "  |  " + c("%d near limit" % near, YELLOW)
    if models:
        summary += "  |  " + ", ".join(m.replace("claude-", "") for m in models)
    lines.append("")
    # Re-arm BOLD after the inner colour's RESET, for the reason highlight()
    # spells out: a RESET clears the line's bold along with the colour, so a
    # naive wrap would leave everything after "near limit" unbolded. A no-op
    # when COLOR is off, since then there are no escapes to replace.
    lines.append(c(summary.replace(RESET, RESET + BOLD), BOLD))
    return lines


def loading_dots():
    """Animated loading marker, derived from the clock.

    A static dim label is indistinguishable from a dead one; deriving the dot
    count from time gives visible motion for free on interactive paints, and
    the watch loop's once-a-second repaint tick (tick_action) is what makes
    those paints happen between collects. This is only ever called on the
    live path -- one-shot and pipe output never render a loading state, so
    their bytes stay stable for snapshot tests. Padded to its widest frame so
    the rest of the line never shifts.
    """
    return ("." * (int(time.time() * 2) % 3 + 1)).ljust(3)


def render_infra(infra):
    """One horizontal line: it is always present and rarely the thing you need."""
    parts = []
    any_unseen = False
    for s in infra:
        if s["up"] is None:
            mark = c(loading_dots(), DIM)
        elif s["up"]:
            mark = c(GLYPHS["ok"] + "up", GREEN)
        elif s.get("unseen"):
            # Never answered on an untouched default port: probably not a
            # crashed service but a port we were never told about, so no red.
            # Stays textual in both dialects -- off? is the unseen state, and
            # it must not read like either the check or the cross.
            mark = c("off?", DIM)
            any_unseen = True
        else:
            mark = c(GLYPHS["fail"] + "DOWN", BOLD, RED)
        extra = ""
        if s["up"] and s["detail"]:
            extra = " " + c(s["detail"], CYAN)
        parts.append("%s:%d %s%s" % (c(s["name"], BOLD), s["port"], mark, extra))
    body = "   ".join(parts)
    if any_unseen:
        body += "   " + c("off? = never up here; set ROOST_*_PORT if it runs elsewhere", DIM)
    if UNICODE:
        return frame_panel("INFRA", [" " + body]) + [""]
    return [c("INFRA  ", BOLD) + body, ""]


def _pct_color(pct):
    if pct is None:
        return DIM
    if pct >= 80:
        return (BOLD, RED)
    if pct >= 50:
        return (YELLOW,)
    return (GREEN,)


def render_usage_caps(usage):
    """Real Anthropic caps, one line, always visible next to INFRA -- same
    "quiet until it matters" weight, refreshed on a 2h cadence rather than
    every frame, since the source is a scrape cache, not a live probe."""
    label = c("CAPS   ", BOLD)
    if usage is None:
        return [label + c("no data yet -- claude-usage-scrape task hasn't run "
                           "(see claude-usage/README.md)", DIM), ""]

    caps = usage.get("caps") or {}
    age = usage.get("age_secs")
    stale = age is not None and age > USAGE_STALE_SECS

    burn = usage.get("burn") or {}

    def cell(key, tag, window_hours=None, burn_key=None):
        c_ = caps.get(key) or {}
        if not c_.get("visible"):
            return None
        pct = c_.get("pct")
        if pct is None:
            return None
        txt = "%s: %s" % (tag, c(str(pct) + "%", *_pct_color(pct)))
        resets = c_.get("resets_epoch")
        # Runway governor (preferred): will the cap run dry BEFORE its reset at
        # the trailing burn rate? That's the only question that matters -- it
        # legalizes front-loaded build weekends that even-burn pacing would
        # flag, and catches slow-motion overruns that look calm day-to-day.
        rate = burn.get(burn_key) if burn_key else None
        if rate and resets:
            runway_h = (100.0 - pct) / rate
            reset_h = max(0.0, (resets - time.time()) / 3600.0)
            if runway_h < reset_h:
                style = (BOLD, RED)
            elif runway_h < reset_h * 1.5:
                style = (YELLOW,)
            else:
                style = (DIM,)
            txt += c("~%.1fd left" % (runway_h / 24.0), *style)
            return txt
        # Fallback while history is thin: even-burn pace delta.
        if window_hours and resets:
            frac = 1.0 - (resets - time.time()) / (window_hours * 3600.0)
            if 0.0 <= frac <= 1.0:
                delta = pct - 100.0 * frac
                if delta > 1:
                    style = (BOLD, RED) if delta >= 20 else (YELLOW,)
                    txt += c("(+%d)" % delta, *style)
        return txt

    parts = [p for p in (
        cell("five_hour", "5h", 5),
        cell("weekly_all_models", "Weekly", 168, "weekly"),
        cell("weekly_sonnet", "Sonnet", 168),
        cell("fable5_max", "Fable5", 168, "fable"),
    ) if p]

    credits = usage.get("openrouter_credits") or {}
    if credits.get("remaining") is not None:
        rem = credits["remaining"]
        # $10 is a full tank; yellow under half, red under $1 -- absolute dollars,
        # not percent, since the balance only refills by an explicit purchase.
        style = (BOLD, RED) if rem < 1 else (YELLOW,) if rem < 5 else (GREEN,)
        parts.append("OR$: " + c("%.2f" % rem, *style))
    spend = (usage.get("spend") or {}).get("gemini") or {}
    if spend.get("spent_usd") is not None and spend.get("cap_usd"):
        pct = 100.0 * spend["spent_usd"] / spend["cap_usd"]
        parts.append("Gem$: " + c("%.2f/%.0f" % (spend["spent_usd"], spend["cap_usd"]),
                                   *_pct_color(pct)))

    if not parts:
        return [label + c("cache present but no caps parsed", DIM), ""]

    if age is None:
        age_txt = "age unknown"
    else:
        age_txt = "%dm ago" % (age / 60) if age < 3600 else "%.1fh ago" % (age / 3600)
    age_style = (BOLD, RED) if stale else (DIM,)
    suffix = c("  (stale, %s)" % age_txt, *age_style) if stale else c("  (%s)" % age_txt, DIM)

    return [label + "  ".join(parts) + suffix, ""]


def style_cell(header, text, row):
    """Colour one padded cell. Padding is already applied, so widths are fixed."""
    if header == "CTX":
        pct = row["ctx_pct"]
        if pct is None:
            return c(text, DIM)
        if pct >= 80:
            return c(text, BOLD, RED)
        if pct >= 50:
            return c(text, YELLOW)
        return c(text, GREEN)
    if header == "TREND":
        # Dim, not coloured by size. Growth is normal -- every working session
        # grows, and colouring it would put warning colour on healthy rows and
        # compete with CTX, which is the column that actually says "act on me".
        return c(text, DIM)
    if header == "WORKER":
        # The identity role: bright blue, always bold. It used to sit unstyled
        # while each model family claimed its own colour -- but a colour per
        # model spends the semantic roles (magenta is divergence, cyan is
        # chrome, green is ok) on a fact the MODEL column already states.
        return c(text, BOLD, IDENT_BLUE)
    if header == "MODEL":
        return c(text, DIM)
    if header == "IDLE":
        # This 1-hour mark only dims the IDLE column for readability. The
        # maintenance sweep itself acts on PARKED_IDLE_HOURS (2h) and
        # STALE_IDLE_HOURS (6h) below, not on this threshold.
        if row["idle_secs"] is None:
            return c(text, DIM)
        if row["idle_secs"] >= 3600:
            return c(text, DIM)
        if row["idle_secs"] <= 60:
            return c(text, GREEN)
        return text
    if header == "NAME":
        return c(text, BOLD)
    if header in ("TOKENS", "WIN", "PID"):
        return c(text, DIM)
    return text


def render_subagents(agents, sel=None):
    """Subagents are the work a session farmed out -- and they are invisible in
    any pid-based view, since they share the parent's process."""
    if not agents:
        return panel("SUBAGENTS", [c("  none running", DIM)])
    lines = []

    cols = [
        # The STATE cell is the one marker slot this table has, so it carries
        # the liveness dot in the Unicode dialect (empty-string glyphs in
        # ASCII): filled while working, hollow otherwise. The word stays --
        # the dot is reinforcement, not replacement.
        ("STATE", lambda r: (GLYPHS["working"] if r["state"] == "working"
                             else GLYPHS["idle"]) + r["state"]),
        # The agent type is the readable name, but the parent only records it in
        # the tool result -- a still-running agent has no type yet, so the hex id
        # is kept as a suffix (and the whole label while running) to stay unique.
        ("AGENT", lambda r: ("%s/%s" % (r["agent_type"], r["agent_id"][:5]))
            if r["agent_type"] else r["agent_id"][:10]),
        ("MODEL", lambda r: (r["model"] or "-").replace("claude-", "")),
        # Absolute over window, not a bare percentage: 48k/200k says both how
        # much is loaded and how much room is left. The window label is inferred
        # (see WINDOW_TIERS); colour still follows ctx_pct.
        ("CTX", lambda r: "-" if not r["ctx_tokens"]
            else "%s/%s" % (compact(r["ctx_tokens"]), r["window"])),
        ("IDLE", lambda r: dur(r["idle_secs"])),
        ("TASK", lambda r: r["task"]),
    ]
    table = [[h for h, _ in cols]] + [[f(r) for _, f in cols] for r in agents]
    w = [max(len(row[i]) for row in table) for i in range(len(cols))]
    lines.append("  " + c("  ".join(table[0][i].ljust(w[i]) for i in range(len(cols))), BOLD))
    for i, (row, r) in enumerate(zip(table[1:], agents)):
        cells = []
        for j, (header, _) in enumerate(cols):
            txt = row[j].ljust(w[j])
            if header == "STATE":
                code = {"working": GREEN, "idle": YELLOW}.get(r["state"], DIM)
                cells.append(c(txt, BOLD, code))
            elif header == "MODEL":
                cells.append(style_cell("MODEL", txt, r))
            elif header == "CTX":
                cells.append(style_cell("CTX", txt, r))
            elif header == "AGENT":
                cells.append(c(txt, DIM))
            else:
                cells.append(txt)
        mark = "> " if i == sel else "  "
        line = mark + "  ".join(cells)
        lines.append(highlight(line) if i == sel else line)

    working = sum(1 for r in agents if r["state"] == "working")
    lines.append("  " + c("%d subagent(s), %d working" % (len(agents), working), DIM))
    return panel("SUBAGENTS", lines)


def render_detail(row, children=None):
    """Read-only row detail. Replaces the panel slot; esc returns."""
    if not row:
        return panel("DETAIL", [c("  no row selected", DIM)], extra="esc returns")
    lines = []
    is_agent = "agent_id" in row and "parent_sid" in row
    lines.append("  " + c("subagent" if is_agent else "worker", BOLD))

    def field(label, value):
        text = "" if value is None else str(value)
        lines.append("  %s  %s" % (c(label.ljust(14), DIM), ascii_safe(text)))

    if is_agent:
        field("agent_id", row.get("agent_id"))
        field("type", row.get("agent_type") or "-")
        field("parent", row.get("parent_sid"))
        field("state", row.get("state"))
    else:
        field("name", row.get("name"))
        field("source", row.get("source") or "claude")
        field("pid", row.get("pid") if row.get("pid") is not None else "none")
        field("sessionId", row.get("session_id"))
        field("cwd", row.get("cwd") or "-")
        field("age", dur(row.get("age_secs")) if row.get("age_secs") is not None else "-")
    field("model", row.get("model") or "-")
    tok = row.get("ctx_tokens")
    win = row.get("window") or "-"
    if tok is not None:
        pct = row.get("ctx_pct")
        pct_s = "" if pct is None else "  %.1f%%" % pct
        field("tokens", "%s / %s%s" % (compact(tok), win, pct_s))
    else:
        field("tokens", "- / %s" % win)
    field("idle", dur(row.get("idle_secs")) if row.get("idle_secs") is not None else "-")
    if not is_agent:
        start_tok = row.get("start_tokens")
        field("startup", "%s tokens (before the first real turn)" % compact(start_tok)
              if start_tok is not None else "-")
    field("cost/turn", "not on disk")
    task = ascii_safe(row.get("task") or "")
    lines.append("  " + c("task", DIM))
    if task:
        width = max(40, shutil.get_terminal_size((100, 40)).columns - 6)
        for part in textwrap.wrap(task, width) or [task]:
            lines.append("    " + part)
    else:
        lines.append(c("    (none)", DIM))
    if not is_agent:
        kids = children if children is not None else []
        lines.append("  " + c("subagents", DIM))
        if not kids:
            lines.append(c("    none", DIM))
        else:
            for a in kids:
                label = a.get("agent_type") or (a.get("agent_id") or "")[:10]
                lines.append("    %s  %s  %s" % (
                    a.get("state", "-"), label, ascii_safe(a.get("task") or "")))
    return panel("DETAIL", lines, extra="esc returns")


def tab_table_focus(focus, view):
    """Tab cycles WORKERS <-> SUBAGENTS. Opens the subagents panel if needed."""
    if focus != "agents":
        return "agents", "agents"
    return "workers", view if view else "agents"


def render_models(models):
    """The INFRA line only ever shows what is resident in VRAM right now -- a
    model that is installed but idle drops out of it entirely. This is the
    full inventory: everything `ollama list` knows about, with residency and
    VRAM called out for whichever happen to be loaded."""
    if not models:
        return panel("LOCAL MODELS",
                     [c("  none installed (or ollama not running)", DIM)])
    lines = []

    cols = [
        ("MODEL", lambda m: m["name"]),
        ("DISK", lambda m: "%.1f GB" % m["disk_gb"]),
        ("STATE", lambda m: "resident" if m["resident"] else "unloaded"),
        ("VRAM", lambda m: "-" if m["vram_gb"] is None else "%.1f GB" % m["vram_gb"]),
        ("UNLOADS IN", lambda m: "-" if m["expires_secs"] is None else dur(m["expires_secs"])),
    ]
    table = [[h for h, _ in cols]] + [[f(m) for _, f in cols] for m in models]
    w = [max(len(row[i]) for row in table) for i in range(len(cols))]
    lines.append("  " + c("  ".join(table[0][i].ljust(w[i]) for i in range(len(cols))), BOLD))
    for row, m in zip(table[1:], models):
        cells = []
        for i, (header, _) in enumerate(cols):
            txt = row[i].ljust(w[i])
            if header == "MODEL":
                cells.append(c(txt, BOLD))
            elif header == "STATE":
                cells.append(c(txt, BOLD, GREEN) if m["resident"] else c(txt, DIM))
            else:
                cells.append(c(txt, DIM))
        lines.append("  " + "  ".join(cells))

    resident = sum(1 for m in models if m["resident"])
    lines.append("  " + c("%d installed, %d resident" % (len(models), resident), DIM))
    return panel("LOCAL MODELS", lines)


# name, key, what it shows. Not a keybinding reference -- the footer hint
# already has the keys -- just what each screen on the display means. roost
# is small enough that this list is the whole manual.
HELP_SCREENS = (
    ("INFRA", None,
     "ollama / litellm / openwebui: up, DOWN, or off? (never answered on a default "
     "port -- likely no such service, or set ROOST_*_PORT to where it lives), "
     "plus what's resident in Ollama's VRAM right now."),
    ("WORKERS", None,
     "every live Claude Code session: model, context window used, idle time, current task. "
     "TREND is how much context the session added over its last few turns, read out of the "
     "transcript, so it is filled in on the first frame. FLOW is a sparkline of recent token "
     "throughput -- '.' is a quiet sample, the ramp is "
     "activity; history starts when roost starts. QUIET collapses idle sessions to one line; "
     "raise the cursor to expand it."),
    ("SUBAGENTS", "s",
     "work a session farmed out. Invisible in any pid-based view, since a subagent shares its "
     "parent's process rather than running as one of its own. AGENT shows the agent's type once "
     "it finishes (hex id while running); CTX is tokens over the inferred window."),
    ("ADVICE", "a",
     "concrete actions, ranked by how many tokens each would save -- which sessions are "
     "expensive to resume, near their context limit, or just idle clutter."),
    ("LOCAL MODELS", "m",
     "everything Ollama has installed, not just what's resident in VRAM -- disk size, "
     "residency, and time until an idle model unloads."),
    ("USAGE", "u",
     "tokens per day per model over the last week, tallied from the transcripts on disk. "
     "An estimate of burn, not the real Anthropic meter; set ROOST_WEEKLY_BUDGET to see "
     "it as a share of your plan. Local (non claude-*) models are flagged and excluded "
     "from the budget math. First open scans a week of transcripts and can pause for a "
     "moment; after that it reads only what was appended."),
    ("GATEWAY", "g",
     "LiteLLM liveliness, the aliases in config.yaml (configured intent, not "
     "whether a backing model will answer), plus batch-run progress from output "
     "files. Missing config is a labelled gap. Green batch rows are writing. "
     "probed is when roost last listed the batch dir; LAST WRITE is when an "
     "output file last landed -- a live panel with a days-old LAST WRITE means "
     "the pipeline stopped, not that roost is stale."),
    ("REMOTE", "r",
     "other machines' roost, over ssh. One row per host in ROOST_REMOTES: workers, "
     "resident models, batch progress, job queue. Fetched on a background thread and "
     "cached, so an unreachable host shows its last good row with an age instead of "
     "hanging the display."),
)


def render_help():
    """What each screen means, not how to drive it -- the footer hint already
    lists the keys, and roost has few enough screens that this fits on one page."""
    labels = ["%s (%s)" % (n, k) if k else n for n, k, _ in HELP_SCREENS]
    lw = max(len(x) for x in labels)
    # Same gutter shape as the worker table: label beside its text, not above
    # it. That halves the panel's height, and the text wraps to the window
    # instead of running off the right edge as one clipped line.
    # The frame borders and their padding cost four more columns in Unicode.
    body = max(20, shutil.get_terminal_size((150, 40)).columns - lw
               - (9 if UNICODE else 5))

    lines = []
    for label, (_, _, text) in zip(labels, HELP_SCREENS):
        wrapped = textwrap.wrap(text, body) or [""]
        for i, part in enumerate(wrapped):
            gutter = c(label.ljust(lw), BOLD) if i == 0 else " " * lw
            lines.append("  " + gutter + "  " + c(part, DIM))
    lines.append("")
    lines.append("  " + c("interactive mode (i) arms the cursor: j/k select, Tab "
                          "switches tables, Enter opens a row, x stop, "
                          "y copy sessionId, esc deselect.", DIM))
    return panel("HELP", lines)


def collect_snapshot(view=None, focus="workers", detail=None):
    """Run every collector the current view needs and stamp the collect clock.

    Collection and rendering used to be fused in frame(), which meant a window
    resize could not reflow the display without paying for a full rescan of
    transcripts and processes. The snapshot is the seam: the watch loop keeps
    the last one and can re-render it at a new width for free. The view/focus
    gating is unchanged from the fused version -- panel collectors only run for
    the panel that is actually open.
    """
    workers = collect_workers()
    live_sids = set(w["session_id"] for w in workers if w.get("session_id"))
    agents = []
    if view in ("agents", "detail") or focus == "agents":
        agents = collect_subagents(live_sids)
    snap = {"workers": workers, "agents": agents, "infra": infra_cached()}
    if view == "detail":
        kids = []
        if detail and detail.get("session_id"):
            kids = [a for a in agents if a.get("parent_sid") == detail.get("session_id")]
            if not kids:
                kids = collect_subagents({detail["session_id"]})
        snap["detail_kids"] = kids
    elif view == "models":
        snap["models"] = collect_local_models()
    elif view == "usage":
        snap["usage"] = collect_usage()
    elif view == "gateway":
        snap["gateway"] = collect_gateway()
    elif view == "remote":
        snap["remote"] = collect_remote()
    prune_caches()
    global _LAST_COLLECT
    _LAST_COLLECT = time.time()
    return snap


def render_frame(snap, view=None, sel=None, focus="workers", detail=None):
    """Render a collected snapshot; touches no collector and no clock.

    Returns (lines, rows, sel). `view` is the open panel: "agents",
    "models", "advice", "usage", "help", "detail", or None for the bare worker table.

    `rows` is what the cursor indexes for the focused table (workers or
    subagents). `sel` comes back clamped: sessions exit between frames, and a
    cursor left pointing past the end would silently address nothing.
    """
    workers = snap["workers"]
    shown, _ = arrange(workers, expand_quiet=sel is not None and focus == "workers")
    worker_rows = [w for _, w in shown]
    agents = snap["agents"]
    if focus == "agents":
        rows = agents
    else:
        rows = worker_rows
    if sel is not None:
        sel = min(sel, len(rows) - 1) if rows else None
    # Infra leads because it is a constant: one quiet line you skim past, which
    # is exactly the weight it deserves until something turns red.
    lines = render_infra(snap["infra"])
    win_note = window_config_note()
    if win_note:
        lines.append(c("WINDOW ", BOLD) + win_note)
        lines.append("")
    # Panels paint *above* the worker table. Below it they disappeared under
    # "taller window" truncation whenever the board was long -- pressing s/m/a
    # looked like a no-op even though the view toggled.
    if view == "agents":
        lines.extend(render_subagents(
            agents, sel=sel if focus == "agents" else None))
    elif view == "detail":
        lines.extend(render_detail(detail, children=snap.get("detail_kids", [])))
    elif view == "models":
        lines.extend(render_models(snap.get("models")))
    elif view == "usage":
        lines.extend(render_usage(snap.get("usage")))
    elif view == "gateway":
        lines.extend(render_gateway(snap.get("gateway")))
    elif view == "remote":
        lines.extend(render_remote(snap.get("remote")))
    elif view == "help":
        lines.extend(render_help())
    elif view == "advice":
        # advise() carries its own leading blank via panel().
        lines.extend(advise(workers))
    lines.extend(render(workers, sel if focus == "workers" else None))
    return lines, rows, sel


def frame(view=None, sel=None, focus="workers", detail=None):
    """Collect and render in one call -- the shape --once and tests rely on."""
    snap = collect_snapshot(view, focus=focus, detail=detail)
    return render_frame(snap, view, sel, focus=focus, detail=detail)


def prune_caches():
    """Evict cache entries whose session or agent vanished since the last frame.

    Every live path and agent re-registers itself each tick, so anything left
    over is history -- without this the caches grow for as long as the dashboard
    stays open, which is days.
    """
    global _CURSOR_TX_INDEX
    for cache in (_SCAN_CACHE, _HARVEST_POS):
        for k in [k for k in cache if k not in _SEEN_PATHS]:
            del cache[k]
    for cache in (_AGENT_META, _AGENT_LABEL):
        for k in [k for k in cache if k not in _SEEN_AGENTS]:
            del cache[k]
    _SEEN_PATHS.clear()
    _SEEN_AGENTS.clear()
    _CURSOR_TX_INDEX = None


# (handle, original mode) when enable_vt() changed the console, else None.
# Conhost keeps a mutated mode after the process exits, so it must be put back.
_VT_ORIGINAL = None


def enable_vt():
    """Turn on ANSI escape handling. Windows consoles have it off by default.

    Without it the cursor-home and erase sequences are ignored, so every frame is
    appended below the last instead of overwriting it -- the screen pages away.
    """
    global _VT_ORIGINAL
    if os.name != "nt":
        return True
    try:
        import ctypes

        k = ctypes.windll.kernel32
        handle = k.GetStdHandle(-11)
        mode = ctypes.c_uint32()
        if not k.GetConsoleMode(handle, ctypes.byref(mode)):
            return False  # redirected, or not attached to a console at all
        ENABLE_VIRTUAL_TERMINAL_PROCESSING = 0x0004
        if mode.value & ENABLE_VIRTUAL_TERMINAL_PROCESSING:
            return True
        if k.SetConsoleMode(handle, mode.value | ENABLE_VIRTUAL_TERMINAL_PROCESSING):
            _VT_ORIGINAL = (handle, mode.value)
            return True
        return False
    except Exception:
        return False


def restore_vt():
    """Put the console mode back exactly as enable_vt() found it."""
    if _VT_ORIGINAL is None:
        return
    try:
        import ctypes

        ctypes.windll.kernel32.SetConsoleMode(_VT_ORIGINAL[0], _VT_ORIGINAL[1])
    except Exception:
        pass


class KeyReader(object):
    """Non-blocking single keypresses, or a no-op when stdin is not a terminal.

    Used so the interval sleep stays interruptible: space repaints immediately
    instead of waiting out the remainder of the tick.
    """

    def __init__(self):
        self.enabled = False
        self._fd = None
        self._saved = None

    def __enter__(self):
        try:
            if not sys.stdin.isatty():
                return self
        except (ValueError, AttributeError):
            return self
        if os.name == "nt":
            try:
                import msvcrt  # noqa: F401

                self.enabled = True
            except ImportError:
                pass
            return self
        try:
            import termios
            import tty

            self._fd = sys.stdin.fileno()
            self._saved = termios.tcgetattr(self._fd)
            tty.setcbreak(self._fd)
            self.enabled = True
        except Exception:
            self._saved = None
        return self

    def __exit__(self, *exc):
        if self._saved is not None:
            try:
                import termios

                # TCSAFLUSH, not TCSADRAIN: discard queued input (the tail of an
                # arrow sequence, keys typed during a slow frame) instead of
                # delivering it to the shell prompt after exit.
                termios.tcsetattr(self._fd, termios.TCSAFLUSH, self._saved)
            except Exception:
                pass
        return False

    def get(self, timeout):
        """One key within `timeout` seconds, else None.

        Returns a single character, or one of the names "UP", "DOWN", "ESC" --
        arrows are multi-byte on both platforms and the caller should not have to
        know either encoding.
        """
        if not self.enabled:
            time.sleep(timeout)
            return None
        if os.name == "nt":
            import msvcrt

            end = time.time() + timeout
            while time.time() < end:
                if msvcrt.kbhit():
                    ch = msvcrt.getwch()
                    # Arrows and function keys arrive as a two-char sequence.
                    if ch in ("\x00", "\xe0"):
                        return {"H": "UP", "P": "DOWN"}.get(msvcrt.getwch())
                    return "ESC" if ch == "\x1b" else ch
                time.sleep(0.02)
            return None
        import select

        # os.read on the raw fd, never sys.stdin.read: the buffered TextIO can
        # slurp several bytes into user space, after which select() on the fd
        # reports "not ready" while input sits in the buffer -- arrow tails then
        # surface as stray characters, or spill to the shell after exit.
        def read1():
            b = os.read(self._fd, 1)
            return b.decode("utf-8", "replace") if b else ""

        r, _, _ = select.select([self._fd], [], [], timeout)
        if not r:
            return None
        ch = read1()
        if ch != "\x1b":
            return ch
        # A bare Esc and the start of an arrow sequence are the same byte. The
        # rest of a real CSI arrives in the same burst, so nothing further within
        # a beat means the user pressed Esc.
        r, _, _ = select.select([self._fd], [], [], 0.05)
        if not r or read1() != "[":
            return "ESC"
        return {"A": "UP", "B": "DOWN"}.get(read1())


def term_size():
    size = shutil.get_terminal_size((150, 40))
    # Redirected output reports 0x0, which would otherwise clip every line to nothing.
    cols = size.columns if size.columns >= 20 else 150
    rows = size.lines if size.lines >= 5 else 40
    return cols, rows


RESIZE_REPAINT_S = 0.12  # minimum gap between live repaints during a drag


class ResizeThrottle:
    """Live repaint policy for a drag-resize.

    The first cut of resize handling waited for the size to hold steady for a
    full slice before repainting at all -- so mid-drag the terminal rewrapped
    the stale frame into a jumble that only cleared once the drag stopped.
    Now a moving size IS the signal: while it keeps changing, the loop
    repaints from the cached snapshot at the current size, throttled to one
    paint per RESIZE_REPAINT_S, and the mismatch left behind when the drag
    stops buys the final paint at the settled size.

    Pure decision logic over an injected clock, so tests can drive a whole
    drag without sleeping. The loop owns the actual painting, which stays
    render-from-cache: no collector runs and _LAST_COLLECT stays put.
    """

    def __init__(self, interval=RESIZE_REPAINT_S):
        self.interval = interval
        self._last_paint = 0.0
        self._last_moving = 0.0

    def poll(self, size, painted_size, now):
        """True when the loop should repaint from cache at `size` now."""
        if size == painted_size:
            return False
        self._last_moving = now
        if now - self._last_paint < self.interval:
            return False  # mid-drag, too soon: a later slice picks it up
        self._last_paint = now
        return True

    def slice_len(self, now):
        """Idle slice length for the key wait: shortened while a drag is in
        flight so live repaints actually land at ~RESIZE_REPAINT_S cadence,
        back to the keypress-latency slice once the size has gone quiet."""
        return 0.05 if now - self._last_moving < 0.5 else 0.2


REPAINT_SECONDS = 1.0  # render-from-cache cadence between collects


def tick_action(now, deadline, painted_at):
    """What an idle slice owes the frame.

    "collect" once the collect deadline has passed; "repaint" once a whole
    REPAINT_SECONDS has elapsed since the last paint; None to keep waiting.
    Collection keeps its own (much longer) deadline. Between collects the frame
    is repainted from the cached snapshot once a second so the clock, the
    "updated Ns ago" chip and the loading dots visibly move -- before this
    tick, repaints only followed collects, so the chip almost always read "0s"
    and the dots never animated. A repaint runs no collector and leaves
    _LAST_COLLECT alone, so the chip keeps reporting data age.
    """
    if now >= deadline:
        return "collect"
    if now - painted_at >= REPAINT_SECONDS:
        return "repaint"
    return None


def wait_slice(now, deadline, painted_at, slice_len):
    """How long the key wait may block before the loop needs control back:
    the collect deadline, the next repaint tick, or the resize-poll slice,
    whichever comes first. Never negative."""
    return max(0.0, min(slice_len, deadline - now, painted_at + REPAINT_SECONDS - now))


def key_hint(width, interactive=False, view=None):
    """The footer key hint, degraded in a designed order rather than clipped.

    It used to be one concatenated string ending "... h help | q quit",
    hard-clipped at the right edge -- so the narrower the window, the sooner
    q and h were destroyed, and they are the two hints with no other way to
    be discovered. Now they are emitted first and survive every tier; the
    optional chips append in a fixed order only while they fit.
    """
    def lit(key, on):
        return c(key, BOLD, GREEN) if on else key

    if interactive:
        i_tag = c("i", BOLD, GREEN) + " interactive " + c("ARMED", BOLD, GREEN)
        cur_tag = "j/k Tab Enter x y esc"
    else:
        i_tag = c("i", BOLD) + " interactive " + c("off  ", DIM)
        cur_tag = c("j/k Tab Enter x y esc", DIM)
    chips = [
        "space refresh",
        i_tag,
        cur_tag,
        lit("a", view == "advice") + " advice",
        lit("s", view == "agents") + " agents",
        lit("m", view == "models") + " models",
        lit("u", view == "usage") + " usage",
        lit("g", view == "gateway") + " gateway",
        lit("r", view == "remote") + " remote",
    ]
    out = "q quit | " + lit("h", view == "help") + " help"
    if visible_len(out) > width:
        # Below the floor the hint keeps shrinking rather than overflowing:
        # the header owns the width budget and must be able to trust that
        # what it asked for is what it gets, or the safety tag to its right
        # is what the edge clips. q is the last letter standing.
        for short in ("q quit", "q"):
            if len(short) <= width:
                return short
        return ""
    for chip in chips:
        cand = out + " | " + chip
        if visible_len(cand) > width:
            break
        out = cand
    return out


EXPERIMENTAL_TAG = " EXPERIMENTAL "


def header_title(cols, host, clock, age_secs=None, keys_enabled=True,
                 interactive=False, view=None, tagged=False):
    """Assemble the header's title line to fit `cols - 1` visible columns.

    Fields shed in a fixed order under width pressure, least important first:
    the key hint's optional chips, then the "updated Ns ago" age chip, then
    the hostname, then the hint's floor ("q quit | h help" thins to "q quit",
    then "q" -- q and h have no other way to be discovered, so they outrank
    the chips that describe them). The product name and the clock come after
    everything else -- a wall display must always answer "when did this last
    update" -- and the clock is the last data element to go. When interactive
    mode is armed the EXPERIMENTAL safety tag is reserved before any field is
    even considered: the one marker that says "x can end a process" is never
    what the right edge clips, even at the 20-column minimum, where it
    outranks the name and finally the clock. paint() clips at cols - 1, so
    the budget here is the same number.
    """
    budget = cols - 1
    tag = c(EXPERIMENTAL_TAG, BOLD, REVERSE, YELLOW) if tagged else ""
    reserve = visible_len(tag) + 1 if tagged else 0   # tag plus one gap
    name = c("roost", BOLD)
    parts = [name, clock]
    used = visible_len(name) + 2 + len(clock)
    if used + reserve > budget:
        # Only below ~31 columns with the tag up: the name yields first, and
        # the clock only if even it cannot share the row with the tag.
        parts = [clock] if len(clock) + reserve <= budget else []
        used = len(clock) if parts else 0
    avail = budget - reserve - used
    floor = (len("q quit | h help") if keys_enabled else len("Ctrl-C to stop")) + 3
    optional = avail - floor  # room beyond the hint floor for host and age
    if host and len(host) + 2 <= optional:
        parts.insert(max(0, len(parts) - 1), c(host, CYAN))  # before the clock
        avail -= len(host) + 2
        optional -= len(host) + 2
    title = "  ".join(parts)
    if age_secs is not None:
        upd = "updated %s ago" % dur(max(0, age_secs))
        if len(upd) + 2 <= optional:
            title += "  " + c(upd, DIM)
            avail -= len(upd) + 2
    if not keys_enabled:
        hint = "Ctrl-C to stop" if len("Ctrl-C to stop") + 3 <= avail else ""
    else:
        # The hint's chips never reword when interactive is armed or a panel
        # opens -- only the colours change, and "off" is padded to ARMED's
        # width -- so the header cannot reflow underfoot. The armed state is
        # spelled out, not just tinted: the one mode that can end a process
        # should never be ambiguous.
        hint = key_hint(avail - 3, interactive, view)
    if hint:
        title += ("   " if title else "") + c(hint, DIM)
    if tagged:
        # Pinned top-right so it sits above the table rather than anywhere
        # the frame can clip it away; the reserve above guarantees the room.
        pad = budget - visible_len(title) - visible_len(tag)
        title += " " * max(1 if title else 0, pad) + tag
    return title


def paint(lines, vt):
    """Redraw in place.

    Clipped to the window in both directions on purpose: a line that wraps, or a
    frame taller than the terminal, scrolls the display -- which looks identical
    to a clear that never happened.
    """
    cols, rows = term_size()
    body = [clip_ansi(ln, cols - 1) for ln in lines]
    # Say so when the frame does not fit. Silent truncation is how a confirmation
    # prompt and an ADVICE panel both went missing without appearing to fail --
    # the screen looked complete, so nothing suggested there was more below it.
    if len(body) > rows - 1:
        hidden = len(body) - (rows - 2)
        body = body[: rows - 2] + [clip_ansi(
            # Attention colour, not dim: this is the line that says the frame
            # is lying about being complete.
            c("%s %d more line(s) below -- taller window, or close a panel "
              "(s/a/m/u/g/r/h)" % (GLYPHS["ell"], hidden), YELLOW), cols - 1)]

    # Version, bottom-right. Stamped onto whatever the last visible line turns
    # out to be -- including the overflow notice above -- so it cannot itself be
    # the thing that gets clipped off. INFRA is not a footer to hang it on: it
    # leads the frame. Padded by visible_len, since escape bytes are not columns
    # and len() would push it off the right edge by the number of colour codes
    # in the line. Dropped rather than wrapped when there is no room: a wrapped
    # line scrolls the display, which looks identical to a clear that never ran.
    if body:
        stamp = c("v" + __version__, DIM)
        room = cols - 1 - visible_len(body[-1]) - visible_len(stamp)
        if room >= 2:
            body[-1] += " " * room + stamp
    if vt:
        # Home, overwrite each line erasing its old tail, then wipe any rows left
        # over from a taller previous frame. Flicker-free, unlike a full clear.
        sys.stdout.write("\033[H" + "".join(ln + "\033[K\n" for ln in body) + "\033[J")
    else:
        os.system("cls" if os.name == "nt" else "clear")
        sys.stdout.write("\n".join(body) + "\n")
    sys.stdout.flush()


def build_parser():
    ap = argparse.ArgumentParser(description="Live Claude workers and local infra on one screen.")
    ap.add_argument("-w", "--watch", nargs="?", const=REFRESH_SECONDS, type=float,
                    metavar="SECS",
                    help="refresh interval in seconds (default %g, set by REFRESH_SECONDS)"
                         % REFRESH_SECONDS)
    ap.add_argument("-1", "--once", action="store_true",
                    help="print a single frame and exit (live is the default)")
    ap.add_argument("--version", action="version", version="roost " + __version__)
    ap.add_argument("--json", action="store_true", help="emit records as JSON and exit")
    ap.add_argument("--no-color", action="store_true", help="disable colour output")
    ap.add_argument("--ascii", action="store_true",
                    help="force the ASCII glyph dialect (also %s=1); an interactive "
                         "UTF-8 terminal otherwise gets rounded frames and Unicode "
                         "glyphs, while pipes, --once and --json are always ASCII"
                         % ASCII_ENV)
    ap.add_argument("--advise", action="store_true",
                    help="start with the ADVICE panel open (toggle live with 'a')")
    ap.add_argument("--no-agents", action="store_true",
                    help="start with the SUBAGENTS panel closed (toggle live with 's')")
    ap.add_argument("--models", action="store_true",
                    help="start with the LOCAL MODELS panel open (toggle live with 'm')")
    ap.add_argument("--usage", action="store_true",
                    help="start with the USAGE panel open (toggle live with 'u'); "
                         "set %s (e.g. 60M) to show weekly burn against a budget"
                         % USAGE_BUDGET_ENV)
    ap.add_argument("--gateway", action="store_true",
                    help="start with the GATEWAY panel open (toggle live with 'g')")
    ap.add_argument("--remote", action="store_true",
                    help="start with the REMOTE panel open (toggle live with 'r'); "
                         "hosts come from %s (comma-separated ssh aliases)" % REMOTES_ENV)
    ap.add_argument("--interactive", action="store_true",
                    help="start with interactive mode armed -- cursor, x/y, and the "
                         "EXPERIMENTAL tag (default off; toggle live with 'i')")
    ap.add_argument("--no-log", action="store_true",
                    help="do not record stopped sessions to %s" % LOG_PATH)
    ap.add_argument("--ollama-port", type=int, metavar="PORT",
                    help="ollama port for the INFRA panel (default %d, or %s)"
                         % (OLLAMA_PORT, OLLAMA_PORT_ENV))
    ap.add_argument("--litellm-port", type=int, metavar="PORT",
                    help="litellm port for the INFRA/GATEWAY panels (default %d, or %s)"
                         % (LITELLM_PORT, LITELLM_PORT_ENV))
    ap.add_argument("--openwebui-port", type=int, metavar="PORT",
                    help="open-webui port for the INFRA panel (default %d, or %s)"
                         % (OPENWEBUI_PORT, OPENWEBUI_PORT_ENV))
    ap.add_argument("--print-completion", choices=["bash", "zsh", "powershell"],
                    help="print a shell completion script and exit")
    ap.epilog = (
        "keys while running:  space = refresh now   a = advice panel   "
        "s = subagents panel   m = local models panel   u = usage panel   "
        "g = gateway panel   r = remote panel   "
        "h or ? = what am I looking at   "
        "i = arm interactive mode   q = quit\n"
        "interactive mode (armed with i):  j/k or arrows move a cursor   "
        "Tab = workers/subagents   Enter = row detail   "
        "x = stop the session (confirms)   y = copy its sessionId   esc = deselect\n"
        "source and issues: https://github.com/gmhoward9289-ops/roost")
    return ap


def _completion_flag_words():
    words = []
    for action in build_parser()._actions:
        if action.option_strings:
            words.extend(action.option_strings)
    return sorted(set(words), key=lambda s: (len(s), s))


# Flags that take a value on the command line. Kept in one place so completion
# scripts stay aligned with argparse without pulling in argcomplete.
def print_completion(shell):
    flags = " ".join(_completion_flag_words())
    if shell == "bash":
        return """# bash completion for roost. Install to
# /usr/share/bash-completion/completions/roost, or eval:
#   eval "$(roost --print-completion bash)"
_roost() {
    local cur prev
    cur="${COMP_WORDS[COMP_CWORD]}"
    prev="${COMP_WORDS[COMP_CWORD-1]}"
    local opts="%s"
    case "$prev" in
        -w|--watch|--ollama-port|--litellm-port|--openwebui-port)
            return 0
            ;;
        --print-completion)
            COMPREPLY=( $(compgen -W "bash zsh powershell" -- "$cur") )
            return 0
            ;;
    esac
    if [[ "$cur" == -* ]]; then
        COMPREPLY=( $(compgen -W "$opts" -- "$cur") )
    fi
}
complete -F _roost roost
""" % flags
    if shell == "zsh":
        return """#compdef roost
# zsh completion for roost. Install to
# /usr/share/zsh/vendor-completions/_roost, or eval:
#   source <(roost --print-completion zsh)
_arguments -S -C \\
  '(-1 --once)'{-1,--once}'[print a single frame and exit]' \\
  '--json[emit records as JSON and exit]' \\
  '--no-color[disable colour output]' \\
  '--ascii[force the ASCII glyph dialect]' \\
  '--advise[start with the ADVICE panel open]' \\
  '--no-agents[start with the SUBAGENTS panel closed]' \\
  '--models[start with the LOCAL MODELS panel open]' \\
  '--usage[start with the USAGE panel open]' \\
  '--gateway[start with the GATEWAY panel open]' \\
  '--remote[start with the REMOTE panel open]' \\
  '--interactive[start with interactive mode armed]' \\
  '--no-log[do not record stopped sessions]' \\
  '--version[print the version and exit]' \\
  '(-w --watch)'{-w,--watch}'[refresh interval in seconds]:seconds:' \\
  '--ollama-port[ollama port for the INFRA panel]:port:' \\
  '--litellm-port[litellm port for the INFRA/GATEWAY panels]:port:' \\
  '--openwebui-port[open-webui port for the INFRA panel]:port:' \\
  '--print-completion[print a shell completion script]:shell:(bash zsh powershell)'
"""
    if shell == "powershell":
        ps_flags = ", ".join("'%s'" % f for f in _completion_flag_words())
        return """# PowerShell completion for roost. Add to your profile:
#   . (roost --print-completion powershell | Out-String | Invoke-Expression)
Register-ArgumentCompleter -Native -CommandName roost -ScriptBlock {
    param($wordToComplete, $commandAst, $cursorPosition)
    $flags = @(%s)
    $valueFlags = @('-w', '--watch', '--ollama-port', '--litellm-port',
                    '--openwebui-port', '--print-completion')
    $prev = $commandAst.CommandElements[
        [Math]::Max(0, $commandAst.CommandElements.Count - 2)].ToString()
    if ($valueFlags -contains $prev) {
        if ($prev -eq '--print-completion') {
            'bash', 'zsh', 'powershell' | Where-Object {
                $_ -like "$wordToComplete*"
            } | ForEach-Object {
                [System.Management.Automation.CompletionResult]::new($_, $_, 'ParameterValue', $_)
            }
        }
        return
    }
    $flags | Where-Object { $_ -like "$wordToComplete*" } | ForEach-Object {
        [System.Management.Automation.CompletionResult]::new($_, $_, 'ParameterName', $_)
    }
}
""" % ps_flags
    raise ValueError("unknown shell %r" % shell)


def main():
    global LOGGING, OLLAMA_PORT, LITELLM_PORT, OPENWEBUI_PORT
    ap = build_parser()
    args = ap.parse_args()

    if args.print_completion:
        sys.stdout.write(print_completion(args.print_completion))
        return

    LOGGING = not args.no_log
    # Dialect is decided once, up front: pipe-safe modes and --ascii pin the
    # ASCII tier before the terminal is consulted, so --once/--json output
    # stays byte-identical whatever the terminal can render.
    set_dialect(choose_dialect(args.once or args.json, args.ascii))
    if args.ollama_port is not None:
        OLLAMA_PORT = args.ollama_port
        _PORT_CONFIGURED["ollama"] = True
    if args.litellm_port is not None:
        LITELLM_PORT = args.litellm_port
        _PORT_CONFIGURED["litellm"] = True
    if args.openwebui_port is not None:
        OPENWEBUI_PORT = args.openwebui_port
        _PORT_CONFIGURED["openwebui"] = True

    if args.json:
        workers = collect_workers()
        live_sids = set(w["session_id"] for w in workers if w.get("session_id"))
        print(json.dumps({
            "schema": SCHEMA_SNAPSHOT,
            "version": __version__,
            "workers": workers,
            "subagents": collect_subagents(live_sids),
            "infra": collect_infra(),
            "usage_caps": collect_usage_caps(),
            "local_models": collect_local_models(),
            "gateway": collect_gateway(),
        }, indent=2))
        return

    # Die by unwinding, not by default disposition: a plain kill would skip the
    # KeyReader __exit__ and the cursor/console restore below, leaving the tty
    # in cbreak/no-echo with the cursor hidden. SystemExit runs both.
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    if hasattr(signal, "SIGHUP"):
        signal.signal(signal.SIGHUP, lambda *_: sys.exit(0))

    vt = enable_vt()

    global COLOR
    # Colour needs escape support and a real terminal. NO_COLOR is the community
    # convention (https://no-color.org) and costs nothing to honour.
    COLOR = (
        vt
        and not args.no_color
        and not os.environ.get("NO_COLOR")
        and sys.stdout.isatty()
    )

    # Panels are toggled live rather than fixed at launch: on a short terminal all
    # of them at once overflow the window, and what you want to see changes.
    # One panel at a time. They used to be independent toggles, which meant
    # opening ADVICE while SUBAGENTS was up pushed it past the bottom of the
    # terminal -- you had to close the other one first to see the one you asked
    # for. Flipping is what "show me the advice" actually means.
    if args.advise:
        view = "advice"
    elif args.models:
        view = "models"
    elif args.usage:
        view = "usage"
    elif args.gateway:
        view = "gateway"
    elif args.remote:
        view = "remote"
    elif args.no_agents:
        view = None
    else:
        view = "agents"

    # Live is the default, as with top/htop -- `-h` is argparse's help and exits,
    # so keys have nothing to act on there. A single frame is opt-in.
    if args.once:
        print("\n".join(frame(view)[0]))
        return
    interval = args.watch if args.watch else REFRESH_SECONDS

    global _INFRA_ALLOW_DEFER
    _INFRA_ALLOW_DEFER = True
    # Kick the INFRA probe thread before the first collect so localhost
    # timeouts overlap with transcript work instead of serialising after it.
    infra_cached()

    if vt:
        sys.stdout.write("\033[2J\033[?25l")  # one clear up front, then hide the cursor
        # Immediate feedback: a cold first frame used to sit on a blank screen
        # for many seconds while hundreds of finished transcripts were scanned.
        host = socket.gethostname()
        msg = (c("roost", BOLD) + "  " + c(host, CYAN)
               + "  " + c("loading" + loading_dots(), DIM) + "\n")
        sys.stdout.write("\033[H" + msg)
        sys.stdout.flush()

    # Off by default: this is the gate on the half that can end a process, and
    # arming it is one deliberate keypress rather than "having a terminal."
    interactive = bool(args.interactive)
    sel = None      # cursor row index, or None when there is no cursor
    focus = "workers"  # "workers" or "agents" -- Tab switches
    detail = None   # row shown in the DETAIL panel, or None
    pending = None  # the worker row awaiting a y/n answer
    note = None     # result of the last action, cleared by the next keypress
    snap = None      # last collected snapshot -- a resize re-renders from it
    collect_due = True  # ticks and keypresses collect; a bare resize does not
    deadline = 0.0
    resize = ResizeThrottle()  # outlives each frame so a drag stays throttled
    try:
        with KeyReader() as keys:
            while True:
                if collect_due or snap is None:
                    snap = collect_snapshot(view, focus=focus, detail=detail)
                    deadline = time.time() + interval
                    collect_due = False
                lines, rows, sel = render_frame(snap, view, sel, focus=focus, detail=detail)
                # A session can exit while its confirmation is on screen. Matching
                # on pid rather than on the row dict is what makes that detectable:
                # every frame rebuilds the dicts, so identity and equality both
                # fail on rows that are in fact the same session.
                if pending and not any(_worker_key(r) == _worker_key(pending) for r in rows):
                    pending, note = None, c("that session exited on its own", DIM)

                # The status line lives in the header, above the table, and the
                # blank placeholder keeps it there so nothing shifts when it
                # fills. It used to be appended under the table, where paint()
                # clipped it away: 24 sessions and their subagents make a frame
                # taller than the terminal, so the confirmation was invisible
                # precisely when there was most to act on, and the next keypress
                # cancelled a prompt that had never been seen.
                if pending:
                    if pending.get("pid") is not None:
                        status = c("stop %s (pid %d)?   y = yes, any other key = no" % (
                            pending["name"], pending["pid"]), BOLD, RED)
                    else:
                        status = c("stop %s?   y = yes, any other key = no" % (
                            pending["name"],), BOLD, RED)
                else:
                    status = note or ""
                painted_size = term_size()
                # The EXPERIMENTAL tag only while interactive mode is armed,
                # because that is the half that can end a process; reading
                # the dashboard has never been the risky part. header_title
                # owns the shed order (hint, then age chip, then host; the
                # clock and the tag never).
                title = header_title(
                    painted_size[0], socket.gethostname(), time.strftime("%H:%M:%S"),
                    age_secs=(None if _LAST_COLLECT is None
                              else time.time() - _LAST_COLLECT),
                    keys_enabled=keys.enabled, interactive=interactive, view=view,
                    tagged=keys.enabled and interactive)
                header = [title, status, ""]
                paint(header + lines, vt)
                painted_at = time.time()

                # Sleep in slices so a keypress lands within ~0.2s rather than
                # at the end of the tick. Each idle slice also samples the
                # terminal size: conhost has no SIGWINCH, so a polled compare
                # is the portable resize signal. While a drag keeps the size
                # moving, the loop repaints live from the cached snapshot --
                # ResizeThrottle caps that at one paint per ~120ms and
                # shortens the slices so those paints actually land -- and
                # the mismatch left behind when the drag stops buys the final
                # paint at the settled size. Independently, a repaint tick
                # fires once a second (tick_action) so the clock, the age
                # chip and the loading dots move between collects. Every such
                # repaint is pure render-from-cache: no collect runs and the
                # collect deadline is left where it was, so the "updated Ns
                # ago" chip keeps telling the truth.
                while True:
                    now = time.time()
                    action = tick_action(now, deadline, painted_at)
                    if action == "collect":
                        collect_due = True
                        break
                    if action == "repaint":
                        break  # render-from-cache; collect_due stays False
                    key = keys.get(wait_slice(now, deadline, painted_at,
                                              resize.slice_len(now)))
                    if key is None:
                        if resize.poll(term_size(), painted_size, time.time()):
                            break  # live repaint from cache at the current size
                        continue
                    note = None
                    collect_due = True  # keypresses repaint from fresh data

                    # The confirmation swallows every key: only an explicit y
                    # stops a session, and q here cancels rather than quitting so
                    # that a reflexive quit cannot be read as consent.
                    if pending is not None:
                        if key in ("y", "Y"):
                            if pending.get("source") == "cursor" or pending.get("pid") is None:
                                note = c("cursor composers cannot be stopped from roost", YELLOW)
                            else:
                                err = terminate(pending["pid"])
                                log_action("stop", pending, ok=err is None, detail=err or "")
                                note = c("stopped %s (pid %d)" % (
                                    pending["name"], pending["pid"]), GREEN) if err is None \
                                    else c(err, BOLD, RED)
                        else:
                            note = c("cancelled", DIM)
                        pending = None
                        break

                    if key in ("q", "Q", "\x03"):
                        return
                    if key == " ":
                        break  # repaint now
                    if key == "ESC":
                        if view == "detail":
                            view = "agents" if focus == "agents" else None
                            detail = None
                        else:
                            sel = None
                        break
                    if key in ("a", "A"):
                        view = None if view == "advice" else "advice"
                        detail = None
                        break  # repaint immediately, do not wait out the tick
                    if key in ("s", "S"):
                        view = None if view == "agents" else "agents"
                        if view != "agents" and focus == "agents":
                            focus = "workers"
                            sel = None
                        detail = None
                        break
                    if key in ("m", "M"):
                        view = None if view == "models" else "models"
                        detail = None
                        break
                    if key in ("u", "U"):
                        view = None if view == "usage" else "usage"
                        detail = None
                        break
                    if key in ("g", "G"):
                        view = None if view == "gateway" else "gateway"
                        detail = None
                        break
                    if key in ("r", "R"):
                        view = None if view == "remote" else "remote"
                        detail = None
                        break
                    if key in ("h", "H", "?"):
                        view = None if view == "help" else "help"
                        detail = None
                        break
                    if key in ("i", "I"):
                        # The one key that arms the whole risky half at once --
                        # cursor, x/y, and the EXPERIMENTAL tag all come alive
                        # together, so there is exactly one thing to remember
                        # before a keypress can end a process.
                        interactive = not interactive
                        if interactive:
                            note = c("interactive armed -- j/k select, x stop, y yank", BOLD, YELLOW)
                        else:
                            # Drop the cursor rather than leave it parked: a
                            # stale sel would resurface on re-arming, pointing
                            # at whatever row happens to occupy that index by
                            # then, not the one it was left on.
                            sel = None
                            focus = "workers"
                            detail = None
                            note = c("interactive off -- view only", DIM)
                        break
                    if key in ("\t",):
                        if not interactive:
                            note = c("press i to arm interactive mode first", YELLOW)
                        else:
                            focus, view = tab_table_focus(focus, view)
                            sel = 0
                            detail = None
                            note = c("focus %s -- Enter for detail" % focus, DIM)
                        break
                    if key in ("\r", "\n"):
                        if not interactive:
                            note = c("press i to arm interactive mode first", YELLOW)
                        elif sel is None or not rows:
                            note = c("select a row first -- j/k or the arrow keys", YELLOW)
                        else:
                            detail = rows[sel]
                            view = "detail"
                        break
                    if key in ("j", "J", "DOWN"):
                        if not interactive:
                            note = c("press i to arm interactive mode first", YELLOW)
                        elif view == "detail":
                            note = c("esc to leave detail first", DIM)
                        else:
                            # Unbounded on purpose -- frame() clamps against the row
                            # count it actually rendered, which is the only correct one.
                            sel = 0 if sel is None else sel + 1
                        break
                    if key in ("k", "K", "UP"):
                        if not interactive:
                            note = c("press i to arm interactive mode first", YELLOW)
                        elif view == "detail":
                            note = c("esc to leave detail first", DIM)
                        else:
                            sel = 0 if sel is None else max(0, sel - 1)
                        break
                    if key in ("x", "X", "y", "Y"):
                        # All three need interactive armed, and x/y also need a
                        # row. Saying so beats doing nothing: a key that silently
                        # no-ops is indistinguishable from a broken one.
                        if not interactive:
                            note = c("press i to arm interactive mode first", YELLOW)
                        elif view == "detail":
                            note = c("detail is read-only -- esc to return", DIM)
                        elif sel is None or not rows:
                            note = c("select a row first -- j/k or the arrow keys", YELLOW)
                        elif focus == "agents" or (
                                rows[sel].get("agent_id")
                                and "parent_sid" in rows[sel]):
                            note = c("subagents are read-only in roost", YELLOW)
                        elif key in ("x", "X"):
                            w = rows[sel]
                            if w.get("source") == "cursor" or w.get("pid") is None:
                                note = c("cursor composers cannot be stopped from roost", YELLOW)
                            else:
                                pending = w
                        else:
                            w = rows[sel]
                            if w.get("source") == "cursor":
                                msg = "copied composer %s" % w["session_id"]
                            else:
                                msg = "copied %s -- claude --resume <paste>" % w["name"]
                            note = c(msg, GREEN) \
                                if to_clipboard(w["session_id"]) \
                                else c("no clipboard helper (%s not found)" % CLIP_CMD[0], YELLOW)
                        break
    except KeyboardInterrupt:
        pass
    finally:
        if vt:
            sys.stdout.write("\033[?25h\n")  # restore the cursor on the way out
            sys.stdout.flush()
        restore_vt()


if __name__ == "__main__":
    main()