delayed_send.py

"""Sleep, then conditionally exec pushover_send.

Skips the send if the controlling tty has been read from within
TYPING_WINDOW seconds, i.e., the user is actively composing a prompt
in the Claude Code TUI. Typing triggers reads on the pty slave, which
bumps atime; the TUI's own redraws only bump mtime.

Usage:
    python delayed_send.py DELAY TTY -- send_argv...

Pass TTY="-" to skip the atime check (just delay-then-send).
"""
from __future__ import annotations

import datetime
import os
import sys
import time
from pathlib import Path

TYPING_WINDOW = 5.0      # atime within this many seconds = "currently typing"
GRACE_SECONDS = 15.0     # after initial sleep, poll this long for typing to start

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


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}] delayed[pid={os.getpid()}] {msg}\n")
    except OSError:
        pass


def _parse_argv() -> tuple[float, str, list[str]]:
    argv = sys.argv
    if len(argv) < 5 or argv[3] != "--":
        sys.stderr.write(
            f"usage: {argv[0]} DELAY_SECONDS TTY_PATH -- COMMAND [ARGS...]\n"
        )
        sys.exit(2)
    try:
        delay = float(argv[1])
    except ValueError:
        sys.stderr.write(
            f"{argv[0]}: DELAY_SECONDS must be numeric, got {argv[1]!r}\n"
        )
        sys.exit(2)
    return delay, argv[2], argv[4:]


def main() -> int:
    delay, tty, send_argv = _parse_argv()

    time.sleep(delay)

    if tty == "-":
        _log("firing (no tty)")
    else:
        # Initial check + polling grace. Catches:
        #  (a) user has been typing -> atime fresh on the very first read
        #  (b) user starts typing within GRACE_SECONDS of the wake-up
        deadline = time.time() + GRACE_SECONDS
        last_age = float("inf")
        while True:
            try:
                # Use whichever is fresher. Under relatime, atime can stop
                # updating once it passes mtime; conversely some terminal
                # paths bump only one. max(...) covers both.
                st = os.stat(tty)
                last_age = time.time() - max(st.st_atime, st.st_mtime)
            except OSError as e:
                _log(f"firing (tty stat failed: {e}) tty={tty}")
                break
            if last_age < TYPING_WINDOW:
                _log(f"skipped (typing, age {last_age:.2f}s) tty={tty}")
                return 0
            if time.time() >= deadline:
                _log(f"firing (no typing in {GRACE_SECONDS:.0f}s grace, age {last_age:.2f}s) tty={tty}")
                break
            time.sleep(1.0)

    os.execv(send_argv[0], send_argv)
    return 0


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