cc_notifier.py

#!/usr/bin/env python3
"""
cc-notifier - macOS notification system for Claude Code hooks

Note to AI: YOU MUST READ ./cc_notifier.context.md BEFORE ANALYZING OR WORKING WITH THIS FILE. It contains important details about this file.
"""

import fcntl
import json
import os
import re
import shlex
import socket
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Optional

# Constants and configuration
VERSION = "0.3.0"
SESSION_DIR = Path("/tmp/cc_notifier")
CLEANUP_AGE_SECONDS = 5 * 24 * 60 * 60
NOTIFICATION_DEDUPLICATION_THRESHOLD_SECONDS = 2.0
MAX_LOG_LINES = 2250  # Trigger trim when exceeded
TRIM_TO_LINES = 1250  # Keep newest lines after trim
HAMMERSPOON_CLI = "/Applications/Hammerspoon.app/Contents/Frameworks/hs/hs"
TERMINAL_NOTIFIER = "/opt/homebrew/bin/terminal-notifier"
PUSH_IDLE_CHECK_INTERVALS_DESKTOP = [3, 20]
PUSH_IDLE_CHECK_INTERVALS_REMOTE = [4]
PUSH_IDLE_CHECK_INTERVALS_ATTACHED = [3, 20]

# Debug configuration
DEBUG = False

# Global state for threading app path to error handler
_CURRENT_APP_PATH: Optional[str] = None


def handle_command_errors(
    command_name: str,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """Decorator to handle command errors with consistent logging and exit."""

    def decorator(func: Callable[..., Any]) -> Callable[..., Any]:
        def wrapper(*args: Any, **kwargs: Any) -> Any:
            try:
                return func(*args, **kwargs)
            except Exception as e:
                log_error(f"Command '{command_name}' failed", e)
                sys.exit(1)

        return wrapper

    return decorator


# ============================================================================
# COMMAND LINE INTERFACE - Main Entry Point and Command Dispatch
# ============================================================================


def main() -> None:
    """Main entry point for cc-notifier command."""

    # Guard against direct execution in hooks
    if not os.getenv("CC_NOTIFIER_WRAPPER"):
        print(
            "ERROR: cc_notifier.py should not be run directly in Claude Code hooks.",
            file=sys.stderr,
        )
        print("Use: cc-notifier wrapper instead", file=sys.stderr)
        print("Running directly will block Claude Code execution!", file=sys.stderr)
        sys.exit(1)

    global DEBUG
    if "--debug" in sys.argv:
        DEBUG = True
        sys.argv.remove("--debug")

    command = sys.argv[1] if len(sys.argv) > 1 else "help"
    debug_log(f"Command: {command}")
    if command in ("--version", "-v"):
        print(f"cc-notifier {VERSION}")
    elif command == "init":
        cmd_init()
    elif command == "notify":
        cmd_notify()
    elif command == "cleanup":
        cmd_cleanup()
    else:
        show_help()
        sys.exit(1)


@handle_command_errors("init")
def cmd_init() -> None:
    """Initialize session by capturing focused window ID and app path."""
    hook_data = HookData.from_stdin()
    iterm2_session_id = ""
    if is_remote_session():
        window_id, app_path = "REMOTE", "REMOTE"
        debug_log("Remote session detected, skipping window capture")
    else:
        try:
            window_id, app_path = get_focused_window_id()
            if is_iterm2_app(app_path):
                iterm2_session_id = get_iterm2_focused_session_id()
        except (RuntimeError, OSError) as e:
            window_id, app_path = "UNAVAILABLE", "UNAVAILABLE"
            debug_log(f"Window capture failed, continuing without: {e}")
    tmux_session_id = get_tmux_session_id() or ""
    save_window_id(
        hook_data.session_id, window_id, app_path, tmux_session_id, iterm2_session_id
    )


@handle_command_errors("notify")
def cmd_notify() -> None:
    """Send intelligent notification if user switched away from original window."""
    global _CURRENT_APP_PATH
    hook_data = HookData.from_stdin()
    session_file = SESSION_DIR / hook_data.session_id

    if check_deduplication(session_file):
        return

    lines = session_file.read_text().strip().split("\n")
    original_window_id = lines[0]
    app_path = lines[1]
    tmux_session_id = lines[3] if len(lines) > 3 else ""
    iterm2_session_id = lines[4] if len(lines) > 4 else ""

    # Set global app path for error handling
    _CURRENT_APP_PATH = app_path

    # Local notifications only in desktop mode
    if not is_remote_session():
        try:
            send_local_notification_if_needed(
                hook_data,
                original_window_id,
                app_path,
                tmux_session_id,
                iterm2_session_id,
            )
        except (RuntimeError, OSError) as e:
            log_error("Local notification failed, continuing to push", e)

    # Push notifications if configured
    push_config = PushConfig.from_env()
    if push_config:
        if tmux_session_id and is_tmux_session_attached(tmux_session_id):
            debug_log(
                f"tmux session {tmux_session_id} attached - using extended idle check"
            )
            intervals = PUSH_IDLE_CHECK_INTERVALS_ATTACHED
        elif is_remote_session():
            intervals = PUSH_IDLE_CHECK_INTERVALS_REMOTE
        else:
            intervals = PUSH_IDLE_CHECK_INTERVALS_DESKTOP
        debug_log(f"Push idle check intervals: {intervals}")
        check_idle_and_notify_push(hook_data, intervals)


@handle_command_errors("cleanup")
def cmd_cleanup() -> None:
    """Clean up session files and perform age-based maintenance."""
    hook_data = HookData.from_stdin()
    cleanup_session(hook_data.session_id)


def show_help() -> None:
    """Display help information."""
    print(f"""cc-notifier {VERSION}

Usage: cc-notifier [--debug] {{init|notify|cleanup|--version}}

Commands:
  init     - Initialize session (capture focused window)
  notify   - Send notification if user switched away (local + push)
  cleanup  - Clean up session files
  --version - Show version information

Options:
  --debug  - Enable debug logging with timestamps

macOS notification system for Claude Code hooks with push notification support.
Set PUSHOVER_API_TOKEN and PUSHOVER_USER_KEY to enable push notifications.""")


# ============================================================================
# CORE UTILITIES - Session Management and Data Structures
# ============================================================================


@dataclass
class HookData:
    """Data structure for Claude Code hook events."""

    session_id: str
    cwd: str = ""
    hook_event_name: str = "Stop"
    message: str = ""

    @classmethod
    def from_stdin(cls) -> "HookData":
        """Parse hook data from JSON stdin input."""
        try:
            data = json.loads(sys.stdin.read())
            valid_fields = {"session_id", "cwd", "hook_event_name", "message"}
            filtered_data = {k: v for k, v in data.items() if k in valid_fields and v}
            hook_data = cls(**filtered_data)
            debug_log(f"Hook: {hook_data.session_id}, {hook_data.hook_event_name}")
            return hook_data
        except json.JSONDecodeError as err:
            raise ValueError("Invalid JSON input from stdin") from err


def check_deduplication(session_file: Path) -> bool:
    """Check if notification should be deduplicated. Returns True if should skip."""
    try:
        with open(session_file, "r+") as f:
            fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
            lines = f.read().strip().split("\n")
            # Lines: [0]=window_id, [1]=app_name, [2]=timestamp, [3]=tmux_session_id (optional)
            if (
                time.time() - float(lines[2])
                < NOTIFICATION_DEDUPLICATION_THRESHOLD_SECONDS
            ):
                return True
            app_path = lines[1] if len(lines) > 1 else ""
            tmux_id = lines[3] if len(lines) > 3 else ""
            iterm2_session_id = lines[4] if len(lines) > 4 else ""
            f.seek(0)
            updated_content = f"{lines[0]}\n{app_path}\n{time.time()}\n{tmux_id}"
            if iterm2_session_id:
                updated_content += f"\n{iterm2_session_id}"
            f.write(updated_content)
            f.truncate()
            return False
    except BlockingIOError:
        return True


def send_local_notification_if_needed(
    hook_data: HookData,
    original_window_id: str,
    app_path: str,
    tmux_session_id: str = "",
    iterm2_session_id: str = "",
) -> None:
    """Send local notification if user switched away from original window.

    Detects three "switched away" scenarios:
    - User switched to a different window entirely
    - User switched iTerm2 tabs within the same window
    - User detached/switched tmux sessions within the same window
    """
    # Without Hammerspoon, check tmux session before sending
    if original_window_id == "UNAVAILABLE":
        if tmux_session_id and is_tmux_session_attached(tmux_session_id):
            debug_log(
                f"Window tracking unavailable but tmux session {tmux_session_id} is attached - suppressing notification"
            )
            return
        debug_log("Window tracking unavailable, sending notification unconditionally")
        title, subtitle, message = create_notification_data(hook_data)
        send_notification(title=title, subtitle=subtitle, message=message)
        return

    current_window_id, current_app_path = get_focused_window_id()
    iterm2_tab_switched = False

    if (
        original_window_id == current_window_id
        and iterm2_session_id
        and is_iterm2_app(app_path)
        and is_iterm2_app(current_app_path)
    ):
        current_iterm2_session_id = get_iterm2_focused_session_id()
        if current_iterm2_session_id and current_iterm2_session_id != iterm2_session_id:
            iterm2_tab_switched = True
            debug_log(
                "Same iTerm2 window but different session ID - user switched tabs"
            )
        elif not current_iterm2_session_id:
            debug_log(
                "Unable to read current iTerm2 session ID - falling back to window/tmux detection"
            )

    if original_window_id == current_window_id and not iterm2_tab_switched:
        # Same window, but check if user switched tmux sessions within it
        if tmux_session_id and not is_tmux_session_attached(tmux_session_id):
            debug_log(
                f"Same window but tmux session {tmux_session_id} detached - user switched tmux sessions"
            )
        else:
            debug_log("User still on original window - no local notification needed")
            return

    # User switched away - send local notification
    title, subtitle, message = create_notification_data(hook_data)

    debug_log(
        f"Sending local notification: original_window={original_window_id}, current_window={current_window_id}, notification='{title}' | '{subtitle}' | '{message}'"
    )

    send_notification(
        title=title,
        subtitle=subtitle,
        message=message,
        focus_window_id=original_window_id,
        focus_iterm2_session_id=iterm2_session_id if is_iterm2_app(app_path) else None,
    )


def save_window_id(
    session_id: str,
    window_id: str,
    app_path: str,
    tmux_session_id: str = "",
    iterm2_session_id: str = "",
) -> None:
    """Save window ID, app path, tmux, and optional iTerm2 session ID."""
    SESSION_DIR.mkdir(exist_ok=True)
    session_file = SESSION_DIR / session_id
    content = f"{window_id}\n{app_path}\n0\n{tmux_session_id}"
    if iterm2_session_id:
        content += f"\n{iterm2_session_id}"
    session_file.write_text(content)
    debug_log(
        f"Session initialized: window_id={window_id}, app_path={app_path}, tmux={tmux_session_id}, iterm2_session={iterm2_session_id}, session_file={session_file}"
    )


def load_window_id(session_id: str) -> str:
    """Load window ID from session file."""
    session_file = SESSION_DIR / session_id
    lines = session_file.read_text().strip().split("\n")
    window_id = lines[0]
    debug_log(f"Session restored: window_id={window_id}, session_file={session_file}")
    return window_id


def cleanup_session(_: str) -> None:
    """Clean up session files and perform age-based maintenance."""
    # Skip session-specific deletion due to Claude Code bug #7911 (session ID mismatch)
    cutoff_time = time.time() - CLEANUP_AGE_SECONDS
    cleaned_files = 0
    for file_path in SESSION_DIR.glob("*"):
        if not file_path.is_file():
            continue
        try:
            if file_path.stat().st_mtime < cutoff_time:
                file_path.unlink(missing_ok=True)
                cleaned_files += 1
        except OSError:
            continue

    if cleaned_files > 0 or DEBUG:
        debug_log(
            f"Session cleanup completed: removed {cleaned_files} old session files"
        )


LOG_FILE = Path.home() / ".cc-notifier" / "cc-notifier.log"


def _trim_log_if_needed() -> None:
    """Trim log file if over MAX_LOG_LINES."""
    if not LOG_FILE.exists():
        return
    lines = LOG_FILE.read_text().splitlines()
    if len(lines) <= MAX_LOG_LINES:
        return
    LOG_FILE.write_text("\n".join(lines[-TRIM_TO_LINES:]) + "\n")


def _write_log_entry(
    level: str, message: str, exception: Optional[Exception] = None
) -> None:
    """Write log entry with automatic trimming."""
    LOG_FILE.parent.mkdir(exist_ok=True)
    _trim_log_if_needed()

    entry = f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] [{level}] {message}"
    if exception:
        entry += f" - {type(exception).__name__}: {exception}"

    with open(LOG_FILE, "a") as f:
        f.write(entry + "\n")


def debug_log(message: str) -> None:
    """Log debug message when DEBUG is enabled."""
    if DEBUG:
        _write_log_entry("DEBUG", message)


def log_error(error_msg: str, exception: Optional[Exception] = None) -> None:
    """Log errors to file and send notification."""
    _write_log_entry("ERROR", error_msg, exception)

    # Determine click action: focus app if available, otherwise open log
    if _CURRENT_APP_PATH:
        execute_action = f'open "{_CURRENT_APP_PATH}"'
    else:
        execute_action = f"open {LOG_FILE}"

    # Send error notification with fallback
    try:
        run_background_command(
            [
                TERMINAL_NOTIFIER,
                "-title",
                "cc-notifier Error",
                "-message",
                error_msg,
                "-sound",
                "Basso",
                "-execute",
                execute_action,
            ]
        )
    except Exception:
        run_background_command(
            [
                "osascript",
                "-e",
                f'display notification "{error_msg}" with title "cc-notifier Error" sound name "Basso"',
            ]
        )


# ============================================================================
# ENVIRONMENT DETECTION - Remote vs Desktop Mode
# ============================================================================


def is_remote_session() -> bool:
    """Detect if running in remote SSH session."""
    ssh_conn = os.getenv("SSH_CONNECTION")
    ssh_client = os.getenv("SSH_CLIENT")
    ssh_tty = os.getenv("SSH_TTY")
    is_remote = bool(ssh_conn or ssh_client or ssh_tty)

    if is_remote:
        detected_by = []
        if ssh_conn:
            detected_by.append(f"SSH_CONNECTION={ssh_conn}")
        if ssh_client:
            detected_by.append(f"SSH_CLIENT={ssh_client}")
        if ssh_tty:
            detected_by.append(f"SSH_TTY={ssh_tty}")
        debug_log(f"Remote session detected: {', '.join(detected_by)}")

    return is_remote


def get_tmux_session_id() -> Optional[str]:
    """Get the current tmux session ID (e.g. '$20'), or None if not in tmux."""
    try:
        result = subprocess.run(
            ["tmux", "display-message", "-p", "#{session_id}"],
            capture_output=True,
            text=True,
            timeout=2,
        )
        if result.returncode == 0 and result.stdout.strip():
            session_id = result.stdout.strip()
            debug_log(f"tmux session ID: {session_id}")
            return session_id
        return None
    except (FileNotFoundError, subprocess.TimeoutExpired):
        return None


def is_tmux_session_attached(session_id: str) -> bool:
    """Check if a tmux session is currently attached (has active clients).

    Args:
        session_id: tmux session ID (e.g. '$20')

    Returns:
        True if attached count > 0, False otherwise.
    """
    try:
        result = subprocess.run(
            [
                "tmux",
                "list-sessions",
                "-f",
                f"#{{==:#{{session_id}},{session_id}}}",
                "-F",
                "#{session_attached}",
            ],
            capture_output=True,
            text=True,
            timeout=2,
        )
        if result.returncode == 0 and result.stdout.strip():
            attached_count = int(result.stdout.strip())
            debug_log(f"tmux session {session_id} attached count: {attached_count}")
            return attached_count > 0
        return False
    except (FileNotFoundError, subprocess.TimeoutExpired, ValueError):
        return False


# ============================================================================
# HAMMERSPOON INTEGRATION - Cross-Space Window Management
# ============================================================================


def get_focused_window_id() -> tuple[str, str]:
    """Get the currently focused window ID and app path using Hammerspoon CLI.

    Returns:
        Tuple of (window_id, app_path)
    """
    try:
        output = run_command(
            [
                HAMMERSPOON_CLI,
                "-c",
                "local w=hs.window.focusedWindow(); if w then local app=w:application(); print(w:id()..'|'..(app and app:path() or 'UNKNOWN')) else print('ERROR') end",
            ]
        )
        if output == "ERROR" or not output or "|" not in output:
            raise RuntimeError("Failed to get focused window ID from Hammerspoon")
        window_id, app_path = output.split("|", 1)
        return window_id, app_path
    except subprocess.TimeoutExpired as e:
        raise RuntimeError(
            f"Hammerspoon command timed out after {e.timeout} seconds"
        ) from e


def is_iterm2_app(app_path: str) -> bool:
    """Return True when app path identifies iTerm2."""
    return app_path.endswith("/iTerm.app") or app_path.endswith("/iTerm2.app")


def get_iterm2_focused_session_id() -> str:
    """Get iTerm2 focused session ID, or empty string when unavailable."""
    script_lines = [
        'tell application "iTerm2"',
        'if not running then return ""',
        "try",
        "return id of current session of current window as text",
        "on error",
        'return ""',
        "end try",
        "end tell",
    ]
    cmd = ["osascript"]
    for line in script_lines:
        cmd.extend(["-e", line])

    try:
        return run_command(cmd, timeout=5)
    except (RuntimeError, subprocess.TimeoutExpired):
        return ""


def _build_iterm2_restore_script(iterm2_session_id: str) -> str:
    """Build AppleScript that focuses iTerm2 on a specific session ID."""
    escaped_session_id = iterm2_session_id.replace("\\", "\\\\").replace('"', '\\"')
    return f"""tell application "iTerm2"
if not running then return
repeat with w in windows
  repeat with t in tabs of w
    repeat with s in sessions of t
      if (id of s as text) is "{escaped_session_id}" then
        tell w to select
        tell t to select
        tell s to select
        activate
        return
      end if
    end repeat
  end repeat
end repeat
end tell"""


def create_focus_command(
    window_id: str, iterm2_session_id: Optional[str] = None
) -> list[str]:
    """
    Create the Hammerspoon focus command for cross-space window focusing.

    This uses a dual-filter approach to avoid infinite hangs that occur
    with setCurrentSpace(nil). The approach combines windows from current
    and other spaces, then searches for the target window ID.

    If the window cannot be found or focused, shows an error notification.

    When iterm2_session_id is provided, chains an AppleScript command after
    the Hammerspoon focus to restore the specific iTerm2 tab/session.

    Args:
        window_id: The window ID to focus
        iterm2_session_id: Optional iTerm2 session ID for tab restoration

    Returns:
        List of command arguments for subprocess execution
    """
    # Template for complex dual-filter cross-space window focusing
    # This solves the macOS Spaces issue without using setCurrentSpace(nil) which causes hangs
    # Shows error notification if window can't be found
    focus_script = f"""local current = require('hs.window.filter').new():setCurrentSpace(true):getWindows()
local other = require('hs.window.filter').new():setCurrentSpace(false):getWindows()
for _,w in pairs(other) do table.insert(current, w) end
for _,w in pairs(current) do
  if w:id()=={window_id} then
    w:focus()
    require('hs.timer').usleep(300000)
    return
  end
end
require('hs.notify').new({{title="cc-notifier", informativeText="Could not restore window focus. Try reopening your terminal or IDE.", soundName="Basso"}}):send()"""
    if not iterm2_session_id:
        return [HAMMERSPOON_CLI, "-c", focus_script]

    hs_cmd = [HAMMERSPOON_CLI, "-c", focus_script]
    osascript_cmd = ["osascript", "-e", _build_iterm2_restore_script(iterm2_session_id)]
    combined = (
        f"{' '.join(shlex.quote(arg) for arg in hs_cmd)}; "
        f"{' '.join(shlex.quote(arg) for arg in osascript_cmd)}"
    )
    return ["/bin/sh", "-c", combined]


# ============================================================================
# NOTIFICATION SYSTEM - macOS Notifications with Click-to-Focus
# ============================================================================


def resolve_title_tokens(hook_data: HookData, template: str) -> dict[str, str]:
    """Build dict of built-in tokens for title formatting.

    Tokens: {hostname}, {tmux_session}, {dir}, {cwd}.
    Only resolves tokens that appear in the template string to avoid
    unnecessary subprocess calls (e.g., tmux when not in tmux).
    """
    tokens: dict[str, str] = {
        "cwd": hook_data.cwd or "",
        "dir": Path(hook_data.cwd).name if hook_data.cwd else "",
    }

    try:
        tokens["hostname"] = socket.gethostname()
    except Exception:
        tokens["hostname"] = ""

    if "{tmux_session}" in template:
        try:
            result = subprocess.run(
                ["tmux", "display-message", "-p", "#S"],
                capture_output=True,
                text=True,
                timeout=2,
            )
            tokens["tmux_session"] = (
                result.stdout.strip() if result.returncode == 0 else ""
            )
        except (FileNotFoundError, subprocess.TimeoutExpired):
            tokens["tmux_session"] = ""

    return tokens


def format_title(hook_data: HookData) -> Optional[str]:
    """Format notification title from CC_NOTIFIER_TITLE_FORMAT env var.

    Supports built-in tokens ({hostname}, {tmux_session}, {dir}, {cwd})
    and generic env var access via {env:VAR_NAME}.

    Returns None when CC_NOTIFIER_TITLE_FORMAT is not set, allowing callers
    to fall back to their own defaults.
    """
    template = os.getenv("CC_NOTIFIER_TITLE_FORMAT")
    if not template:
        return None

    # Pre-pass: resolve {env:VAR_NAME} tokens before .format()
    template = re.sub(
        r"\{env:([^}]+)\}",
        lambda m: os.getenv(m.group(1), ""),
        template,
    )

    tokens = resolve_title_tokens(hook_data, template)
    return template.format(**tokens)


def create_notification_data(
    hook_data: HookData, for_push: bool = False
) -> tuple[str, str, str]:
    """Create complete notification data (title, subtitle, message)."""
    # Generate subtitle and message
    subtitle = Path(hook_data.cwd).name if hook_data.cwd else "Task Completed"
    message = (
        hook_data.message
        if (hook_data.hook_event_name == "Notification" and hook_data.message)
        else "Completed task"
    )

    # Generate title: custom format takes over when set, otherwise use original defaults
    custom_title = format_title(hook_data)
    if custom_title is not None:
        title = custom_title
    elif for_push:
        title = subtitle
    else:
        title = "Claude Code 🔔"

    # Apply debug decorations
    if DEBUG:
        if for_push:
            now = time.time()
            dt = time.localtime(now)
            milliseconds = int((now % 1) * 1000)
            timestamp = f"{time.strftime('%H:%M:%S', dt)}.{milliseconds:03d}"
            title = f"{title} [{timestamp}]"
        else:
            title = f"\\[DEBUG] {title}"

    return title, subtitle, message


def send_notification(
    title: str,
    subtitle: str,
    message: str,
    focus_window_id: Optional[str] = None,
    focus_iterm2_session_id: Optional[str] = None,
) -> None:
    """Send a macOS notification with optional click-to-focus functionality."""
    cmd = [
        TERMINAL_NOTIFIER,
        "-title",
        title,
        "-subtitle",
        subtitle,
        "-message",
        message,
        "-sound",
        "Glass",
        "-ignoreDnD",
    ]

    # Add click-to-focus functionality if window ID provided
    if focus_window_id:
        focus_cmd = create_focus_command(focus_window_id, focus_iterm2_session_id)
        execute_cmd = " ".join(shlex.quote(arg) for arg in focus_cmd)
        cmd.extend(["-execute", execute_cmd])

    # Send notification in background
    try:
        run_background_command(cmd)
        if DEBUG:
            debug_log(f"Notification sent: focus_window_id={focus_window_id}")
    except Exception as e:
        debug_log(f"Notification failed: {type(e).__name__}")
        raise


# ============================================================================
# PUSH NOTIFICATIONS - Idle Detection and Pushover Integration
# API Documentation: https://pushover.net/api
# ============================================================================


@dataclass
class PushConfig:
    """Push notification service configuration."""

    token: str
    user: str

    @classmethod
    def from_env(cls) -> Optional["PushConfig"]:
        """Create PushConfig from environment variables."""
        token = os.getenv("PUSHOVER_API_TOKEN")
        user = os.getenv("PUSHOVER_USER_KEY")

        if token and user:
            return cls(token=token, user=user)
        return None


def build_push_url(hook_data: HookData) -> Optional[str]:
    """Build push notification URL from env var template.

    Substitutes {cwd}, {session_id}, and all title tokens ({hostname},
    {tmux_session}, {dir}, {env:VAR}) with actual values.

    Returns:
        URL with placeholders substituted, or None if not configured.
    """
    url_template = os.getenv("CC_NOTIFIER_PUSH_URL")
    if not url_template:
        return None

    # Pre-pass: resolve {env:VAR_NAME} tokens
    url_template = re.sub(
        r"\{env:([^}]+)\}",
        lambda m: os.getenv(m.group(1), ""),
        url_template,
    )

    tokens = resolve_title_tokens(hook_data, url_template)
    tokens["session_id"] = hook_data.session_id
    url = url_template.format(**tokens)
    debug_log(f"Push URL built: {url}")
    return url


def send_pushover_notification(
    config: PushConfig, title: str, message: str, url: Optional[str] = None
) -> bool:
    """Send notification via Pushover API.

    Args:
        config: Pushover API configuration
        title: Notification title
        message: Notification message
        url: Optional URL to open when notification is tapped

    Returns:
        True if Pushover API returned {"status":1}, False otherwise.
        Handles network errors, JSON parsing errors, and API failures gracefully.
    """
    # Enforce Pushover API limits: 250 char title, 1024 char message
    title = title[:250] if len(title) > 250 else title
    message = message[:1024] if len(message) > 1024 else message

    data_dict = {
        "token": config.token,
        "user": config.user,
        "title": title,
        "message": message,
    }
    if url:
        data_dict["url"] = url

    data = urllib.parse.urlencode(data_dict).encode("utf-8")

    req = urllib.request.Request(
        "https://api.pushover.net/1/messages.json",
        data=data,
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )

    try:
        with urllib.request.urlopen(req, timeout=10) as response:
            if response.status == 200:
                response_data = json.loads(response.read().decode("utf-8"))
                success = bool(response_data.get("status") == 1)
                debug_log(
                    f"Push notification result: status={response.status}, success={success}"
                )
                return success
            debug_log(
                f"Push notification result: status={response.status}, success=False"
            )
            return False
    except (urllib.error.URLError, json.JSONDecodeError, KeyError) as e:
        debug_log(f"Push notification result: error={type(e).__name__}, success=False")
        return False


def get_macos_idle_time() -> int:
    """Get macOS system idle time in seconds using ioreg."""
    try:
        output = run_command(["ioreg", "-c", "IOHIDSystem"], timeout=5)
        for line in output.splitlines():
            if "HIDIdleTime" in line:
                idle_nanoseconds = int(line.split("=", 1)[1].strip())
                return idle_nanoseconds // 1_000_000_000
        raise RuntimeError("HIDIdleTime not found in ioreg output")
    except subprocess.TimeoutExpired as e:
        raise RuntimeError(f"ioreg command timed out after {e.timeout} seconds") from e


def get_tty_idle_time() -> int:
    """Get TTY idle time in seconds based on last read operation (user input)."""
    try:
        # Use TTY path captured by wrapper before backgrounding
        # This gives us the actual terminal device (works with tmux/screen)
        tty_path = os.getenv("CC_NOTIFIER_TTY")
        if not tty_path:
            raise RuntimeError("CC_NOTIFIER_TTY not set by wrapper")

        debug_log(f"TTY detection: CC_NOTIFIER_TTY={tty_path!r}")
        tty_stat = os.stat(tty_path)
        last_read_time = tty_stat.st_atime
        current_time = time.time()
        idle_seconds = int(current_time - last_read_time)
        debug_log(
            f"TTY idle: path={tty_path}, st_atime={last_read_time:.1f}, current={current_time:.1f}, idle={idle_seconds}s"
        )
        return idle_seconds
    except (OSError, ValueError) as e:
        debug_log(f"TTY idle error: {type(e).__name__}: {e}")
        raise RuntimeError("Unable to get TTY idle time") from e


def get_idle_time() -> int:
    """Get idle time in seconds, environment-aware."""
    if is_remote_session():
        return get_tty_idle_time()
    return get_macos_idle_time()


def check_idle_and_notify_push(hook_data: HookData, check_times: list[int]) -> None:
    """Check if user is idle at specified intervals and send push notification if away.

    Simple logic: If idle time is less than elapsed time, user was active during check period.
    """
    push_config = PushConfig.from_env()
    if not push_config:
        return

    if not check_times:
        raise ValueError("check_times cannot be empty")

    mode = "remote" if is_remote_session() else "desktop"
    debug_log(f"Push check started: mode={mode}")

    previous_time = 0
    for check_time in check_times:
        time.sleep(check_time - previous_time)

        try:
            idle_time = get_idle_time()
            # If idle time < elapsed time, user was active during check period
            user_active = idle_time < check_time

            debug_log(
                f"Push check: elapsed={check_time}s, idle={idle_time}s, "
                f"user_active={user_active}"
            )

            if user_active:
                debug_log("Push check exit: User is active")
                return
        except RuntimeError as e:
            debug_log(f"Push check exit: idle detection error ({e})")
            return

        previous_time = check_time

    # User has been idle through all checks, send push notification
    title, _, message = create_notification_data(hook_data, for_push=True)
    push_url = build_push_url(hook_data)
    debug_log(f"Sending push notification: '{title}'")
    send_pushover_notification(push_config, title, message, url=push_url)


# ============================================================================
# SUBPROCESS UTILITIES - Common patterns for external command execution
# ============================================================================


def run_command(cmd: list[str], timeout: int = 10) -> str:
    """Run command and return stdout, raising RuntimeError on failure."""
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
    if result.returncode != 0:
        raise RuntimeError(f"Command failed: {' '.join(cmd)}")
    return result.stdout.strip()


def run_background_command(cmd: list[str]) -> None:
    """Run command in background (non-blocking)."""
    subprocess.Popen(cmd)


if __name__ == "__main__":
    main()