netsky 0.1.6

#!/usr/bin/env -S uv run
# /// script
# requires-python = ">=3.11"
# ///
"""Auto-approve Claude Code permission prompts on a tmux pane.

Polls `tmux capture-pane -t <target> -p`, scans the visible viewport
for the canonical 1/2/3 dialog ("Do you want to proceed?" + "1. Yes"),
and on match sends the digit '1' followed by Enter to the target pane.

Safety rails (do not weaken without owner sign-off):
  - Approve-only. Never sends "2" (which would write to settings) or
    "3" (deny). The only key sequence ever sent is the literal digit
    '1' followed by Enter.
  - Strict matcher. Both `Do you want to` AND `1. Yes` must be
    present; either one missing means no action. If claude-code
    reshapes the prompt and one of the needles disappears, the
    watcher quietly stops matching rather than guess.
  - Debounced. After a successful send, the watcher refuses to send
    again until a poll observes the prompt has cleared. At most one
    keystroke per prompt instance.
  - Defensive. Every tmux subprocess is bounded by a short timeout.
    Any error in capture, send, hashing, or logging is caught, logged,
    and the loop continues.

Audit log: ~/.netsky/state/permissions-watcher.log

Usage:
    uv run scripts/permissions-watcher.py [--target agent0]
                                          [--interval 2]

Stop: kill the python process (Ctrl-C, or
`pkill -f permissions-watcher.py`).
"""

from __future__ import annotations

import argparse
import hashlib
import signal
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path
from types import FrameType

LOG_PATH = Path.home() / ".netsky" / "state" / "permissions-watcher.log"
DEFAULT_TARGET = "agent0"
DEFAULT_INTERVAL = 2.0
CAPTURE_TIMEOUT = 5.0
SEND_TIMEOUT = 5.0
# Match TICKER_LOG_ROTATE_BYTES in netsky-core/src/consts.rs. On rotation
# the current log is renamed to `<path>.1`, overwriting any prior `.1`.
# Same destructive choice as the Rust side; a chained `.1 -> .2` is a
# deferred improvement.
LOG_ROTATE_BYTES = 5 * 1024 * 1024

# Canonical Claude Code permission prompt: "Do you want to proceed?"
# (or a close variant) plus a numbered choice list whose first option
# is "1. Yes". Both substrings must be present in the captured pane
# before the watcher will act.
PROMPT_NEEDLE = "Do you want to"
YES_NEEDLE = "1. Yes"


def _now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _rotate_if_needed(log_path: Path) -> None:
    """Rotate `log_path` to `<log_path>.1` once it exceeds LOG_ROTATE_BYTES.

    Best-effort. Any OS error is swallowed so a rotation hiccup never
    blocks the watcher -- logging is advisory, the keystroke path is not.
    """
    try:
        if log_path.stat().st_size < LOG_ROTATE_BYTES:
            return
    except OSError:
        return
    rotated = log_path.with_suffix(log_path.suffix + ".1")
    try:
        if rotated.exists():
            rotated.unlink()
        log_path.rename(rotated)
    except OSError:
        pass


def log(msg: str) -> None:
    line = f"{_now()} {msg}\n"
    try:
        LOG_PATH.parent.mkdir(parents=True, exist_ok=True)
        _rotate_if_needed(LOG_PATH)
        with LOG_PATH.open("a") as f:
            f.write(line)
    except OSError as exc:
        sys.stderr.write(f"log-write-failed: {exc}: {line}")
    sys.stdout.write(line)
    sys.stdout.flush()


def capture_pane(target: str) -> str | None:
    try:
        result = subprocess.run(
            ["tmux", "capture-pane", "-t", target, "-p"],
            capture_output=True,
            text=True,
            stdin=subprocess.DEVNULL,
            timeout=CAPTURE_TIMEOUT,
            check=False,
        )
    except (subprocess.TimeoutExpired, OSError) as exc:
        log(f"capture-error target={target} err={exc!r}")
        return None
    if result.returncode != 0:
        return None
    # capture-pane -p returns the whole visible viewport, padded with
    # blank lines at the bottom. Match against the full viewport rather
    # than a fixed tail: a TUI box at the top of a mostly-empty pane
    # would otherwise be missed if we only kept the last N lines.
    return result.stdout


def detect_prompt(pane: str) -> bool:
    return PROMPT_NEEDLE in pane and YES_NEEDLE in pane


def send_approval(target: str) -> bool:
    """Send '1' then Enter to the target pane. True on success."""
    try:
        # `-l` makes tmux send the digit as literal text rather than
        # looking it up as a key binding. Enter is sent as a named key
        # so claude-code's TUI sees it as <CR>.
        subprocess.run(
            ["tmux", "send-keys", "-t", target, "-l", "1"],
            timeout=SEND_TIMEOUT,
            check=True,
        )
        subprocess.run(
            ["tmux", "send-keys", "-t", target, "Enter"],
            timeout=SEND_TIMEOUT,
            check=True,
        )
    except (
        subprocess.TimeoutExpired,
        subprocess.CalledProcessError,
        OSError,
    ) as exc:
        log(f"send-error target={target} err={exc!r}")
        return False
    return True


def hash_pane(pane: str) -> str:
    return hashlib.sha256(pane.encode("utf-8", "replace")).hexdigest()[:16]


def run(target: str, interval: float) -> None:
    log(f"watcher-start target={target} interval={interval}s log={LOG_PATH}")
    # When True, we just sent an approval and are waiting for the prompt
    # to clear from the pane before we will send again. This is the
    # primary debounce: at most one keystroke per prompt instance, and
    # we only re-arm after observing at least one poll with no prompt
    # detected (i.e. the TUI has dismissed the dialog).
    awaiting_clear = False

    def shutdown(signum: int, _frame: FrameType | None) -> None:
        log(f"watcher-stop signal={signum}")
        sys.exit(0)

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    while True:
        try:
            pane = capture_pane(target)
            if pane is None:
                time.sleep(interval)
                continue

            if not detect_prompt(pane):
                if awaiting_clear:
                    log(f"prompt-cleared target={target}")
                    awaiting_clear = False
                time.sleep(interval)
                continue

            if awaiting_clear:
                # Same prompt still on screen after our keystroke; wait
                # for it to clear before doing anything else.
                time.sleep(interval)
                continue

            h = hash_pane(pane)
            log(f"prompt-detected hash={h} target={target}")
            if send_approval(target):
                log(f"approval-sent hash={h} keys=1+Enter target={target}")
                awaiting_clear = True
            else:
                # Send failed; stay un-debounced so the next tick retries.
                log(f"approval-failed hash={h} target={target}")
        except Exception as exc:  # noqa: BLE001
            # Never crash the loop on an unexpected error.
            log(f"loop-error err={exc!r}")
        time.sleep(interval)


def main() -> int:
    parser = argparse.ArgumentParser(
        description=("Auto-approve Claude Code permission prompts via tmux send-keys.")
    )
    parser.add_argument(
        "--target",
        default=DEFAULT_TARGET,
        help=f"tmux session/pane id (default: {DEFAULT_TARGET})",
    )
    parser.add_argument(
        "--interval",
        type=float,
        default=DEFAULT_INTERVAL,
        help=f"poll interval seconds (default: {DEFAULT_INTERVAL})",
    )
    args = parser.parse_args()
    if args.interval <= 0:
        parser.error("--interval must be positive")
    try:
        run(args.target, args.interval)
    except SystemExit:
        raise
    except Exception as exc:  # noqa: BLE001
        log(f"watcher-fatal err={exc!r}")
        return 1
    return 0


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