claude_hook.py

"""
Claude Code hook dispatcher to Pushover push notification.

Wired up from ~/.claude/settings.json. Reads hook event JSON from stdin
and selectively fires push notifications via pushover_send.py.

Filtering rules (chosen to be useful, not annoying):
    UserPromptSubmit     -> clear error flag (no push)
    Notification (live)  -> push priority 1, sound=magic, "needs input"  (bypasses DND)
    Notification (idle)  -> skip (idle_prompt fires ~60s after every Stop)
    PostToolUse (error)  -> stash error flag (no push)
    Stop (errored)       -> push priority 1, sound=falling, "errored"   (bypasses DND)
    Stop (ok)            -> push priority 1, sound=pushover, "done"     (bypasses DND)

iOS sound = haptic (coupled). Flip the iPhone ringer switch to silent if you
want Watch buzz without phone audio.

Per-project opt-out: drop a `.no-pushover` file in the project's cwd.

State files live in $TMPDIR and are namespaced by tty (or parent pid as a
fallback) so concurrent Claude Code sessions on the same host do not collide.

Rate limit: max one "done" push per RATE_LIMIT_SECONDS. "needs input" and
"errored" pushes always fire (they are time-sensitive and rare).

Debug logging is opt-in: set CLAUDE_PUSHOVER_DEBUG=1 in the environment to
append every event and outcome to /tmp/claude_pushover_debug.log.
"""
from __future__ import annotations

import datetime
import json
import os
import re
import signal
import subprocess
import sys
import tempfile
import time
from pathlib import Path

HERE = Path(__file__).resolve().parent
SEND_SCRIPT = HERE / "pushover_send.py"
DELAYED_SCRIPT = HERE / "delayed_send.py"
PYTHON = sys.executable

OPT_OUT_MARKER = ".no-pushover"
RATE_LIMIT_SECONDS = 5

_TMPDIR = Path(tempfile.gettempdir())
_session_key_cache: str | None = None

DEBUG_LOG: Path | None = (
    Path("/tmp/claude_pushover_debug.log")
    if os.environ.get("CLAUDE_PUSHOVER_DEBUG")
    else None
)


def _find_user_tty() -> str:
    """Walk up the parent process chain to find the claude TUI's tty.

    The hook itself has no controlling terminal (claude pipes stdin/stdout),
    so /dev/tty does not work. Instead we readlink fd 0 of each ancestor
    until we find a /dev/pts/N. Returns "-" if none found (skip atime check).
    """
    pid = os.getppid()
    seen: set[int] = set()
    while pid > 1 and pid not in seen:
        seen.add(pid)
        try:
            target = os.readlink(f"/proc/{pid}/fd/0")
            if target.startswith("/dev/pts/") or target.startswith("/dev/tty"):
                return target
        except OSError:
            pass
        try:
            with open(f"/proc/{pid}/status") as f:
                ppid = 0
                for line in f:
                    if line.startswith("PPid:"):
                        ppid = int(line.split()[1])
                        break
                pid = ppid
        except OSError:
            break
    return "-"


def _session_key() -> str:
    """Per-session namespace for state files so concurrent Claude sessions
    on the same host do not stomp each other's state.
    """
    global _session_key_cache
    if _session_key_cache is not None:
        return _session_key_cache
    tty = _find_user_tty()
    if tty != "-":
        _session_key_cache = Path(tty).name
    else:
        _session_key_cache = f"ppid{os.getppid()}"
    return _session_key_cache


def _state_file(name: str) -> Path:
    return _TMPDIR / f"claude_pushover_{_session_key()}_{name}"


def _log(msg: str) -> None:
    if DEBUG_LOG is None:
        return
    try:
        stamp = datetime.datetime.now().isoformat(timespec="seconds")
        with DEBUG_LOG.open("a") as f:
            f.write(f"[{stamp}] {msg}\n")
    except OSError:
        pass


# Delay bands: short turn = user is sitting here, longer delay so the next
# prompt has a chance to cancel the ping. Long turn = user walked away,
# fire immediately. UserPromptSubmit kills any still-pending delayed send.
def _delay_for_duration(seconds: float) -> int:
    if seconds < 60:
        return 20
    if seconds < 120:
        return 15
    if seconds < 300:
        return 10
    return 0


def _turn_duration() -> float:
    try:
        return time.time() - float(_state_file("turn_start").read_text().strip())
    except (OSError, ValueError):
        return float("inf")  # missing -> treat as long, no delay


def _proc_starttime(pid: int) -> int | None:
    """Field 22 of /proc/<pid>/stat: process start time in clock ticks since boot.

    Used to detect PID reuse: a recycled PID will have a different starttime,
    so we can avoid sending SIGTERM to an unrelated process.
    """
    try:
        with open(f"/proc/{pid}/stat") as f:
            data = f.read()
    except OSError:
        return None
    rparen = data.rfind(")")
    if rparen == -1:
        return None
    fields = data[rparen + 1:].split()
    try:
        return int(fields[19])  # field 22; fields[0] is field 3 after the comm parens
    except (IndexError, ValueError):
        return None


def _record_pending(pid: int) -> None:
    starttime = _proc_starttime(pid)
    payload = f"{pid}:{starttime}" if starttime is not None else f"{pid}:"
    _state_file("pending_pid").write_text(payload)


def _cancel_pending() -> None:
    pending_file = _state_file("pending_pid")
    try:
        text = pending_file.read_text().strip()
    except OSError:
        return
    pid_str, _, st_str = text.partition(":")
    try:
        pid = int(pid_str)
    except ValueError:
        pending_file.unlink(missing_ok=True)
        return
    expected_st: int | None = None
    if st_str:
        try:
            expected_st = int(st_str)
        except ValueError:
            expected_st = None
    if expected_st is not None:
        actual_st = _proc_starttime(pid)
        if actual_st != expected_st:
            _log(
                f"skipping cancel: pid {pid} starttime mismatch "
                f"(recorded {expected_st}, current {actual_st}); assuming PID was recycled"
            )
            pending_file.unlink(missing_ok=True)
            return
    try:
        os.killpg(pid, signal.SIGTERM)
        _log(f"cancelled pending send pid={pid}")
    except (ProcessLookupError, PermissionError):
        pass
    pending_file.unlink(missing_ok=True)


def _spawn_send(message: str, *, title: str, priority: int, sound: str | None,
                delay_seconds: int = 0) -> None:
    """Fire-and-forget: detach so the hook returns immediately.

    When delay_seconds > 0, wraps the send in delayed_send.py inside a new
    session so the next UserPromptSubmit can cancel it via killpg.
    """
    cmd: list = [PYTHON, SEND_SCRIPT, message,
                 "--title", title, "--priority", str(priority)]
    if sound:
        cmd += ["--sound", sound]
    if delay_seconds > 0:
        tty = _find_user_tty()
        argv: list = [PYTHON, DELAYED_SCRIPT, str(delay_seconds), tty, "--", *cmd]
        proc = subprocess.Popen(
            argv,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )
        _record_pending(proc.pid)
        _log(f"scheduled send in {delay_seconds}s pid={proc.pid} tty={tty} title={title!r}")
    else:
        subprocess.Popen(
            cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
        )


def _rate_limited() -> bool:
    try:
        last = _state_file("last_send").stat().st_mtime
        if time.time() - last < RATE_LIMIT_SECONDS:
            return True
    except FileNotFoundError:
        pass
    return False


def push(message: str, *, title: str, priority: int = 0, sound: str | None = None,
         delay_seconds: int = 0, bypass_rate_limit: bool = False) -> None:
    """Send a push. When bypass_rate_limit is True the rate-limit check is
    skipped entirely AND the rate-limit anchor is not advanced, so an urgent
    push (errored, needs-input) does not wedge subsequent done pushes.
    """
    if not bypass_rate_limit and _rate_limited():
        _log(f"rate limited, skipping push: {title}")
        return
    if not bypass_rate_limit:
        _state_file("last_send").touch()
    _log(f"push title={title!r} priority={priority} sound={sound} delay={delay_seconds} msg={message[:80]!r}")
    _spawn_send(message, title=title, priority=priority, sound=sound,
                delay_seconds=delay_seconds)


def read_hook_input() -> dict:
    try:
        raw = sys.stdin.read()
    except OSError as e:
        _log(f"read_hook_input: stdin read failed: {e}")
        return {}
    if not raw.strip():
        return {}
    try:
        return json.loads(raw)
    except json.JSONDecodeError as e:
        _log(f"read_hook_input: JSON parse failed: {e}; first 200 chars: {raw[:200]!r}")
        return {}


def project_name(data: dict) -> str:
    cwd = data.get("cwd") or os.getcwd()
    return Path(cwd).name or "claude"


def opted_out(data: dict) -> bool:
    cwd = data.get("cwd") or os.getcwd()
    return (Path(cwd) / OPT_OUT_MARKER).exists()


def tool_errored(data: dict) -> bool:
    response = data.get("tool_response") or {}
    if not isinstance(response, dict):
        return False
    if response.get("is_error") is True:
        return True
    for key in ("exitCode", "exit_code"):
        val = response.get(key)
        if isinstance(val, int) and val != 0:
            return True
    if response.get("error"):
        return True
    if str(response.get("status", "")).lower() in {"error", "failed"}:
        return True
    return False


# Anchored to start-of-string or sentence boundary so we do not match
# phrases like "could not have gone better" or "i tried but failed to" mid-prose.
# The "i (encountered|got|hit|ran into) an error" forms are unambiguous enough
# to allow anywhere; "error[:!]" requires punctuation that signals a real error.
_ERROR_PATTERNS = (
    re.compile(r"\bi (?:encountered|got|hit|ran into) an error\b", re.IGNORECASE),
    re.compile(
        r"(?:^|[.!?;]\s+|\n)\s*(?:failed to|unable to|could not|couldn't)\b",
        re.IGNORECASE,
    ),
    re.compile(r"\berror[:!]", re.IGNORECASE),
)

ERROR_NEGATIONS = (
    "no error", "without error", "no failure", "no issue", "successfully",
)


def response_indicates_error(data: dict) -> bool:
    msg = str(data.get("last_assistant_message", "") or "")
    if not msg:
        return False
    lower = msg.lower()
    if any(neg in lower for neg in ERROR_NEGATIONS):
        return False
    return any(p.search(msg) for p in _ERROR_PATTERNS)


def clean(s: str) -> str:
    """Strip light markdown noise + collapse whitespace for Watch rendering."""
    if not s:
        return ""
    s = s.replace("**", "").replace("`", "")
    return " ".join(s.split())


def truncate(s: str, n: int = 120) -> str:
    s = clean(s)
    if len(s) <= n:
        return s
    return s[: n - 3].rstrip() + "..."


def parse_pending_tool(transcript_path: str) -> tuple[str, dict] | None:
    """Scan transcript bottom-up for the most recent tool_use block.
    Returns (tool_name, tool_input) or None.
    """
    if not transcript_path:
        return None
    try:
        with open(transcript_path) as f:
            lines = f.readlines()
    except OSError:
        return None
    for line in reversed(lines):
        try:
            msg = json.loads(line)
        except json.JSONDecodeError:
            continue
        content = (msg.get("message") or {}).get("content")
        if not isinstance(content, list):
            continue
        for block in content:
            if isinstance(block, dict) and block.get("type") == "tool_use":
                return block.get("name", ""), block.get("input") or {}
    return None


def format_tool_hint(name: str, tool_input: dict, max_len: int = 110) -> str:
    """Build a short, watch-friendly hint about what the pending tool wants to do."""
    inp = tool_input or {}
    if name == "Bash":
        cmd = str(inp.get("command", "")).split("\n", 1)[0]
        return cmd[:max_len]
    if name in {"Edit", "Write", "MultiEdit", "Read"}:
        path = str(inp.get("file_path") or "")
        return Path(path).name or path
    if name == "NotebookEdit":
        path = str(inp.get("notebook_path") or "")
        return Path(path).name or path
    if name == "WebFetch":
        return str(inp.get("url", ""))[:max_len]
    if name == "WebSearch":
        return str(inp.get("query", ""))[:max_len]
    if name in {"Grep", "Glob"}:
        return str(inp.get("pattern", ""))[:max_len]
    # Generic fallback: the longest stringy value tends to be the salient one.
    # Skip None so json.dumps(None) == "null" (4 chars) does not beat shorter strings.
    best = ""
    for v in inp.values():
        if v is None:
            continue
        s = v if isinstance(v, str) else json.dumps(v)
        if len(s) > len(best):
            best = s
    return best[:max_len]


def main() -> int:
    data = read_hook_input()
    event = data.get("hook_event_name", "")
    _log(f"event={event}")

    if opted_out(data):
        _log("opted out via .no-pushover marker")
        return 0

    proj = project_name(data)

    if event == "UserPromptSubmit":
        _cancel_pending()
        _state_file("turn_start").write_text(str(time.time()))
        _state_file("error_flag").unlink(missing_ok=True)

    elif event == "Notification":
        ntype = str(data.get("notification_type", "")).lower()
        if ntype in {"idle_prompt", "idle"}:
            return 0
        pending = parse_pending_tool(data.get("transcript_path", ""))
        if pending:
            tname, tinput = pending
            hint = format_tool_hint(tname, tinput)
            body = f"{tname} -> {hint}" if hint else tname
        else:
            body = data.get("message") or "needs your attention"
        delay = _delay_for_duration(_turn_duration())
        push(truncate(body), title=f"🔵 needs input · {proj}",
             priority=1, sound="magic", delay_seconds=delay,
             bypass_rate_limit=True)

    elif event == "PostToolUse":
        if tool_errored(data):
            _state_file("error_flag").write_text("1")

    elif event == "Stop":
        errored = _state_file("error_flag").exists() or response_indicates_error(data)
        _state_file("error_flag").unlink(missing_ok=True)
        last = truncate(str(data.get("last_assistant_message", "") or ""))
        if errored:
            push(last or "Tool errored this turn",
                 title=f"🔴 errored · {proj}",
                 priority=1, sound="falling",
                 bypass_rate_limit=True)
        else:
            delay = _delay_for_duration(_turn_duration())
            push(last or "Done",
                 title=f"🟢 done · {proj}",
                 priority=1, sound="pushover", delay_seconds=delay)

    return 0


if __name__ == "__main__":
    sys.exit(main())