jbeda · August 1, 2026 23:13
diff --git a/README.md b/README.md
diff --git a/agentproc.py b/agentproc.py
 """File-based introspection of Claude Code subagents — "/proc for agents".

 Single parser for the per-session sidecar directory Claude Code writes next to
 each transcript (format contract + design decisions: notes/agentproc-2026-07-22.md
 in this repo; verified empirically on Claude Code 2.1.219):

    <config>/projects/<slug>/<session-id>.jsonl        # parent transcript
    <config>/projects/<slug>/<session-id>/subagents/
        agent-<id>.meta.json    # identity, written once at spawn (mtime = spawn)
        agent-<id>.jsonl        # live-appended transcript (mtime = last activity)
        workflows/<run-id>/     # agents spawned by the Workflow tool, same pair
            journal.jsonl       # append-only started/result log for the run

 Completion has three shapes. For Agent-tool subagents it is only visible in the
 *parent* transcript: a `<task-notification>` whose task-id equals the agent id
 (background agents), or a real tool_result for the meta's toolUseId (sync agents
 — distinguished from the background launch ack, whose content starts with "Async
 agent launched successfully"). Workflow agents have neither field; their
 completion is a `{"type":"result","agentId":...}` line in their run's
 journal.jsonl, backstopped by the run's own task-notification. The agent's own
 transcript has no terminal record in any case.

 Consumers: statusline/statusline.py (same dir, `import agentproc`) and
 bin/agent-ps. Stdlib only.
 """

 import datetime
 import json
 import os
 import re

 ASYNC_ACK = "Async agent launched successfully"
 STOPPED_MSG = "Successfully stopped task:"

 # The scan cache stores per-file byte offsets, so a changed parser would
 # otherwise never re-read bytes an older one already consumed — leaving stale
 # state pinned forever. The cache is stamped with this value and reset on any
 # mismatch, forcing one full re-scan.
 #
 # It is derived from this module's own mtime/size rather than hand-bumped: a
 # hand-bumped constant has to be edited in the *same* write as the parser
 # change, or a live 1 Hz consumer (the statusline) runs the in-between state and
 # persists a cache stamped new but produced by the old parser. That is not
 # theoretical — it happened on 2026-07-24 and pinned two phantom agents that
 # survived the very fix written to clear them. Editing this file now invalidates
 # every cache automatically; the cost is one full re-scan.
 _SCHEMA = 3  # bump only to force invalidation without touching the file


 def _cache_version():
    try:
        st = os.stat(__file__)
        return f"{_SCHEMA}:{int(st.st_mtime)}:{st.st_size}"
    except OSError:
        return f"{_SCHEMA}:?"


 CACHE_VERSION = _cache_version()

 # Statuses
 RUNNING = "running"
 DONE = "done"
 IDLE = "idle"  # in-process teammates only — see _teammate_idle_in_line

 # A teammate's idle notification lands at the same instant its transcript stops
 # growing (observed equal to the second), but the two clocks are different
 # sources, so allow a little slack before calling an append "newer".
 IDLE_EPS = 2.0


 def sidecar_dir(transcript_path):
    """Session sidecar directory for a transcript path, or None."""
    if not transcript_path or not transcript_path.endswith(".jsonl"):
        return None
    return transcript_path[: -len(".jsonl")]


 def _read_metas(subdir):
    """{agent_id: meta dict + _spawn_ts/_meta_path/_dir} for every agent ever
    spawned in this session (meta files are written once, at spawn).

    Walks the whole tree: Agent-tool subagents sit directly in `subagents/`,
    Workflow agents in `subagents/workflows/<run-id>/`. `_dir` records which,
    since it locates the agent's transcript and its run journal."""
    metas = {}
    for root, _dirs, names in os.walk(subdir):
        for name in names:
            if not (name.startswith("agent-") and name.endswith(".meta.json")):
                continue
            aid = name[len("agent-"): -len(".meta.json")]
            path = os.path.join(root, name)
            try:
                with open(path) as f:
                    meta = json.load(f)
                meta["_spawn_ts"] = os.path.getmtime(path)
            except Exception:
                continue
            meta["_meta_path"] = path
            meta["_dir"] = root
            metas[aid] = meta
    return metas


 def _load_cache(cache_path):
    """Prior scan state {offsets: {path: byte_offset}, done: [...], runs: {...}}.

    Per-path offsets because completions are spread across several files (the
    session root, each parent subagent's own transcript, and each workflow run's
    journal). `runs` maps a workflow run's transcript dir to the background task
    id that reports the whole run finished. A rotated/truncated individual file
    is handled in the scan loop, not here. Any unreadable or old-schema cache
    resets to empty — the cache is a pure optimization, so a reset just forces a
    full re-scan."""
    empty = {"offsets": {}, "done": [], "runs": {}, "idles": {}}
    if not cache_path:
        return empty
    try:
        with open(cache_path) as f:
            cache = json.load(f)
        got = {k: cache.get(k) for k in ("offsets", "done", "runs", "idles")}
        if not (isinstance(got["offsets"], dict) and isinstance(got["done"], list)
                and isinstance(got["runs"], dict) and isinstance(got["idles"], dict)
                and cache.get("v") == CACHE_VERSION):
            raise ValueError
        return got
    except Exception:
        return empty


 def _mark_done_from_transcript_line(line, tid_to_aid, done, runs, idles):
    """Record what this parent-transcript line reports finished.

    Two completion shapes, each behind a cheap substring gate before any JSON
    parse:
      * background — a `<task-notification>` carrying `<task-id>`; the id is an
        agent id for an Agent-tool spawn and a task id for a whole Workflow run,
        so both go into `done` and the caller decides which it matched;
      * sync — a real tool_result for a spawning toolUseId (not the launch ack,
        which `_tool_results_in_line` filters out).

    Also harvests the Workflow launch record's transcriptDir -> taskId mapping
    into `runs`, which is what lets a finished run sweep its own agents done."""
    if "<task-id>" in line:
        i = line.find("<task-id>")
        j = line.find("</task-id>", i)
        if j > i:
            tid = line[i + len("<task-id>"): j].replace("\\n", "").strip()
            if tid:
                done.add(tid)
    if "tool_use_id" in line and any(t in line for t in tid_to_aid):
        done.update(_tool_results_in_line(line, tid_to_aid))
    if '"transcriptDir"' in line:
        runs.update(_workflow_runs_in_line(line))
    if '"task_id"' in line:
        done.update(_stopped_tasks_in_line(line))
    if "idle_notification" in line:
        for who, ts in _teammate_idle_in_line(line).items():
            if ts > idles.get(who, 0):
                idles[who] = ts


 def _mark_done_from_journal_line(line, done):
    """Add the agent id this workflow-journal line reports as complete.

    The run's journal is append-only with `started` then `result` records per
    agent; only `result` means finished. An agent whose result was null (died on
    a terminal error, or the user skipped it) never gets a `result` line at all —
    the run-level sweep in scan() is what eventually clears those."""
    if '"result"' not in line:
        return
    try:
        rec = json.loads(line)
    except Exception:
        return
    if rec.get("type") == "result" and rec.get("agentId"):
        done.add(rec["agentId"])


 _IDLE_RE = re.compile(r'\{[^{}]*"type"\s*:\s*"idle_notification"[^{}]*\}')


 def _record_texts(rec):
    """Every free-text body in a transcript record (str content, text blocks,
    and string tool_result contents)."""
    msg = rec.get("message")
    content = msg.get("content") if isinstance(msg, dict) else None
    if isinstance(content, str):
        yield content
    elif isinstance(content, list):
        for block in content:
            if not isinstance(block, dict):
                continue
            for key in ("text", "content"):
                if isinstance(block.get(key), str):
                    yield block[key]
    if isinstance(rec.get("content"), str):
        yield rec["content"]


 def _teammate_idle_in_line(line):
    """{teammate name: idle timestamp} announced by this line.

    In-process teammates (`taskKind: in_process_teammate`, spawned into a team)
    are peers, not tasks: they finish a piece of work, report it, and sit
    *idle-available* waiting to be messaged again, then wake and work more. They
    therefore never produce a completion marker, and treating them as RUNNING
    forever overcounts every teammate that has ever been spawned.

    What they do emit, into the spawning session's transcript, is

        <teammate-message teammate_id="pr21-review" color="blue">
        {"type":"idle_notification","from":"pr21-review","timestamp":"...Z",
         "idleReason":"available"}
        </teammate-message>

    keyed by teammate **name** (not agent id), with the JSON embedded in a text
    body — so it needs a decoded-string search, not a raw-line one. Comparing
    the latest such timestamp against the agent transcript's mtime gives idle vs
    working precisely, rather than by a staleness threshold: a teammate that has
    appended since its last idle notification is working, however long ago that
    was."""
    found = {}
    try:
        rec = json.loads(line)
    except Exception:
        return found
    for text in _record_texts(rec):
        if "idle_notification" not in text:
            continue
        for match in _IDLE_RE.finditer(text):
            try:
                note = json.loads(match.group(0))
                who = note["from"]
                stamp = note.get("timestamp") or rec.get("timestamp")
                ts = datetime.datetime.fromisoformat(
                    str(stamp).replace("Z", "+00:00")).timestamp()
            except Exception:
                continue
            if not found.get(who) or ts > found[who]:
                found[who] = ts
    return found


 def _first_record_ts(path, max_lines=25):
    """Timestamp of the first timestamped record in a transcript, or None.

    Needed because a teammate's `meta.json` is **rewritten when it is woken by a
    message**, so its mtime is the last wake, not the spawn — measured drift of
    +829s on a teammate re-messaged 14 minutes after spawning, against exactly 0s
    for every ordinary subagent. The transcript's first record is the durable
    spawn time. Reads only the head of the file."""
    try:
        with open(path, errors="replace") as f:
            for _, line in zip(range(max_lines), f):
                if '"timestamp"' not in line:
                    continue
                try:
                    stamp = json.loads(line).get("timestamp")
                    return datetime.datetime.fromisoformat(
                        str(stamp).replace("Z", "+00:00")).timestamp()
                except Exception:
                    continue
    except OSError:
        pass
    return None


 def _stopped_tasks_in_line(line):
    """Task ids killed by a `TaskStop` in this line.

    A killed task emits **no** `<task-notification>` at all (verified 2026-07-24
    on a TaskStop'd workflow run) — this tool_result is the only marker it
    leaves, and its `task_id` is snake_case, unlike the `taskId` in a launch
    record. Covers both a whole Workflow run (`task_type: local_workflow`, which
    then sweeps its agents) and a background Agent spawn (task_id == agent id).
    Matched narrowly on TaskStop's success message, since TaskList/TaskGet
    output also carries `task_id` fields — for tasks that are still running."""
    try:
        rec = json.loads(line)
    except Exception:
        return set()
    res = rec.get("toolUseResult")
    if not isinstance(res, dict):
        return set()
    tid, msg = res.get("task_id"), res.get("message")
    if tid and isinstance(msg, str) and msg.startswith(STOPPED_MSG):
        return {tid}
    return set()


 def _workflow_runs_in_line(line):
    """{run transcript dir: background task id} from a Workflow launch record.

    The Workflow tool returns immediately; its toolUseResult carries the run's
    sidecar dir alongside the task id that a later `<task-notification>` uses to
    announce the whole run finished. Keyed by dir rather than runId so the match
    is exact, with no basename assumption."""
    try:
        rec = json.loads(line)
    except Exception:
        return {}
    res = rec.get("toolUseResult")
    if not isinstance(res, dict):
        return {}
    tdir, task = res.get("transcriptDir"), res.get("taskId")
    return {tdir: task} if tdir and task else {}


 def _tool_results_in_line(line, tid_to_aid):
    """Agent ids completed by tool_result blocks in this transcript line.

    A tool_result for a spawning toolUseId means the agent finished — except
    the background-launch ack, whose content carries ASYNC_ACK."""
    finished = set()
    try:
        rec = json.loads(line)
    except Exception:
        return finished
    msg = rec.get("message")
    content = msg.get("content") if isinstance(msg, dict) else None
    if not isinstance(content, list):
        return finished
    for block in content:
        if not (isinstance(block, dict) and block.get("type") == "tool_result"):
            continue
        aid = tid_to_aid.get(block.get("tool_use_id"))
        if not aid:
            continue
        if ASYNC_ACK in json.dumps(block.get("content", "")):
            continue  # launch ack, not completion
        finished.add(aid)
    return finished


 def _scan(transcripts, journals, metas, cache_path=None):
    """(done ids, {run dir: task id}) from every completion source.

    An Agent-tool subagent's completion is recorded in the transcript of
    whichever agent spawned it: the session root for depth-1 agents, and a
    *parent subagent's* own transcript (`agent-<parentAgentId>.jsonl`) for
    anything deeper. Workflow agents are recorded in their run's journal
    instead. Scanning only the root therefore leaves nested and workflow agents
    pinned RUNNING forever even after they finish — the caller passes the root
    plus every parent transcript plus every run journal. Incremental per file:
    only bytes appended past each file's cached offset are read; `done` and
    `runs` accumulate across calls and files."""
    tid_to_aid = {
        m["toolUseId"]: aid for aid, m in metas.items() if m.get("toolUseId")
    }
    cache = _load_cache(cache_path)
    done = set(cache["done"])
    runs = dict(cache["runs"])
    idles = dict(cache["idles"])
    offsets = dict(cache["offsets"])

    sources = [(transcripts,
                lambda ln: _mark_done_from_transcript_line(
                    ln, tid_to_aid, done, runs, idles)),
               (journals,
                lambda ln: _mark_done_from_journal_line(ln, done))]
    for paths, handle in sources:
        for path in paths:
            try:
                size = os.path.getsize(path)
            except OSError:
                continue
            off = offsets.get(path, 0)
            if off > size:  # file rotated/truncated — re-read from the top
                off = 0
            if off >= size:
                continue
            try:
                with open(path, errors="replace") as f:
                    f.seek(off)
                    # Offsets we write are always line boundaries (we read to EOF
                    # in whole lines) and keyed by path, so a seek never lands
                    # mid-line.
                    for line in f:
                        handle(line)
                    offsets[path] = f.tell()
            except OSError:
                continue

    if cache_path:
        try:
            with open(cache_path, "w") as f:
                json.dump({"v": CACHE_VERSION, "offsets": offsets,
                           "done": sorted(done), "runs": runs,
                           "idles": idles}, f)
        except OSError:
            pass
    return done, runs, idles


 def scan(transcript_path, cache_path=None):
    """List all subagents of a session, each a dict:

        id, status (RUNNING/DONE), spawn_ts, last_ts,
        agentType, description, model, spawnDepth, toolUseId, worktreePath,
        workflow (run id, or None for an Agent-tool spawn)

    Returns None when the transcript has no sidecar (can't tell — caller
    should fall back), and [] when the sidecar exists but has no agents.
    Never raises. `cache_path` enables the incremental scan (recommended for
    1 Hz callers like the statusline renderer)."""
    side = sidecar_dir(transcript_path)
    if not side or not os.path.isdir(side):
        return None
    subdir = os.path.join(side, "subagents")
    if not os.path.isdir(subdir):
        return []
    metas = _read_metas(subdir)
    if not metas:
        return []
    # An agent's transcript sits next to its meta, wherever that is.
    transcript_of = {
        aid: os.path.join(m["_dir"], f"agent-{aid}.jsonl") for aid, m in metas.items()
    }
    # Completions live in the transcript of whichever agent spawned each
    # subagent: the session root for depth-1, a parent subagent's own transcript
    # for deeper nesting. Scan the root plus every parent transcript present in
    # the sidecar. This covers arbitrary depth — a deep agent's parent is itself
    # a subagent whose transcript sits alongside its own meta.
    parent_ids = {
        m.get("parentAgentId") for m in metas.values() if m.get("parentAgentId")
    }
    parent_transcripts = [
        p for p in (transcript_of.get(pid) for pid in sorted(parent_ids))
        if p and os.path.exists(p)
    ]
    # Workflow runs report per-agent completion in their own journal.
    run_dirs = sorted({
        m["_dir"] for m in metas.values()
        if os.path.exists(os.path.join(m["_dir"], "journal.jsonl"))
    })
    journals = [os.path.join(d, "journal.jsonl") for d in run_dirs]
    done, runs, idles = _scan(
        [transcript_path, *parent_transcripts], journals, metas, cache_path)

    # Run-level sweep: once the whole Workflow run is announced finished, none
    # of its agents can still be alive. Needed because an agent whose result was
    # null never gets a journal `result` line, and because a killed run leaves
    # its in-flight agents with no marker at all.
    for d in run_dirs:
        task = runs.get(d)
        if task and task in done:
            done.update(aid for aid, m in metas.items() if m["_dir"] == d)

    agents = []
    for aid, meta in metas.items():
        jsonl = transcript_of[aid]
        try:
            last_ts = os.path.getmtime(jsonl)
        except OSError:
            last_ts = meta["_spawn_ts"]
        run_id = (os.path.basename(meta["_dir"])
                  if os.path.join(meta["_dir"], "journal.jsonl") in journals else None)
        # A teammate has no completion marker; it reports going idle-available
        # instead, and can be woken again later. Idle iff it has not appended to
        # its transcript since its latest idle notification.
        teammate = meta.get("taskKind") == "in_process_teammate"
        idle_ts = idles.get(meta.get("name")) if teammate else None
        spawn_ts = meta["_spawn_ts"]
        if teammate:  # meta mtime is the last wake, not the spawn
            spawn_ts = _first_record_ts(jsonl) or spawn_ts
        if aid in done:
            status = DONE
        elif idle_ts is not None and last_ts <= idle_ts + IDLE_EPS:
            status = IDLE
        else:
            status = RUNNING
        agents.append({
            "id": aid,
            "status": status,
            "spawn_ts": spawn_ts,
            "last_ts": last_ts,
            "agentType": meta.get("agentType"),
            "description": meta.get("description"),
            "model": meta.get("model"),
            "spawnDepth": meta.get("spawnDepth"),
            "toolUseId": meta.get("toolUseId"),
            "worktreePath": meta.get("worktreePath"),
            "workflow": run_id,
            "name": meta.get("name"),
            "taskKind": meta.get("taskKind"),
            "transcript": jsonl,
        })
    # Sorted on the corrected spawn time, so a woken teammate keeps its place.
    agents.sort(key=lambda a: a["spawn_ts"])
    return agents


 def inflight(agents):
    """The RUNNING subset of a scan() result (empty for None).

    Excludes IDLE teammates: they are alive but not working, so counting them
    as in flight overstates what is churning (see _teammate_idle_in_line)."""
    return [a for a in (agents or []) if a["status"] == RUNNING]


 def unfinished(agents):
    """RUNNING plus IDLE — everything still alive, working or not."""
    return [a for a in (agents or []) if a["status"] in (RUNNING, IDLE)]


 def last_usage(agent_transcript, tail_bytes=65536):
    """Best-effort (model, usage-dict) from the last assistant record of an
    agent transcript, reading only the file tail. (None, None) on failure."""
    try:
        size = os.path.getsize(agent_transcript)
        with open(agent_transcript, errors="replace") as f:
            f.seek(max(0, size - tail_bytes))
            lines = f.read().splitlines()
    except OSError:
        return None, None
    for line in reversed(lines):
        if '"usage"' not in line:
            continue
        try:
            rec = json.loads(line)
        except Exception:
            continue
        msg = rec.get("message")
        if isinstance(msg, dict) and msg.get("usage"):
            return msg.get("model"), msg["usage"]
    return None, None
diff --git a/statusline.png b/statusline.png
diff --git a/statusline.py b/statusline.py
 #!/usr/bin/env python3
 """Claude Code powerline status line + turn-state hook (single file, no deps).

 Two modes:

  * Render mode (default): Claude Code invokes this with the status-line JSON on
    stdin and renders whatever we print. See
    https://code.claude.com/docs/en/statusline.md for the input schema.

  * Hook mode: `statusline.py hook <start|resume|blocked|done>` — invoked by
    several Claude Code hooks (see statusline/README.md). Reads session_id from
    the hook stdin JSON and stamps a per-session state file that render mode
    reads back to show whether the agent is churning, blocked waiting on you, or
    done (your move), plus how long the current turn has been running.
      start   — UserPromptSubmit (new turn; reset the elapsed clock)
      resume  — PostToolUse, any tool (back to churning; also clears "blocked"
                after you grant a plain permission)
      blocked — PermissionRequest / PreToolUse for those tools (paused on you)
      done    — Stop / StopFailure / Notification idle_prompt (main turn over)
      bgcount — SubagentStop (refresh the in-flight background-agent count)
      session — SessionStart (name the tmux window before the first prompt)
      ended   — SessionEnd (drop the title marker, leave the plain repo name)

 The same state also drives the tmux window title (see set_tmux_title): one
 claude per repo means one window per repo, titled by repo name (worktree-aware)
 with a leading marker for every live state — churning, background-agents-only,
 blocked, your-move — so you can scan many windows at a glance.

 If the main turn is done but background agents are still running, the
 indicator stays "busy" — a cyan gear + count + running timer — and only flips
 to green idle once every background agent has finished. Agent count and spawn
 times come from the session's on-disk sidecar dir via agentproc.py (the
 "/proc for agents"; see notes/agentproc-2026-07-22.md) when the statusline
 stdin carries transcript_path; the hook-tracked agent set is the fallback.
 The timer in that state is anchored to the oldest in-flight agent's spawn
 (min'd with turn_start), not just turn_start: intervening chat turns reset
 turn_start, and that must not restart the clock on a long-running bg agent
 from an earlier prompt.

 Colors are Catppuccin Mocha, matching the tmux bar in ~/src/dotfiles. Truecolor
 ANSI, so it needs a terminal that supports 24-bit color (Ghostty does).
 """

 import json
 import os
 import re
 import socket
 import subprocess
 import sys
 import tempfile
 import time
 import unicodedata

 # ---- Catppuccin Mocha palette (matches ~/src/dotfiles tmux bar) -----------
 # Mocha's accents are pastel — LIGHT. So the two directions are not symmetric:
 # an accent used as *text* sits on a dark segment bg, and an accent used as a
 # segment *bg* needs CRUST as its text, never a light base. Same rule the tmux
 # styles in ~/src/dotfiles/link/.tmux.conf follow.
 CRUST    = (0x11, 0x11, 0x1b)  # text on a light accent bg
 MANTLE   = (0x18, 0x18, 0x25)  # dark segment bg (matches the terminal field)
 SURFACE0 = (0x31, 0x32, 0x44)  # bg highlight (tmux status bar bg)
 SURFACE1 = (0x45, 0x47, 0x5a)
 OVERLAY0 = (0x6c, 0x70, 0x86)  # secondary / muted text
 SUBTEXT0 = (0xa6, 0xad, 0xc8)  # body text (tmux status fg)
 TEXT     = (0xcd, 0xd6, 0xf4)  # emphasized text
 YELLOW   = (0xf9, 0xe2, 0xaf)
 PEACH    = (0xfa, 0xb3, 0x87)  # "orange" accent
 RED      = (0xf3, 0x8b, 0xa8)
 PINK     = (0xf5, 0xc2, 0xe7)
 MAUVE    = (0xcb, 0xa6, 0xf7)  # "violet" accent
 BLUE     = (0x89, 0xb4, 0xfa)  # tmux session segment
 TEAL     = (0x94, 0xe2, 0xd5)  # "cyan" accent
 GREEN    = (0xa6, 0xe3, 0xa1)  # tmux current-window segment

 # ---- Nerd Font glyphs (all confirmed rendering in Ghostty) ----------------
 SEP        = ""  # powerline right arrow (solid)
 G_RUN      = ""  # hourglass-half  (churning)
 G_ATTN     = ""  # bell            (waiting on you)
 G_IDLE     = ""  # check           (turn done)
 G_REPO     = ""  # git logo
 G_BRANCH   = ""  # branch
 G_DIR      = ""  # folder
 G_HOST     = ""  # server (hostname)
 G_BG       = ""  # cogs — background agents still running
 G_CTX      = ""  # microchip       (context)
 G_MODEL    = ""  # cogs            (model)
 G_COST     = ""  # dollar
 DIRTY      = "✱"  # heavy asterisk (uncommitted)


 def esc(fg=None, bg=None):
    out = ""
    if fg:
        out += f"\x1b[38;2;{fg[0]};{fg[1]};{fg[2]}m"
    if bg:
        out += f"\x1b[48;2;{bg[0]};{bg[1]};{bg[2]}m"
    return out


 RESET = "\x1b[0m"


 def bg_subagent_ids(data, exclude_id=None):
    """IDs of in-flight background *subagents* from a hook's background_tasks.

    Present on Stop / SubagentStop stdin (v2.1.145+). Returns None when the key
    is absent (i.e. this hook can't tell us), so callers can keep prior info
    instead of clobbering it.

    Two payload quirks (verified empirically on 2.1.217, see HANDOFF):
      * entries carry a `status` field — only "running" ones are in-flight
        (absent status counts as running, for older versions);
      * at an agent's own SubagentStop, background_tasks still lists THAT agent
        as "running" — callers pass its agent_id as exclude_id or the count
        never reaches zero from SubagentStop alone.
    """
    tasks = data.get("background_tasks")
    if tasks is None:
        return None
    ids = []
    for i, t in enumerate(tasks):
        if not isinstance(t, dict) or t.get("type") != "subagent":
            continue
        if t.get("status", "running") != "running":
            continue
        tid = t.get("id") or f"bg{i}"
        if exclude_id and tid == exclude_id:
            continue
        ids.append(tid)
    return ids


 def repo_name(cwd):
    """Repo name, worktree-friendly: the *main* working tree's directory name.

    `--show-toplevel` returns the *worktree* root, whose basename is usually the
    branch name (wt/worktrunk name linked worktrees after their branch) — not
    what we want in the window title. `--git-common-dir` instead points at the
    main repo's `.git` even from inside a linked worktree, so its parent is the
    canonical repo directory.

    Returns None outside a git repo, rather than the cwd basename: callers that
    want a label can fall back themselves, while the tmux title uses None to
    mean "don't rename" — otherwise a transient cwd (a Bash `cd` into a scratch
    dir) would rename the window after a temp directory.
    """
    common = git(cwd, "rev-parse", "--git-common-dir")
    if common:
        # --git-common-dir may be relative (".git") in the main worktree; join
        # against cwd and normalize so both main and linked worktrees resolve to
        # <repo>/.git, whose parent is the repo dir.
        common = os.path.abspath(os.path.join(cwd, common))
        name = os.path.basename(os.path.dirname(common))
        if name:
            return name
    return None


 def set_tmux_title(st, cwd, plain=False):
    """Reflect turn state into the tmux window title (one claude per repo).

    Only acts inside tmux. Renames the *window* holding this pane so the marker
    shows up in the window list even when the pane isn't focused — the whole
    point being to spot, at a glance across many windows, which agents need you:

        ● repo   blocked on you (permission / question / plan approval)
        ✓ repo   turn done, your move (or a background-free idle)
        ▶ repo   main turn churning
        ⚙ repo   main turn done, background agents still running

    Every live state carries a marker: "busy" used to be the *absence* of one,
    which meant reading the window list as "no glyph = running" — exactly the
    thing you can't spot at a glance. Only SessionEnd (`plain`) leaves the bare
    repo name, so a marker-free window means no live agent at all.

    Uses `tmux rename-window`, which also disables automatic-rename for that
    window, so the name sticks instead of being overwritten by the running
    process name. One claude per repo means one window per repo, so the repo
    name is a stable, unambiguous label.

    The repo name and last-set title are cached in `st` so the (now catch-all)
    resume hook doesn't shell out to git or tmux on every single tool call —
    only an actual title change re-renames. `plain` forces the marker-free label
    (used by SessionEnd to leave the window cleanly named after the repo).
    """
    pane = os.environ.get("TMUX_PANE")
    if not os.environ.get("TMUX") or not pane:
        return
    # Cache keyed by the cwd it was derived from, and only ever accept a real
    # git repo. A tool call that leaves the session's cwd somewhere else briefly
    # (a Bash `cd` into the scratchpad) must not permanently retitle the window:
    # before this, the first computed value won forever via `st.get("repo") or
    # ...`, so one detour left the window named "scratchpad" for the rest of the
    # session.
    if cwd and st.get("repo_cwd") != cwd:
        found = repo_name(cwd)
        if found:
            st["repo"], st["repo_cwd"] = found, cwd
    repo = st.get("repo")
    if not repo:
        return
    state = st.get("state", "idle")
    bg = max(st.get("bg", 0) or 0, len(st.get("agents") or {}))
    if plain:
        title = repo
    elif state == "attention":
        title = f"● {repo}"
    elif state == "idle" and not bg:
        title = f"✓ {repo}"
    elif state == "idle":  # main turn over, background agents still going
        title = f"⚙ {repo}"
    else:  # churning — don't interrupt
        title = f"▶ {repo}"
    if st.get("wtitle") == title:
        return
    try:
        subprocess.run(
            ["tmux", "rename-window", "-t", pane, title],
            capture_output=True, timeout=1.0,
        )
        st["wtitle"] = title
    except Exception:
        pass


 def state_path(session_id):
    d = os.path.join(tempfile.gettempdir(), "claude-statusline")
    os.makedirs(d, exist_ok=True)
    sid = session_id or "unknown"
    # keep it filesystem-safe
    sid = "".join(c if c.isalnum() or c in "-_" else "_" for c in sid)
    return os.path.join(d, f"{sid}.json")


 # --------------------------------------------------------------------------
 # Hook mode: stamp the per-session state file.
 # --------------------------------------------------------------------------
 def run_hook(action):
    try:
        data = json.load(sys.stdin)
    except Exception:
        data = {}
    # Debug capture (opt-in): `touch $TMPDIR/claude-statusline/debug-on` to
    # append every hook invocation's raw payload to debug.jsonl in the same dir.
    # Used to inspect what fields each hook actually carries (background_tasks,
    # subagent events, ...) on the current Claude Code version. Remove the flag
    # file to stop; the log is never read by the renderer.
    dbg_dir = os.path.join(tempfile.gettempdir(), "claude-statusline")
    if os.path.exists(os.path.join(dbg_dir, "debug-on")):
        try:
            with open(os.path.join(dbg_dir, "debug.jsonl"), "a") as f:
                json.dump({"t": time.time(), "action": action, "payload": data}, f)
                f.write("\n")
        except Exception:
            pass
    session_id = data.get("session_id", "")
    path = state_path(session_id)
    cwd = data.get("cwd") or os.getcwd()
    now = time.time()

    prev = {}
    try:
        with open(path) as f:
            prev = json.load(f)
    except Exception:
        pass

    # carry prior fields forward; individual actions override what they touch
    st = dict(prev)
    st.setdefault("bg", 0)
    # `agents` — subagents (sync AND background) currently believed in flight:
    # {agent_id: {"first": ts, "last": ts}} — first and latest sighting.
    # Populated by their tool calls (subagent tool use fires the parent's
    # catch-all PostToolUse, tagged with agent_id — verified on 2.1.217), pruned
    # by SubagentStop / Stop recounts. This is the only signal that sees
    # *synchronous* subagents, which never appear in background_tasks. "first"
    # anchors the cyan timer to how long the oldest agent has been running
    # (rather than the latest prompt's turn_start — see render()).
    agents = {}
    for i, v in (st.get("agents") or {}).items():
        # migrate pre-2026-07-22 state files that stored a bare last_seen float
        agents[i] = v if isinstance(v, dict) else {"first": v, "last": v}
    st["agents"] = agents

    # who generated this hook event: a subagent's tool call carries agent_id
    aid = data.get("agent_id")

    # in-flight *background* subagent ids, if this payload carries them
    # (Stop / SubagentStop include background_tasks). None => no info.
    bg_ids = bg_subagent_ids(data, exclude_id=aid)

    if action == "session":
        # SessionStart — no active turn yet; present as idle "your move". Drop
        # any stale fields but keep a cached repo/wtitle so we don't re-rename.
        st = {"state": "idle", "bg": 0, "agents": {},
              "repo": prev.get("repo"), "wtitle": prev.get("wtitle")}
    elif action == "ended":
        # SessionEnd — agent gone; leave the window cleanly named after the repo
        # (marker dropped). State value is irrelevant, title is forced plain.
        st["state"] = "idle"
    elif action == "start":
        # new turn (UserPromptSubmit) — reset the elapsed clock. Keep `agents`:
        # a bg agent from a prior turn is still in flight, and wiping it here
        # would lose the first-seen time that anchors its timer (the bug where
        # the bar showed 1:59 while the agents panel showed 19m42s — every new
        # prompt restarted the clock). Ghost entries (an agent that died without
        # its SubagentStop, e.g. on an interrupt) are cleared by the next Stop
        # recount, which fully replaces the set.
        st.update(state="running", turn_start=now, bg=0)
        st.pop("blocked_by", None)  # you just typed; nothing is waiting on you
    elif action == "resume":
        # PostToolUse, any tool. Two distinct meanings:
        #   * main-agent tool (no agent_id): back to churning after a mid-turn
        #     pause — keep the clock. Catch-all so a plain permission grant
        #     clears "blocked" too.
        #   * subagent tool (agent_id set): does NOT mean the main agent is
        #     working — record the subagent as alive, and only clear "blocked"
        #     back to whatever the main agent was doing before (a bg agent's
        #     permission grant must return to idle/cyan, not blue running).
        if aid:
            agents.setdefault(aid, {"first": now})["last"] = now
            # Only the agent that *caused* the block may clear it. Clearing on
            # any subagent sighting wiped the main agent's own prompt: with a
            # background agent churning, its next tool call (seconds away)
            # turned "● waiting on you" back into "▶ busy" while an
            # AskUserQuestion sat unanswered on screen.
            if st.get("state") == "attention" and st.get("blocked_by") == aid:
                st["state"] = st.get("resume_to", "running")
                st.pop("blocked_by", None)
        elif st.get("state") == "attention" and st.get("blocked_by"):
            # main agent working on while a *subagent* waits on you: it's still
            # waiting, so keep the marker — only that subagent clears it.
            pass
        else:
            st["state"] = "running"
            st.setdefault("turn_start", now)
    elif action == "blocked":
        # paused waiting on you (permission / AskUserQuestion / ExitPlanMode).
        # Remember what to restore on resume: "idle" when the interruption came
        # from a background agent after the main turn ended, else "running".
        # `blocked_by` records *whose* prompt this is: a subagent's agent_id, or
        # None/absent for the main agent. resume/bgcount use it so only the
        # blocked party can clear the marker.
        if st.get("state") in ("running", "idle"):
            st["resume_to"] = st["state"]
        st["state"] = "attention"
        st["blocked_by"] = aid
        st.setdefault("turn_start", now)
    elif action == "bgcount":
        # SubagentStop — main phase unchanged. Drop the stopped agent, then
        # reconcile with the payload's background_tasks (authoritative for bg
        # agents; sync agents live only in the tracked dict).
        agents.pop(aid, None)
        # the agent that was waiting on you is gone (answered, or died) — its
        # marker would otherwise stick with nobody left to clear it
        if st.get("state") == "attention" and st.get("blocked_by") == aid:
            st["state"] = st.get("resume_to", "running")
            st.pop("blocked_by", None)
        if bg_ids is not None:
            st["bg"] = len(bg_ids)
            for i in bg_ids:
                agents.setdefault(i, {"first": now, "last": now})
    else:  # done — turn ended (Stop / StopFailure / idle_prompt): main is done.
        # keep turn_start so the timer keeps running while background agents
        # finish. Sync subagents cannot outlive the main turn, so when the
        # payload has a recount, it fully replaces the tracked set.
        # ...unless a *background* agent is still waiting on you: the main turn
        # being over doesn't answer its prompt, and "your move" (✓) would hide
        # the one thing that needs you.
        if st.get("state") == "attention" and st.get("blocked_by"):
            # ...but the main turn *is* over, so when that agent is finally
            # unblocked (or dies) it must restore to idle, not to the "running"
            # captured back when it blocked — else the window goes back to ▶
            # with nothing running.
            st["resume_to"] = "idle"
        else:
            st["state"] = "idle"
        st.setdefault("turn_start", now)
        if bg_ids is not None:
            st["bg"] = len(bg_ids)
            # keep first-seen for agents that survive the recount
            st["agents"] = {
                i: agents.get(i) or {"first": now, "last": now} for i in bg_ids
            }

    # mirror the (just-updated) turn state into the tmux window title first, so
    # the repo/wtitle cache it stamps onto `st` gets persisted below.
    set_tmux_title(st, cwd, plain=(action == "ended"))

    try:
        with open(path, "w") as f:
            json.dump(st, f)
    except Exception:
        pass
    return 0


 # --------------------------------------------------------------------------
 # Render mode helpers
 # --------------------------------------------------------------------------
 def git(cwd, *args):
    try:
        out = subprocess.run(
            ["git", "-C", cwd, *args],
            capture_output=True, text=True, timeout=1.0,
        )
        if out.returncode == 0:
            return out.stdout.strip()
    except Exception:
        pass
    return ""


 def fmt_elapsed(secs):
    secs = int(secs)
    m, s = divmod(secs, 60)
    if m >= 60:
        h, m = divmod(m, 60)
        return f"{h}:{m:02d}:{s:02d}"
    return f"{m}:{s:02d}"


 def truncate(s, width=24):
    """Middle-ellipsize an over-long string so it can't blow out the status line.

    Keeps head and tail (the informative ends of a branch name — prefix like
    `feat/` and the distinguishing suffix) and drops the middle for `…`.
    """
    if len(s) <= width:
        return s
    keep = width - 1  # room for the ellipsis
    head = (keep + 1) // 2
    tail = keep - head
    return s[:head] + "…" + s[-tail:]


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


 def vis_width(s):
    """Display cells a string occupies, ignoring SGR escapes.

    The branch segment embeds its own color changes for the dirty marker, so
    counting raw len() would overstate it wildly. Nerd Font glyphs sit in the
    Private Use Area, which unicodedata reports as ambiguous width; Ghostty
    draws them single-width, so only true wide/fullwidth chars count as 2.
    """
    n = 0
    for ch in ANSI_RE.sub("", s):
        if unicodedata.combining(ch):
            continue
        n += 2 if unicodedata.east_asian_width(ch) in ("W", "F") else 1
    return n


 def term_width(default=80):
    """Usable columns for one status row.

    Claude Code captures our stdout rather than handing us the tty, so ioctl
    and `tput cols` see nothing — it exports COLUMNS/LINES instead (v2.1.153+,
    verified set on 2.1.220). Keep one column of slack: a row that exactly
    fills the terminal triggers the terminal's own hard wrap, which drops a
    stray fragment on the next line and looks worse than wrapping ourselves.
    """
    try:
        return max(20, int(os.environ["COLUMNS"]) - 1)
    except (KeyError, ValueError, TypeError):
        return default


 def wrap_segments(segs, width):
    """Pack segments into rows that fit `width`, balancing the rows evenly.

    Each segment costs its text plus its two padding spaces and the powerline
    separator that follows it. Plain greedy packing fills row one to the brim
    and leaves whatever spills over stranded alone on row two, which reads as
    broken rather than wrapped. So: greedy once to learn the minimum number of
    rows the content needs, then binary-search the narrowest row width that
    still fits in that many rows, and pack to that.

    Segment order is never changed, which makes this a linear partition — the
    result is the most even split reachable without adding a row. A segment
    wider than the whole terminal still gets its own row rather than looping.
    """
    if not segs:
        return []
    costs = [vis_width(text) + 3 for text, _, _ in segs]

    def pack(limit):
        rows, row, used = [], [], 0
        for seg, cost in zip(segs, costs):
            if row and used + cost > limit:
                rows.append(row)
                row, used = [], 0
            row.append(seg)
            used += cost
        if row:
            rows.append(row)
        return rows

    need = len(pack(width))
    lo, hi = max(costs), max(width, max(costs))
    while lo < hi:
        mid = (lo + hi) // 2
        if len(pack(mid)) <= need:
            hi = mid
        else:
            lo = mid + 1
    return pack(lo)


 def render():
    try:
        data = json.load(sys.stdin)
    except Exception:
        data = {}

    cwd = data.get("cwd") or data.get("workspace", {}).get("current_dir") or os.getcwd()
    session_id = data.get("session_id", "")

    # ---- state file (turn indicator) -------------------------------------
    st = {}
    try:
        with open(state_path(session_id)) as f:
            st = json.load(f)
    except Exception:
        pass
    turn_state = st.get("state", "idle")

    # ---- subagents: files first, hook state as fallback ------------------
    # The session sidecar dir (agentproc, see notes/agentproc-2026-07-22.md)
    # is the reliable source for which subagents are in flight and when they
    # spawned — exact spawn mtimes, no ghost entries, no SubagentStop payload
    # quirks. The hook-tracked `agents` dict remains the fallback when the
    # statusline stdin has no transcript_path or the scan fails.
    file_agents = None
    if data.get("transcript_path"):
        try:
            import agentproc
            scanned = agentproc.scan(
                data["transcript_path"],
                cache_path=state_path(session_id)[:-len(".json")] + ".scan.json",
            )
            if scanned is not None:
                file_agents = agentproc.inflight(scanned)
        except Exception:
            file_agents = None

    # ---- git facts -------------------------------------------------------
    root = git(cwd, "rev-parse", "--show-toplevel")
    ws = data.get("workspace", {}) or {}
    repo_meta = ws.get("repo") or {}
    # repo_name() first: inside a linked worktree `--show-toplevel` is the
    # *worktree* dir, so its basename showed the agent's worktree
    # ("agent-ab6f265413c42acac") instead of the repo. The local name here used
    # to shadow the module-level repo_name() helper, which is why the
    # worktree-aware logic never reached the status bar.
    repo_label = repo_name(cwd) or repo_meta.get("name") \
        or os.path.basename(root or cwd or "")
    branch = ""
    dirty = False
    reldir = ""
    if root:
        branch = git(cwd, "symbolic-ref", "--short", "HEAD") or git(cwd, "rev-parse", "--short", "HEAD")
        dirty = bool(git(cwd, "status", "--porcelain"))
        rel = os.path.relpath(cwd, root)
        reldir = "" if rel == "." else rel

    worktree = data.get("worktree") or {}
    in_worktree = bool(worktree) or bool(ws.get("git_worktree"))
    if in_worktree and worktree.get("branch"):
        branch = worktree["branch"]

    # ---- context window --------------------------------------------------
    ctx = data.get("context_window") or {}
    ctx_pct = ctx.get("used_percentage")

    # ---- model / cost ----------------------------------------------------
    model = data.get("model", {}) or {}
    model_name = model.get("display_name", "?")
    # Claude Code spells extended context as "Opus 4.8 (1M context)" — compact it
    model_name = model_name.replace(" (1M context)", " [1m]").replace("(1M context)", "[1m]")
    if "[1m]" not in model_name and "[1m]" in (model.get("id", "") or ""):
        model_name += " [1m]"
    cost = (data.get("cost", {}) or {}).get("total_cost_usd") or 0.0

    # ---- assemble segments: (text, fg, bg) -------------------------------
    segs = []

    # 1. turn state + elapsed
    #    running/attention: main agent is active or blocked on you.
    #    idle but agents in flight: main is done, background agents still
    #      churning — keep the timer going (total task time) and show the count,
    #      so "done" only when everything (main + background) has finished.
    #    running with agents in flight: append the count (sync subagents show
    #      up here — they never outlive the main turn).
    if file_agents is not None:
        n_agents = len(file_agents)
    else:
        n_agents = max(st.get("bg", 0) or 0, len(st.get("agents") or {}))
    bg_running = turn_state == "idle" and n_agents > 0
    if turn_state == "running":
        icon, sbg, sfg = G_RUN, BLUE, CRUST
    elif turn_state == "attention":
        icon, sbg, sfg = G_ATTN, PEACH, CRUST
    elif bg_running:
        icon, sbg, sfg = G_BG, TEAL, CRUST
    else:
        icon, sbg, sfg = G_IDLE, GREEN, CRUST

    #    Timer base: normally turn_start. But once the main turn is over and
    #    only agents are in flight (cyan — or orange when a bg agent is blocked
    #    on a permission), turn_start is the *latest* prompt's clock, which any
    #    intervening chat turn resets — while a bg agent spawned earlier keeps
    #    running. Anchor on the oldest in-flight agent's first sighting instead
    #    (min with turn_start, so with no intervening prompt the timer still
    #    covers the whole task from the prompt that spawned the agents).
    timer_from = st.get("turn_start")
    if bg_running or (turn_state == "attention" and st.get("resume_to") == "idle"):
        if file_agents is not None:
            firsts = [a["spawn_ts"] for a in file_agents]
        else:
            firsts = [
                v.get("first") if isinstance(v, dict) else v
                for v in (st.get("agents") or {}).values()
            ]
            firsts = [f for f in firsts if isinstance(f, (int, float))]
        if firsts:
            timer_from = min([timer_from, *firsts]) if timer_from else min(firsts)

    show_timer = (turn_state in ("running", "attention") or bg_running) and timer_from
    if bg_running:
        label = f"{icon} {n_agents}"
        if show_timer:
            label += f"  {fmt_elapsed(time.time() - timer_from)}"
    elif show_timer:
        label = f"{icon} {fmt_elapsed(time.time() - timer_from)}"
        if turn_state == "running" and n_agents:
            label += f"  {G_BG} {n_agents}"
    else:
        label = icon
    segs.append((label, sfg, sbg))

    # 1b. hostname (which machine am I on — claudes-plan vs laptop)
    host = socket.gethostname().split(".")[0]
    segs.append((f"{G_HOST} {host}", MAUVE, MANTLE))

    # 2. repo
    segs.append((f"{G_REPO} {repo_label}", TEXT, SURFACE0))

    # 3. branch (+ dirty)
    if branch:
        btxt = f"{G_BRANCH} {truncate(branch)}"
        if dirty:
            btxt += f" {esc(fg=RED)}{DIRTY}{esc(fg=SUBTEXT0)}"
        wt = " " if in_worktree else ""  # code-fork glyph for worktree
        segs.append((btxt + wt, SUBTEXT0, MANTLE))

    # 4. dir (only when not at repo root)
    if reldir:
        segs.append((f"{G_DIR} {reldir}", SUBTEXT0, SURFACE0))

    # 5. context meter
    if ctx_pct is not None:
        if ctx_pct >= 70:
            cbg = RED
        elif ctx_pct >= 30:
            cbg = YELLOW
        else:
            cbg = GREEN
        segs.append((f"{G_CTX} {ctx_pct:.0f}%", CRUST, cbg))
    else:
        segs.append((f"{G_CTX} --", OVERLAY0, SURFACE0))

    # 6. model
    segs.append((f"{G_MODEL} {model_name}", TEAL, MANTLE))

    # 7. cost (only if nonzero)
    if cost and cost > 0:
        # SUBTEXT0, not OVERLAY0: muted-on-SURFACE0 is only 2.57:1, which the
        # dollar figure was unreadable at. This is 5.65:1, still a step below
        # the repo name (TEXT, 8.69:1) so the hierarchy holds.
        segs.append((f"{G_COST} {cost:.2f}", SUBTEXT0, SURFACE0))

    # ---- powerline render ------------------------------------------------
    # Claude Code renders each line of our stdout as its own status row, so an
    # over-long bar wraps into a second row instead of being truncated.
    out = []
    for row in wrap_segments(segs, term_width()):
        for i, (text, fg, bg) in enumerate(row):
            out.append(esc(fg, bg) + f" {text} ")
            nxt_bg = row[i + 1][2] if i + 1 < len(row) else None
            if nxt_bg is not None:
                out.append(esc(fg=bg, bg=nxt_bg) + SEP)
            else:
                out.append(RESET + esc(fg=bg) + SEP + RESET)
        out.append("\n")
    sys.stdout.write("".join(out).rstrip("\n"))


 def main():
    if len(sys.argv) >= 3 and sys.argv[1] == "hook":
        return run_hook(sys.argv[2])
    render()
    return 0


 if __name__ == "__main__":
    sys.exit(main())
Segment	Content	Notes
Turn state	churning + `M:SS` (+ `N` if subagents in flight) / blocked on you / ` N` bg agents + `M:SS` / your move	blue / peach / teal / green. Timer runs until all background agents finish too. Sync subagents show as a count inside the blue churning segment. The count is agents working, so an idle-available teammate is excluded.
Host	short hostname	which machine, mauve
Repo	repo name	the main working tree's name (via `--git-common-dir`), so a linked worktree shows the repo rather than the worktree directory
Branch	branch, `✱` if dirty, `` if in a worktree	dirty asterisk is red
Dir	path relative to repo root	hidden when at the repo root
Context	block-meter + `NN%`	green < 30% · yellow 30–70% · red ≥ 70% (tuned to run lean)
Model	display name; `(1M context)` compacted to `[1m]`
Cost	session `$`	hidden when `$0.00`
State	Icon / color	Set by (hook → action)
churning	blue	`UserPromptSubmit`→`start` (resets clock); `PostToolUse` (any tool)→`resume` (keeps clock; back to work after you answer a question or grant a permission)
blocked on you	orange	`PermissionRequest`→`blocked`; `PreToolUse`[`AskUserQuestion`/`ExitPlanMode`]→`blocked`; `Notification`[`permission_prompt`]→`blocked`
background agents	`N` cyan	derived: main is `done` but `N` background agents still in flight — keeps the timer running
your move	green	`Stop`→`done`; `StopFailure`→`done` (turn ended on API error); `Notification`[`idle_prompt`]→`done`
Window title	Meaning
`● repo`	blocked on you — permission / question / plan approval (orange bell state)
`✓ repo`	turn done, your move (green idle, no background agents)
`▶ repo`	main turn churning — don't interrupt
`⚙ repo`	main turn done, background agents still running
`repo`	no live agent (session ended)
	"""File-based introspection of Claude Code subagents — "/proc for agents".

	Single parser for the per-session sidecar directory Claude Code writes next to
	each transcript (format contract + design decisions: notes/agentproc-2026-07-22.md
	in this repo; verified empirically on Claude Code 2.1.219):

	<config>/projects/<slug>/<session-id>.jsonl # parent transcript
	<config>/projects/<slug>/<session-id>/subagents/
	agent-<id>.meta.json # identity, written once at spawn (mtime = spawn)
	agent-<id>.jsonl # live-appended transcript (mtime = last activity)
	workflows/<run-id>/ # agents spawned by the Workflow tool, same pair
	journal.jsonl # append-only started/result log for the run

	Completion has three shapes. For Agent-tool subagents it is only visible in the
	parent transcript: a `<task-notification>` whose task-id equals the agent id
	(background agents), or a real tool_result for the meta's toolUseId (sync agents
	— distinguished from the background launch ack, whose content starts with "Async
	agent launched successfully"). Workflow agents have neither field; their
	completion is a `{"type":"result","agentId":...}` line in their run's
	journal.jsonl, backstopped by the run's own task-notification. The agent's own
	transcript has no terminal record in any case.

	Consumers: statusline/statusline.py (same dir, `import agentproc`) and
	bin/agent-ps. Stdlib only.
	"""

	import datetime
	import json
	import os
	import re

	ASYNC_ACK = "Async agent launched successfully"
	STOPPED_MSG = "Successfully stopped task:"

	# The scan cache stores per-file byte offsets, so a changed parser would
	# otherwise never re-read bytes an older one already consumed — leaving stale
	# state pinned forever. The cache is stamped with this value and reset on any
	# mismatch, forcing one full re-scan.
	#
	# It is derived from this module's own mtime/size rather than hand-bumped: a
	# hand-bumped constant has to be edited in the same write as the parser
	# change, or a live 1 Hz consumer (the statusline) runs the in-between state and
	# persists a cache stamped new but produced by the old parser. That is not
	# theoretical — it happened on 2026-07-24 and pinned two phantom agents that
	# survived the very fix written to clear them. Editing this file now invalidates
	# every cache automatically; the cost is one full re-scan.
	_SCHEMA = 3 # bump only to force invalidation without touching the file


	def _cache_version():
	try:
	st = os.stat(__file__)
	return f"{_SCHEMA}:{int(st.st_mtime)}:{st.st_size}"
	except OSError:
	return f"{_SCHEMA}:?"


	CACHE_VERSION = _cache_version()

	# Statuses
	RUNNING = "running"
	DONE = "done"
	IDLE = "idle" # in-process teammates only — see _teammate_idle_in_line

	# A teammate's idle notification lands at the same instant its transcript stops
	# growing (observed equal to the second), but the two clocks are different
	# sources, so allow a little slack before calling an append "newer".
	IDLE_EPS = 2.0


	def sidecar_dir(transcript_path):
	"""Session sidecar directory for a transcript path, or None."""
	if not transcript_path or not transcript_path.endswith(".jsonl"):
	return None
	return transcript_path[: -len(".jsonl")]


	def _read_metas(subdir):
	"""{agent_id: meta dict + _spawn_ts/_meta_path/_dir} for every agent ever
	spawned in this session (meta files are written once, at spawn).

	Walks the whole tree: Agent-tool subagents sit directly in `subagents/`,
	Workflow agents in `subagents/workflows/<run-id>/`. `_dir` records which,
	since it locates the agent's transcript and its run journal."""
	metas = {}
	for root, _dirs, names in os.walk(subdir):
	for name in names:
	if not (name.startswith("agent-") and name.endswith(".meta.json")):
	continue
	aid = name[len("agent-"): -len(".meta.json")]
	path = os.path.join(root, name)
	try:
	with open(path) as f:
	meta = json.load(f)
	meta["_spawn_ts"] = os.path.getmtime(path)
	except Exception:
	continue
	meta["_meta_path"] = path
	meta["_dir"] = root
	metas[aid] = meta
	return metas


	def _load_cache(cache_path):
	"""Prior scan state {offsets: {path: byte_offset}, done: [...], runs: {...}}.

	Per-path offsets because completions are spread across several files (the
	session root, each parent subagent's own transcript, and each workflow run's
	journal). `runs` maps a workflow run's transcript dir to the background task
	id that reports the whole run finished. A rotated/truncated individual file
	is handled in the scan loop, not here. Any unreadable or old-schema cache
	resets to empty — the cache is a pure optimization, so a reset just forces a
	full re-scan."""
	empty = {"offsets": {}, "done": [], "runs": {}, "idles": {}}
	if not cache_path:
	return empty
	try:
	with open(cache_path) as f:
	cache = json.load(f)
	got = {k: cache.get(k) for k in ("offsets", "done", "runs", "idles")}
	if not (isinstance(got["offsets"], dict) and isinstance(got["done"], list)
	and isinstance(got["runs"], dict) and isinstance(got["idles"], dict)
	and cache.get("v") == CACHE_VERSION):
	raise ValueError
	return got
	except Exception:
	return empty


	def _mark_done_from_transcript_line(line, tid_to_aid, done, runs, idles):
	"""Record what this parent-transcript line reports finished.

	Two completion shapes, each behind a cheap substring gate before any JSON
	parse:
	* background — a `<task-notification>` carrying `<task-id>`; the id is an
	agent id for an Agent-tool spawn and a task id for a whole Workflow run,
	so both go into `done` and the caller decides which it matched;
	* sync — a real tool_result for a spawning toolUseId (not the launch ack,
	which `_tool_results_in_line` filters out).

	Also harvests the Workflow launch record's transcriptDir -> taskId mapping
	into `runs`, which is what lets a finished run sweep its own agents done."""
	if "<task-id>" in line:
	i = line.find("<task-id>")
	j = line.find("</task-id>", i)
	if j > i:
	tid = line[i + len("<task-id>"): j].replace("\\n", "").strip()
	if tid:
	done.add(tid)
	if "tool_use_id" in line and any(t in line for t in tid_to_aid):
	done.update(_tool_results_in_line(line, tid_to_aid))
	if '"transcriptDir"' in line:
	runs.update(_workflow_runs_in_line(line))
	if '"task_id"' in line:
	done.update(_stopped_tasks_in_line(line))
	if "idle_notification" in line:
	for who, ts in _teammate_idle_in_line(line).items():
	if ts > idles.get(who, 0):
	idles[who] = ts


	def _mark_done_from_journal_line(line, done):
	"""Add the agent id this workflow-journal line reports as complete.

	The run's journal is append-only with `started` then `result` records per
	agent; only `result` means finished. An agent whose result was null (died on
	a terminal error, or the user skipped it) never gets a `result` line at all —
	the run-level sweep in scan() is what eventually clears those."""
	if '"result"' not in line:
	return
	try:
	rec = json.loads(line)
	except Exception:
	return
	if rec.get("type") == "result" and rec.get("agentId"):
	done.add(rec["agentId"])


	_IDLE_RE = re.compile(r'\{[^{}]"type"\s:\s"idle_notification"[^{}]\}')


	def _record_texts(rec):
	"""Every free-text body in a transcript record (str content, text blocks,
	and string tool_result contents)."""
	msg = rec.get("message")
	content = msg.get("content") if isinstance(msg, dict) else None
	if isinstance(content, str):
	yield content
	elif isinstance(content, list):
	for block in content:
	if not isinstance(block, dict):
	continue
	for key in ("text", "content"):
	if isinstance(block.get(key), str):
	yield block[key]
	if isinstance(rec.get("content"), str):
	yield rec["content"]


	def _teammate_idle_in_line(line):
	"""{teammate name: idle timestamp} announced by this line.

	In-process teammates (`taskKind: in_process_teammate`, spawned into a team)
	are peers, not tasks: they finish a piece of work, report it, and sit
	idle-available waiting to be messaged again, then wake and work more. They
	therefore never produce a completion marker, and treating them as RUNNING
	forever overcounts every teammate that has ever been spawned.

	What they do emit, into the spawning session's transcript, is

	<teammate-message teammate_id="pr21-review" color="blue">
	{"type":"idle_notification","from":"pr21-review","timestamp":"...Z",
	"idleReason":"available"}
	</teammate-message>

	keyed by teammate name (not agent id), with the JSON embedded in a text
	body — so it needs a decoded-string search, not a raw-line one. Comparing
	the latest such timestamp against the agent transcript's mtime gives idle vs
	working precisely, rather than by a staleness threshold: a teammate that has
	appended since its last idle notification is working, however long ago that
	was."""
	found = {}
	try:
	rec = json.loads(line)
	except Exception:
	return found
	for text in _record_texts(rec):
	if "idle_notification" not in text:
	continue
	for match in _IDLE_RE.finditer(text):
	try:
	note = json.loads(match.group(0))
	who = note["from"]
	stamp = note.get("timestamp") or rec.get("timestamp")
	ts = datetime.datetime.fromisoformat(
	str(stamp).replace("Z", "+00:00")).timestamp()
	except Exception:
	continue
	if not found.get(who) or ts > found[who]:
	found[who] = ts
	return found


	def _first_record_ts(path, max_lines=25):
	"""Timestamp of the first timestamped record in a transcript, or None.

	Needed because a teammate's `meta.json` is **rewritten when it is woken by a
	message**, so its mtime is the last wake, not the spawn — measured drift of
	+829s on a teammate re-messaged 14 minutes after spawning, against exactly 0s
	for every ordinary subagent. The transcript's first record is the durable
	spawn time. Reads only the head of the file."""
	try:
	with open(path, errors="replace") as f:
	for _, line in zip(range(max_lines), f):
	if '"timestamp"' not in line:
	continue
	try:
	stamp = json.loads(line).get("timestamp")
	return datetime.datetime.fromisoformat(
	str(stamp).replace("Z", "+00:00")).timestamp()
	except Exception:
	continue
	except OSError:
	pass
	return None


	def _stopped_tasks_in_line(line):
	"""Task ids killed by a `TaskStop` in this line.

	A killed task emits no `<task-notification>` at all (verified 2026-07-24
	on a TaskStop'd workflow run) — this tool_result is the only marker it
	leaves, and its `task_id` is snake_case, unlike the `taskId` in a launch
	record. Covers both a whole Workflow run (`task_type: local_workflow`, which
	then sweeps its agents) and a background Agent spawn (task_id == agent id).
	Matched narrowly on TaskStop's success message, since TaskList/TaskGet
	output also carries `task_id` fields — for tasks that are still running."""
	try:
	rec = json.loads(line)
	except Exception:
	return set()
	res = rec.get("toolUseResult")
	if not isinstance(res, dict):
	return set()
	tid, msg = res.get("task_id"), res.get("message")
	if tid and isinstance(msg, str) and msg.startswith(STOPPED_MSG):
	return {tid}
	return set()


	def _workflow_runs_in_line(line):
	"""{run transcript dir: background task id} from a Workflow launch record.

	The Workflow tool returns immediately; its toolUseResult carries the run's
	sidecar dir alongside the task id that a later `<task-notification>` uses to
	announce the whole run finished. Keyed by dir rather than runId so the match
	is exact, with no basename assumption."""
	try:
	rec = json.loads(line)
	except Exception:
	return {}
	res = rec.get("toolUseResult")
	if not isinstance(res, dict):
	return {}
	tdir, task = res.get("transcriptDir"), res.get("taskId")
	return {tdir: task} if tdir and task else {}


	def _tool_results_in_line(line, tid_to_aid):
	"""Agent ids completed by tool_result blocks in this transcript line.

	A tool_result for a spawning toolUseId means the agent finished — except
	the background-launch ack, whose content carries ASYNC_ACK."""
	finished = set()
	try:
	rec = json.loads(line)
	except Exception:
	return finished
	msg = rec.get("message")
	content = msg.get("content") if isinstance(msg, dict) else None
	if not isinstance(content, list):
	return finished
	for block in content:
	if not (isinstance(block, dict) and block.get("type") == "tool_result"):
	continue
	aid = tid_to_aid.get(block.get("tool_use_id"))
	if not aid:
	continue
	if ASYNC_ACK in json.dumps(block.get("content", "")):
	continue # launch ack, not completion
	finished.add(aid)
	return finished


	def _scan(transcripts, journals, metas, cache_path=None):
	"""(done ids, {run dir: task id}) from every completion source.

	An Agent-tool subagent's completion is recorded in the transcript of
	whichever agent spawned it: the session root for depth-1 agents, and a
	parent subagent's own transcript (`agent-<parentAgentId>.jsonl`) for
	anything deeper. Workflow agents are recorded in their run's journal
	instead. Scanning only the root therefore leaves nested and workflow agents
	pinned RUNNING forever even after they finish — the caller passes the root
	plus every parent transcript plus every run journal. Incremental per file:
	only bytes appended past each file's cached offset are read; `done` and
	`runs` accumulate across calls and files."""
	tid_to_aid = {
	m["toolUseId"]: aid for aid, m in metas.items() if m.get("toolUseId")
	}
	cache = _load_cache(cache_path)
	done = set(cache["done"])
	runs = dict(cache["runs"])
	idles = dict(cache["idles"])
	offsets = dict(cache["offsets"])

	sources = [(transcripts,
	lambda ln: _mark_done_from_transcript_line(
	ln, tid_to_aid, done, runs, idles)),
	(journals,
	lambda ln: _mark_done_from_journal_line(ln, done))]
	for paths, handle in sources:
	for path in paths:
	try:
	size = os.path.getsize(path)
	except OSError:
	continue
	off = offsets.get(path, 0)
	if off > size: # file rotated/truncated — re-read from the top
	off = 0
	if off >= size:
	continue
	try:
	with open(path, errors="replace") as f:
	f.seek(off)
	# Offsets we write are always line boundaries (we read to EOF
	# in whole lines) and keyed by path, so a seek never lands
	# mid-line.
	for line in f:
	handle(line)
	offsets[path] = f.tell()
	except OSError:
	continue

	if cache_path:
	try:
	with open(cache_path, "w") as f:
	json.dump({"v": CACHE_VERSION, "offsets": offsets,
	"done": sorted(done), "runs": runs,
	"idles": idles}, f)
	except OSError:
	pass
	return done, runs, idles


	def scan(transcript_path, cache_path=None):
	"""List all subagents of a session, each a dict:

	id, status (RUNNING/DONE), spawn_ts, last_ts,
	agentType, description, model, spawnDepth, toolUseId, worktreePath,
	workflow (run id, or None for an Agent-tool spawn)

	Returns None when the transcript has no sidecar (can't tell — caller
	should fall back), and [] when the sidecar exists but has no agents.
	Never raises. `cache_path` enables the incremental scan (recommended for
	1 Hz callers like the statusline renderer)."""
	side = sidecar_dir(transcript_path)
	if not side or not os.path.isdir(side):
	return None
	subdir = os.path.join(side, "subagents")
	if not os.path.isdir(subdir):
	return []
	metas = _read_metas(subdir)
	if not metas:
	return []
	# An agent's transcript sits next to its meta, wherever that is.
	transcript_of = {
	aid: os.path.join(m["_dir"], f"agent-{aid}.jsonl") for aid, m in metas.items()
	}
	# Completions live in the transcript of whichever agent spawned each
	# subagent: the session root for depth-1, a parent subagent's own transcript
	# for deeper nesting. Scan the root plus every parent transcript present in
	# the sidecar. This covers arbitrary depth — a deep agent's parent is itself
	# a subagent whose transcript sits alongside its own meta.
	parent_ids = {
	m.get("parentAgentId") for m in metas.values() if m.get("parentAgentId")
	}
	parent_transcripts = [
	p for p in (transcript_of.get(pid) for pid in sorted(parent_ids))
	if p and os.path.exists(p)
	]
	# Workflow runs report per-agent completion in their own journal.
	run_dirs = sorted({
	m["_dir"] for m in metas.values()
	if os.path.exists(os.path.join(m["_dir"], "journal.jsonl"))
	})
	journals = [os.path.join(d, "journal.jsonl") for d in run_dirs]
	done, runs, idles = _scan(
	[transcript_path, *parent_transcripts], journals, metas, cache_path)

	# Run-level sweep: once the whole Workflow run is announced finished, none
	# of its agents can still be alive. Needed because an agent whose result was
	# null never gets a journal `result` line, and because a killed run leaves
	# its in-flight agents with no marker at all.
	for d in run_dirs:
	task = runs.get(d)
	if task and task in done:
	done.update(aid for aid, m in metas.items() if m["_dir"] == d)

	agents = []
	for aid, meta in metas.items():
	jsonl = transcript_of[aid]
	try:
	last_ts = os.path.getmtime(jsonl)
	except OSError:
	last_ts = meta["_spawn_ts"]
	run_id = (os.path.basename(meta["_dir"])
	if os.path.join(meta["_dir"], "journal.jsonl") in journals else None)
	# A teammate has no completion marker; it reports going idle-available
	# instead, and can be woken again later. Idle iff it has not appended to
	# its transcript since its latest idle notification.
	teammate = meta.get("taskKind") == "in_process_teammate"
	idle_ts = idles.get(meta.get("name")) if teammate else None
	spawn_ts = meta["_spawn_ts"]
	if teammate: # meta mtime is the last wake, not the spawn
	spawn_ts = _first_record_ts(jsonl) or spawn_ts
	if aid in done:
	status = DONE
	elif idle_ts is not None and last_ts <= idle_ts + IDLE_EPS:
	status = IDLE
	else:
	status = RUNNING
	agents.append({
	"id": aid,
	"status": status,
	"spawn_ts": spawn_ts,
	"last_ts": last_ts,
	"agentType": meta.get("agentType"),
	"description": meta.get("description"),
	"model": meta.get("model"),
	"spawnDepth": meta.get("spawnDepth"),
	"toolUseId": meta.get("toolUseId"),
	"worktreePath": meta.get("worktreePath"),
	"workflow": run_id,
	"name": meta.get("name"),
	"taskKind": meta.get("taskKind"),
	"transcript": jsonl,
	})
	# Sorted on the corrected spawn time, so a woken teammate keeps its place.
	agents.sort(key=lambda a: a["spawn_ts"])
	return agents


	def inflight(agents):
	"""The RUNNING subset of a scan() result (empty for None).

	Excludes IDLE teammates: they are alive but not working, so counting them
	as in flight overstates what is churning (see _teammate_idle_in_line)."""
	return [a for a in (agents or []) if a["status"] == RUNNING]


	def unfinished(agents):
	"""RUNNING plus IDLE — everything still alive, working or not."""
	return [a for a in (agents or []) if a["status"] in (RUNNING, IDLE)]


	def last_usage(agent_transcript, tail_bytes=65536):
	"""Best-effort (model, usage-dict) from the last assistant record of an
	agent transcript, reading only the file tail. (None, None) on failure."""
	try:
	size = os.path.getsize(agent_transcript)
	with open(agent_transcript, errors="replace") as f:
	f.seek(max(0, size - tail_bytes))
	lines = f.read().splitlines()
	except OSError:
	return None, None
	for line in reversed(lines):
	if '"usage"' not in line:
	continue
	try:
	rec = json.loads(line)
	except Exception:
	continue
	msg = rec.get("message")
	if isinstance(msg, dict) and msg.get("usage"):
	return msg.get("model"), msg["usage"]
	return None, None