Show Claude Code session status in the GNOME panel
Answers one question at a glance: is any session waiting for me, and
which one. With several sessions open the cost is not knowing what each
is doing, it is noticing that one stopped an hour ago.
Claude Code hooks write one JSON file per session under
~/.local/state/claude-code-status; the extension watches the directory
with Gio.FileMonitor, so nothing polls and there is no daemon.
Two distinctions carry the design:
* blocked (permission prompt) is kept apart from waiting (turn done).
Merged, a finished task looks as urgent as a stuck one, which is
exactly the judgement the indicator exists to make.
* the panel names the oldest session in the top state, not the latest.
The session you forget is the one that has been waiting longest.
PostToolUse is registered although it looks redundant: it is the only
event that fires after a permission is granted, so without it a session
stays blocked in the panel for the rest of the turn. It writes only on
an actual state change, so the usual case costs no I/O.
Stop and SessionEnd are synchronous, unlike the rest. Both fire as the
process is about to go quiet, and an async hook racing that exit gets
killed before it writes -- claude -p left a session pinned at busy.
Concurrent hooks for one session serialise on an flock plus a timestamp
guard; tests/test-hook.sh covers each separately, because the burst test
passes on the timestamp guard alone.
Sessions running in zellij are located by tab name rather than by path,
matched through dump-layout on the working directory. The dump carries
no pane ids, so ZELLIJ_PANE_ID cannot be used; rows that do not resolve
stay inert instead of pretending a click does something.
lib/sessions.js deliberately imports nothing from the shell resource
namespace, which lets the riskiest logic -- liveness, ordering, partial
reads, monitoring -- run under plain gjs in tests/test-sessions.js.
This commit is contained in:
Executable
+291
@@ -0,0 +1,291 @@
|
||||
#!/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 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.
|
||||
NOTIFICATION_STATES = {
|
||||
"permission_prompt": "blocked",
|
||||
"agent_needs_input": "blocked",
|
||||
"elicitation_dialog": "blocked",
|
||||
"idle_prompt": "waiting",
|
||||
"agent_completed": "waiting",
|
||||
}
|
||||
|
||||
EVENT_STATES = {
|
||||
"SessionStart": "idle",
|
||||
"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")
|
||||
if not os.path.exists(marker):
|
||||
return
|
||||
try:
|
||||
with open(marker, "a") as fh:
|
||||
fh.write(json.dumps(event, 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"
|
||||
if name == "Notification":
|
||||
return NOTIFICATION_STATES.get(event.get("notification_type"))
|
||||
# A subagent finishing its own turn is not the session becoming free; the
|
||||
# main agent is still working. SubagentStop is a distinct event and is not
|
||||
# registered, but Stop carries agent_id when raised inside an agent.
|
||||
if name == "Stop" and event.get("agent_id"):
|
||||
return None
|
||||
return EVENT_STATES.get(name)
|
||||
|
||||
|
||||
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_cmdline(pid):
|
||||
try:
|
||||
with open("/proc/%d/cmdline" % pid, "rb") as fh:
|
||||
return fh.read().replace(b"\0", b" ").decode("utf-8", "replace")
|
||||
except OSError:
|
||||
return ""
|
||||
|
||||
|
||||
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 find_claude_pid():
|
||||
"""Nearest ancestor that is the claude process itself, or 0 if unknown.
|
||||
|
||||
The hook is spawned through a shell, so the immediate parent is usually 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 pid is gone, and the obvious fallback -- the shell that spawned this
|
||||
hook -- exits milliseconds later, which would make the session flicker in
|
||||
and out of the panel forever.
|
||||
"""
|
||||
pid = os.getppid()
|
||||
for _ in range(6):
|
||||
if pid <= 1:
|
||||
break
|
||||
# Matched anywhere in the command line, not just argv[0]: installs that
|
||||
# run it as `node .../claude/cli.js` are just as valid as a direct one.
|
||||
if "claude" in read_cmdline(pid):
|
||||
return pid
|
||||
pid = parent_of(pid)
|
||||
return 0
|
||||
|
||||
|
||||
def alive(pid):
|
||||
return pid > 0 and os.path.exists("/proc/%d" % pid)
|
||||
|
||||
|
||||
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:
|
||||
pid = json.load(fh).get("pid", 0)
|
||||
except (OSError, ValueError, AttributeError):
|
||||
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):
|
||||
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 apply_event(event, state, path, now):
|
||||
"""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":
|
||||
for victim in (path, path + ".lock"):
|
||||
try:
|
||||
os.unlink(victim)
|
||||
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 = event.get("message", "") if state == "blocked" else ""
|
||||
|
||||
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 idle until the next tool call corrected it.
|
||||
if event.get("hook_event_name") == "SessionStart" and event.get("source") == "compact":
|
||||
return
|
||||
# Refuse events that lost a race with a newer one.
|
||||
if previous.get("event_ts", 0) > now:
|
||||
return
|
||||
# Nothing new to publish: stay quiet so the directory monitor stays quiet.
|
||||
if previous.get("state") == state and previous.get("message", "") == message:
|
||||
return
|
||||
|
||||
claude_pid = find_claude_pid()
|
||||
env = read_environ(claude_pid) if claude_pid else {}
|
||||
|
||||
write_atomic(path, {
|
||||
"session_id": event.get("session_id"),
|
||||
"state": state,
|
||||
"cwd": 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,
|
||||
"event_ts": now,
|
||||
# 0 means "could not tell"; the reader must not take that for "dead".
|
||||
"pid": claude_pid,
|
||||
"event": event.get("hook_event_name", ""),
|
||||
"notification_type": event.get("notification_type", ""),
|
||||
"message": message,
|
||||
"zellij_session": env.get("ZELLIJ_SESSION_NAME", ""),
|
||||
"zellij_pane": env.get("ZELLIJ_PANE_ID", ""),
|
||||
"transcript": event.get("transcript_path", ""),
|
||||
})
|
||||
|
||||
|
||||
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
|
||||
|
||||
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.
|
||||
with open(path + ".lock", "w") as lock:
|
||||
fcntl.flock(lock, fcntl.LOCK_EX)
|
||||
apply_event(event, state, path, now)
|
||||
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)
|
||||
Reference in New Issue
Block a user