claude-code-gnome-extension/hooks/claude-status-hook.py

#!/usr/bin/env python3
"""Claude Code hook -> per-session state file for the GNOME panel indicator.

Reads one hook event as JSON on stdin and maps it to a session state, written
to $XDG_STATE_HOME/claude-code-status/<session_id>.json. The GNOME extension
watches that directory; no polling, no daemon, no socket.

Design notes that are easy to get wrong:

* Writes are skipped when the state does not change. PostToolUse fires on every
  tool call, and its only job here is to clear "blocked" once a permission has
  been granted -- letting it rewrite the file each time would make the directory
  monitor fire hundreds of times per turn for no new information.

* Hooks are registered async, so two events can race (the last PostToolUse of a
  turn against that turn's Stop). The whole read-decide-write runs under a file
  lock and an event older than the stored one is refused; without both, a late
  "busy" buries "waiting" and the panel claims a session is working while it
  actually waits for input.

* No stdlib import beyond what is needed: this runs once per tool call.
"""

import errno
import fcntl
import json
import os
import sys
import time

STATE_DIR = os.path.join(
    os.environ.get("XDG_STATE_HOME") or os.path.expanduser("~/.local/state"),
    "claude-code-status",
)

# Notification covers several unrelated things; only some mean "the session
# stopped and is asking me something". auth_success / elicitation_complete /
# elicitation_response are progress chatter and must not touch the state.
# Observed: a question put to the user (AskUserQuestion) arrives as
# "permission_prompt" too, with the same generic message as a tool asking to
# run. The two are therefore not separable here, which is why the panel has one
# "blocked" state rather than telling a permission from a question.
NOTIFICATION_STATES = {
    "permission_prompt": "blocked",
    "agent_needs_input": "blocked",
    "elicitation_dialog": "blocked",
    "idle_prompt": "waiting",
    "agent_completed": "waiting",
}

# Background tasks that mean the session is still working. A background shell
# is not one: a dev server left running says nothing about whether the session
# needs you, and counting it would pin the chip at "working" for as long as it
# lives.
BUSY_TASK_TYPES = {"subagent"}

# Flags that mean "run one prompt and exit" rather than "open a session". See
# is_headless: such a run has no terminal to be called over to.
HEADLESS_FLAGS = {"-p", "--print"}

# Events that mean the question has been dealt with. Anything else leaves a
# "blocked" session blocked: a subagent finishing, or the next tool starting,
# says nothing about the prompt still sitting on your screen, and clearing it
# would hide the one state this indicator exists to surface.
BLOCK_CLEARING = {"PostToolUse", "UserPromptSubmit", "Stop", "SessionStart"}

# Nothing here is displayed at full length, and both ends of the pipe have to
# survive a hostile or merely absurd value: a multi-megabyte cwd would be copied
# into the state file, read back by the compositor and handed to Pango.
MAX_TEXT = 512
# A stored timestamp further ahead than this is not a concurrent write, it is
# corruption -- and left alone it would refuse every later event forever,
# freezing the session's displayed state for good.
FUTURE_SLACK = 60  # seconds


def number(value, default=0):
    """Coerce a value read back from disk. Files are ordinary user-writable
    JSON: one corrupt field must not take the hook down for every session."""
    try:
        n = float(value)
    except (TypeError, ValueError):
        return default
    return default if n != n or n in (float("inf"), float("-inf")) else n


def clip(value):
    text = value if isinstance(value, str) else ""
    return text[:MAX_TEXT]

EVENT_STATES = {
    # A session that has just opened is waiting for your first prompt, which is
    # the same thing as one that has finished a turn: the input line is free
    # and the next move is yours. It had a state of its own once; it was only
    # ever reachable before the first prompt, so it bought a fourth glyph in
    # the panel that nobody saw.
    "SessionStart": "waiting",
    "PreToolUse": "busy",
    "UserPromptSubmit": "busy",
    "PreCompact": "busy",
    "PostToolUse": "busy",
    "Stop": "waiting",
}


def debug_log(event):
    """Append raw events when a 'debug' marker file exists in the state dir.

    Gated on a file rather than an env var because the hook inherits claude's
    environment, which cannot be changed without restarting the session.
    """
    marker = os.path.join(STATE_DIR, "debug")
    try:
        # Not followed through a symlink, and not grown without limit: this
        # records prompts verbatim and the command lines of ancestor processes.
        if os.lstat(marker).st_size > 8 << 20:
            return
    except OSError:
        return
    try:
        fd = os.open(marker, os.O_WRONLY | os.O_APPEND | os.O_NOFOLLOW)
        with os.fdopen(fd, "a") as fh:
            chain, pid = [], os.getppid()
            for _ in range(6):
                if pid <= 1:
                    break
                chain.append("%d %s" % (pid, read_cmdline(pid)[:90]))
                pid = parent_of(pid)
            stamped = dict(event, _at=time.strftime("%H:%M:%S"), _ancestry=chain)
            fh.write(json.dumps(stamped, sort_keys=True)[:2000] + "\n")
    except OSError:
        pass


def derive_state(event):
    """Return the new state, 'end' to drop the session, or None to ignore."""
    name = event.get("hook_event_name")
    if name == "SessionEnd":
        return "end"
    # A notification is honoured whoever raised it: it means a human is needed,
    # and that is just as true when the agent that got stuck is a subagent.
    if name == "Notification":
        return NOTIFICATION_STATES.get(event.get("notification_type"))
    # SubagentStop carries agent_id itself, so it has to pass the filter below.
    # It is the one event that can free a session whose main agent stopped a
    # long time ago; which way it goes needs the snapshot and is decided in
    # apply_event.
    if name == "SubagentStop":
        return "busy"
    # A subagent's own tool calls also reach the parent session's hooks
    # (measured: PostToolUse carrying agent_id and agent_type). They are
    # ignored: the session is working either way, and a single subagent's shell
    # commands alone would be hundreds of writes.
    if event.get("agent_id"):
        return None
    return EVENT_STATES.get(name)


def running_agents(event):
    """How many subagents the session still has running, or None if this event
    does not say.

    Read from the event's own ``background_tasks`` rather than counted from
    starts and stops. Counting was what this did first, and it was wrong in the
    direction that matters: ``PreToolUse`` only ever sees a top-level launch --
    a subagent spawning its own subagents does it through hooks carrying
    ``agent_id``, which are dropped above -- while ``SubagentStop`` arrives for
    every subagent at every depth. Each nested one therefore subtracted from a
    batch it had never joined, and the session was declared free with its work
    still running. Measured live: one background agent, fourteen nested stops
    behind it, and a session shown as waiting from the first of them on.

    The field is present exactly where it decides something: on ``Stop`` and
    ``SubagentStop``, the two events that can end a turn. Everywhere else it is
    absent, and the stored value stands.
    """
    tasks = event.get("background_tasks")
    if not isinstance(tasks, list):
        return None
    running = 0
    for task in tasks:
        if not isinstance(task, dict):
            continue
        if task.get("type") in BUSY_TASK_TYPES and task.get("status") == "running":
            running += 1
    return min(running, 999)


def read_environ(pid):
    """Environment of a process as a dict, empty if it is gone or not ours."""
    try:
        with open("/proc/%d/environ" % pid, "rb") as fh:
            raw = fh.read()
    except OSError:
        return {}
    env = {}
    for entry in raw.split(b"\0"):
        if not entry:
            continue
        key, sep, value = entry.partition(b"=")
        if sep:
            env[key.decode("utf-8", "replace")] = value.decode("utf-8", "replace")
    return env


def read_argv(pid):
    """Command line of a process as a list of arguments, empty if it is gone.

    Split on the NUL separators the kernel actually puts there, not on spaces:
    an argument may contain spaces of its own, and `claude "when to use -p"`
    must not read as an argument list containing a bare "-p".
    """
    try:
        with open("/proc/%d/cmdline" % pid, "rb") as fh:
            raw = fh.read()
    except OSError:
        return []
    return [arg.decode("utf-8", "replace") for arg in raw.split(b"\0") if arg]


def read_cmdline(pid):
    return " ".join(read_argv(pid))


def parent_of(pid):
    try:
        with open("/proc/%d/status" % pid, "r") as fh:
            for line in fh:
                if line.startswith("PPid:"):
                    return int(line.split()[1])
    except (OSError, ValueError):
        pass
    return 0


def looks_like_claude(argv):
    """Is this argument list the claude binary itself?

    Matched per argument, never against the joined string. The hook is spawned
    as `/bin/sh -c /.../claude-status-hook.py`, so its parent's command line
    contains the word "claude" -- in a path -- without being claude at all.
    Latching onto that shell records a pid that exits milliseconds later, and
    the session then flickers in and out of the panel.
    """
    for token in argv:
        base = os.path.basename(token)
        if base == "claude":
            return True
        # npm-style install: node /path/to/claude-code/cli.js
        if base == "cli.js" and "claude" in token:
            return True
    return False


def is_headless(argv):
    """Was this claude started to print one answer and exit?

    `claude -p` has no input line and nobody sitting in front of it, so it can
    never be blocked on you and there is nothing to walk over to. Left in, a
    script that runs a few dozen of them turns the panel into a flicker of chips
    that are gone before they can be read -- and the same goes for the Agent SDK
    and editor integrations, which drive claude the same way.

    Only exact tokens count. A prompt is an ordinary argument, and `claude "what
    does -p do"` is an interactive session that must keep its chip.
    """
    return any(token in HEADLESS_FLAGS for token in argv)


def find_claude():
    """Nearest ancestor that is the claude process itself, with its arguments.

    Returns (pid, argv), or (0, []) if it cannot be identified. The hook is
    spawned through a shell, so the immediate parent is not claude. Walking
    beyond a handful of levels risks latching onto an outer claude when one
    session drives another, so the search stops early.

    Returning 0 rather than guessing matters: the reader deletes state files
    whose process is gone, and the obvious fallback -- the shell that spawned
    this hook -- exits immediately.
    """
    pid = os.getppid()
    for _ in range(6):
        if pid <= 1:
            break
        argv = read_argv(pid)
        if looks_like_claude(argv):
            return pid, argv
        pid = parent_of(pid)
    return 0, []


def pid_start_time(pid):
    """Field 22 of /proc/<pid>/stat: when the process started, in clock ticks.

    Pins a pid to one particular process. Pids are reused, and state files
    outlive reboots -- without this, a file left by a crashed session whose pid
    is later handed to something unrelated reads as a live session forever.
    """
    try:
        with open("/proc/%d/stat" % pid) as fh:
            data = fh.read()
    except OSError:
        return 0
    # Field 2 is the command name, parenthesised, and may itself contain spaces
    # and a ')'. Everything after the last ')' is field 3 onwards.
    tail = data[data.rfind(")") + 2:].split()
    try:
        return int(tail[19])
    except (IndexError, ValueError):
        return 0


def alive(pid, start=0):
    """Is that pid still the process it was? Inputs come from disk, so both
    arguments are coerced rather than trusted."""
    pid = int(number(pid))
    if pid <= 0 or not os.path.exists("/proc/%d" % pid):
        return False
    # A file written before start times were recorded has nothing to compare.
    start = number(start)
    return not start or pid_start_time(pid) == start


def sweep_dead(keep):
    """Drop state files whose claude process is gone.

    A killed terminal never sends SessionEnd, so files leak. Cleaning up on
    SessionStart keeps the sweep off the hot path -- the extension only has to
    hide stale entries, not own their lifetime.
    """
    try:
        names = os.listdir(STATE_DIR)
    except OSError:
        return
    for name in names:
        if not name.endswith(".json") or name == keep:
            continue
        path = os.path.join(STATE_DIR, name)
        try:
            with open(path, "r") as fh:
                stale = json.load(fh)
            if not isinstance(stale, dict):
                continue
            pid = stale.get("pid", 0)
        except (OSError, ValueError, AttributeError):
            continue
        except Exception:
            # One unreadable file must not abort the sweep, and above all must
            # not abort the caller: this runs before the hook writes its own
            # state, so an exception here would stop new sessions appearing at
            # all, for as long as the bad file sits there.
            continue
        # pid 0 means the hook could not identify the process; there is nothing
        # to test for liveness, so leave it to the reader's age cutoff.
        if pid and not alive(pid, stale.get("pid_start", 0)):
            for victim in (path, path + ".lock"):
                try:
                    os.unlink(victim)
                except OSError:
                    pass


def write_atomic(path, payload):
    tmp = "%s.%d.tmp" % (path, os.getpid())
    with open(tmp, "w") as fh:
        json.dump(payload, fh)
    os.replace(tmp, path)


def open_lock(path):
    """Open the lock file without following a symlink.

    A plain open() on a symlinked lock path truncates whatever it points at.
    Nothing is escalated by that on a single-user machine, but a status
    indicator has no business truncating files it was pointed at.
    """
    flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW
    try:
        return os.fdopen(os.open(path, flags, 0o600), "r+")
    except OSError as exc:
        if exc.errno in (errno.ELOOP, errno.EMLINK):
            return None
        raise


def apply_event(event, state, path, now, claude_pid):
    """Read the current state, decide, and write. Must run under the lock."""
    if event.get("hook_event_name") == "SessionStart":
        # Swept before any early return: a resumed session keeps its id, so its
        # SessionStart finds an unchanged state and would otherwise bail out
        # before ever reaching the sweep.
        sweep_dead(keep=os.path.basename(path))

    if state == "end":
        # Only the state file. Unlinking the lock while holding it drops mutual
        # exclusion -- a hook already blocked on the old inode and one that
        # creates a new file are then both inside the critical section.
        try:
            os.unlink(path)
        except OSError:
            pass
        return

    previous = None
    try:
        with open(path, "r") as fh:
            previous = json.load(fh)
    except (OSError, ValueError):
        previous = None
    if not isinstance(previous, dict):
        previous = None

    # Normalised here so the comparison below is against what would actually be
    # stored: comparing a stored "" to a raw notification message rewrites the
    # file on every idle_prompt for no change at all.
    message = clip(event.get("message", "")) if state == "blocked" else ""

    name = event.get("hook_event_name")
    agents = int(number(previous.get("agents"))) if previous else 0
    agents = max(0, min(agents, 999))
    # Whether the main agent has finished its turn. Tracked separately from the
    # state because with background subagents both are true at once: the turn is
    # over and work is still running.
    stopped = bool(previous.get("stopped")) if previous else False

    # An event that lost a race carries an older timestamp than what is already
    # stored. Its *state* must not be applied -- that is last-writer-wins, and
    # replaying an old one would resurrect a state the session has left.
    previous_ts = number(previous.get("event_ts")) if previous else 0
    if previous_ts > now + FUTURE_SLACK:
        previous_ts = 0   # corrupt, and trusting it would freeze this session
    stale = previous_ts > now

    # Applied whether or not this event lost the race above: a snapshot says
    # what was running when the event fired, and even a slightly stale one is a
    # better answer than a number left over from an older event still. Nothing
    # accumulates, so nothing leaks when a subagent dies without ever sending
    # its SubagentStop -- the next event carrying a list corrects the count.
    snapshot = running_agents(event)
    if snapshot is not None:
        agents = snapshot

    if stale:
        # Keep the stored state and flag; the snapshot above still stands.
        state = previous.get("state", state)
    else:
        if name in ("SessionStart", "UserPromptSubmit"):
            stopped = False
        elif name == "Stop":
            stopped = True
        elif name == "PostToolUse":
            stopped = False

        if name == "SubagentStop":
            # The last subagent finishing is what finally frees a session whose
            # main agent stopped long ago.
            state = "waiting" if (stopped and agents == 0) else "busy"
        elif state == "waiting" and agents > 0:
            # The turn ended but the batch is still running, and the session
            # will pick the results up itself. Calling it "waiting" would send
            # you to a terminal that does not need you.
            state = "busy"

        # A pending question outlives everything except an answer to it.
        if (previous and previous.get("state") == "blocked"
                and state != "blocked" and name not in BLOCK_CLEARING):
            state = "blocked"
            message = previous.get("message", "")

    # The pid is resolved by the caller and compared in the unchanged-check
    # below, not merely stored. A session resumed under a new pid, or one whose
    # pid was recorded wrongly, would otherwise keep the stale value for as
    # long as its state happens not to change -- and the reader, finding that
    # process gone, would drop a perfectly live session from the panel.
    if previous:
        # Auto-compaction raises SessionStart again, in the middle of a turn the
        # session is still working on. Taking it at face value would flip a busy
        # session to waiting until the next tool call corrected it.
        if event.get("hook_event_name") == "SessionStart" and event.get("source") == "compact":
            return
        # Nothing new to publish: stay quiet so the directory monitor stays quiet.
        if (previous.get("state") == state
                and previous.get("message", "") == message
                and previous.get("agents", 0) == agents
                and bool(previous.get("stopped")) == stopped
                and previous.get("pid", 0) == claude_pid):
            return

    env = read_environ(claude_pid) if claude_pid else {}

    write_atomic(path, {
        "session_id": event.get("session_id"),
        "state": state,
        "cwd": clip(event.get("cwd") or ""),
        # Age is measured from the moment the state was entered, not from the
        # last event, so "waiting 40 min" survives unrelated later writes.
        "since": previous["since"] if previous and previous.get("state") == state else now,
        # When the session itself began, as opposed to when it entered this
        # state. Seniority between chips is decided on this: a session that
        # changed state a moment ago has not become the younger of the two.
        "started": (previous.get("started") if previous else None) or now,
        "event_ts": max(now, previous_ts),
        # 0 means "could not tell"; the reader must not take that for "dead".
        "pid": claude_pid,
        "pid_start": pid_start_time(claude_pid) if claude_pid else 0,
        "event": event.get("hook_event_name", ""),
        "notification_type": event.get("notification_type", ""),
        "message": clip(message),
        "agents": agents,
        "stopped": stopped,
        # Kept from the previous write when this event could not identify the
        # process: a momentary failure should not blank out where the session is.
        "zellij_session": clip(env.get("ZELLIJ_SESSION_NAME")
            or (previous.get("zellij_session", "") if previous else "")),
    })


def main():
    now = time.time()
    try:
        event = json.load(sys.stdin)
    except (ValueError, OSError):
        return 0
    if not isinstance(event, dict):
        return 0

    session_id = event.get("session_id")
    if not session_id or "/" in session_id:
        return 0

    debug_log(event)

    state = derive_state(event)
    if state is None:
        return 0

    # After the debug log, so that a session missing from the panel can still be
    # explained by the log, and before the state file is touched at all: a
    # headless run must not even delete on SessionEnd, since it never wrote.
    claude_pid, claude_argv = find_claude()
    if is_headless(claude_argv):
        return 0

    os.makedirs(STATE_DIR, exist_ok=True)
    path = os.path.join(STATE_DIR, "%s.json" % session_id)

    # Hooks for one session run concurrently -- the last PostToolUse of a turn
    # races that turn's Stop. Comparing timestamps is not enough on its own:
    # without a lock both processes read the same "previous" and the loser's
    # write still lands last, pinning a finished session at "busy". The lock is
    # a separate file because write_atomic replaces the inode of the real one.
    lock = open_lock(path + ".lock")
    if lock is None:
        return 0
    with lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        apply_event(event, state, path, now, claude_pid)
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception:
        # A hook that fails loudly would spam every session with error output;
        # a missing panel update is the cheaper failure.
        sys.exit(0)