diff --git a/cronicle.hcl b/cronicle.hcl new file mode 100644 index 0000000..3aa6d64 --- /dev/null +++ b/cronicle.hcl @@ -0,0 +1,24 @@ +# pi_wake_agent schedules (cronicle backend). +# +# IMPORTANT (see AGENTS.md): on this host the cronicle daemon is NOT running +# (not supervised), so these schedules are INERT. The LIVE scheduler is system +# crnd via crontab (see `crontab -l`). This HCL is only served when +# `cronicle run --path ` is supervised. The pi_test 20m wake is +# installed into crontab as well, so it actually fires regardless. + +schedule "pi_wake_pi_test_20m" { + cron = "@every 20m" + + task "wake" { + command = ["/mnt/dolphinng5_predict/pi_wake_agent.py", "--run", "--interval", "20m", "--sessions", "pi_test", "--pane", "terminal_1", "--msg", "Please carefully continue your work"] + } +} + +# Sample multi-session schedule (kept from the original config). +schedule "pi_wake_s1_s2_1h" { + cron = "@every 1h" + + task "wake" { + command = ["/mnt/dolphinng5_predict/pi_wake_agent.py", "--run", "--interval", "1h", "--sessions", "s1,s2", "--msg", "test"] + } +} diff --git a/pi_wake_agent.py b/pi_wake_agent.py index 7d5b3d0..e4e6020 100644 --- a/pi_wake_agent.py +++ b/pi_wake_agent.py @@ -17,6 +17,7 @@ import argparse import logging import os import re +import shutil import shlex import signal import subprocess @@ -36,8 +37,20 @@ AGENT_NICK = "pi_nvnemo" H5I_AGENT = "pi_nvnemo" H5I_BUS_ROOT = Path("/mnt/dolphinng5_predict") DEFAULT_INTERVAL = "1h" +CRONICLE_CONFIG = Path(os.environ.get("PI_WAKE_CRONICLE_CONFIG", "/mnt/dolphinng5_predict/cronicle.hcl")) +BACKEND_ENV = "PI_WAKE_BACKEND" +CRON_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" +PYTHON_BIN = "/usr/bin/python3" + +# Bootstrap PATH at import: crnd fires this script under a minimal env, so +# make zjet/h5i/cronicle discoverable for the subprocess calls below. (The +# shebang resolves python3 via the caller PATH; the script is stdlib-only.) +os.environ["PATH"] = CRON_PATH + os.environ.get("PATH", "") # ─── Logging Setup ─────────────────────────────────────────────────────── +ZELIG_BIN = 'zellij' + + def setup_logging(debug: bool = False) -> logging.Logger: log_rotate() logger = logging.getLogger("pi_wake_agent") @@ -115,50 +128,234 @@ def interval_to_human(interval: str) -> str: return interval -# ─── Cron Management ──────────────────────────────────────────────────── +# ─── Cronicle Support ──────────────────────────────────────────────────── +def cronicle_available() -> bool: + """True if the cronicle binary is installed (backend present).""" + return shutil.which("cronicle") is not None + + +def cronicle_daemon_running() -> bool: + """True if a cronicle daemon is actually serving (binary alone is not enough). + + On this host cronicle has been flaky (the walker never ran) and its daemon + is NOT supervised (see AGENTS.md). Schedules written to cronicle.hcl only + fire while a daemon is serving it. + """ + try: + result = subprocess.run(["pgrep", "-f", "cronicle run --path"], + capture_output=True, text=True, timeout=5) + return result.returncode == 0 + except Exception: + return False + + +def use_cronicle() -> bool: + """Decide the live scheduler backend. + + PI_WAKE_BACKEND override (cronicle|crontab) wins for tests/ops. Otherwise + cronicle is used only when the binary is installed AND its daemon is + running; otherwise crontab (system crnd is the live scheduler here). + """ + backend = os.environ.get(BACKEND_ENV, "").strip().lower() + if backend == "cronicle": + return True + if backend == "crontab": + return False + return cronicle_available() and cronicle_daemon_running() + + def cron_comment(sessions: List[str], interval: str) -> str: sessions_str = ",".join(sessions) return f"{CRON_COMMENT_PREFIX}:{sessions_str}:{interval}" -def install_cron(sessions: List[str], interval: str, message: str, logger: logging.Logger) -> None: - cron_sched = interval_to_cron(interval) - comment = cron_comment(sessions, interval) - sessions_arg = f"'{','.join(sessions)}'" - cmd = f"cd {H5I_BUS_ROOT} && export H5I_AGENT={H5I_AGENT} && {SCRIPT_PATH} --run --sessions {sessions_arg} --msg {shlex.quote(message)}" +def cronicle_config_path() -> Path: + """Return path to cronicle config file.""" + return CRONICLE_CONFIG + + +def cronicle_schedule_name(sessions: List[str], interval: str) -> str: + """Generate unique schedule name for cronicle.""" + sessions_str = "_".join(sessions) + return f"pi_wake_{sessions_str}_{interval}" + + +def wake_command_args(sessions: List[str], interval: str, message: str, pane: Optional[str] = None) -> List[str]: + """Build the argv list for the --run wake command (shared by both backends). + + Always passes --interval so the cronicle/crnd entry is self-describing. + --pane targets a specific zjet pane (e.g. the bottom Soundcheck pane). + """ + args = [str(SCRIPT_PATH), "--run", "--interval", interval, "--sessions", ",".join(sessions)] + if pane: + args += ["--pane", pane] + args += ["--msg", message] + return args + + +def cronicle_install(sessions: List[str], interval: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install wake schedule using cronicle HCL config.""" + schedule_name = cronicle_schedule_name(sessions, interval) - result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) - existing = result.stdout if result.returncode == 0 else "" - lines = [line for line in existing.splitlines() if comment not in line] - lines.append(f"{cron_sched} {cmd} # {comment}") - new_cron = "\n".join(lines) + "\n" - subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True) - logger.info(f"Installed cron: {cron_sched} -> {sessions} every {interval}") + # Build task command + sessions_arg = f"'{','.join(sessions)}'" + cmd = f"cd {H5I_BUS_ROOT} && export H5I_AGENT={H5I_AGENT} && {SCRIPT_PATH} --run --interval {interval} --sessions {sessions_arg} --msg {shlex.quote(message)}" + + # Read existing config + config_path = cronicle_config_path() + existing = "" + if config_path.exists(): + existing = config_path.read_text() + + # Remove existing schedule with same name + lines = existing.splitlines() + new_lines = [] + i = 0 + while i < len(lines): + if f'schedule "{schedule_name}"' in lines[i]: + # Skip this schedule block (find closing brace) + brace_count = 0 + while i < len(lines): + brace_count += lines[i].count('{') + brace_count -= lines[i].count('}') + i += 1 + if brace_count == 0: + break + continue + new_lines.append(lines[i]) + i += 1 + + # Add new schedule + cron_sched = interval_to_cron(interval) + cron_sched = cron_sched.replace("*/", "@every ").replace("0 */", "@every ").replace(" * * *", "") + # Convert cron to cronicle format + match = re.match(r"^(\d+)([hm])$", interval) + if match: + value, unit = int(match.group(1)), match.group(2) + if unit == "h": + cronicle_cron = f"@every {value}h" + else: + cronicle_cron = f"@every {value}m" + else: + cronicle_cron = "@every 1h" + + new_schedule = f''' +schedule "{schedule_name}" {{ + cron = "{cronicle_cron}" + + task "wake" {{ + command = [{", ".join(chr(34) + a + chr(34) for a in wake_command_args(sessions, interval, message, pane))}] + }} +}}''' + + new_config = "\n".join(new_lines) + "\n\n" + new_schedule + "\n" + config_path.write_text(new_config) + logger.info(f"Installed cronicle schedule: {schedule_name} -> {sessions} every {interval}") -def remove_cron(sessions: List[str], interval: str, logger: logging.Logger) -> None: - comment = cron_comment(sessions, interval) - result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) - existing = result.stdout if result.returncode == 0 else "" - lines = [line for line in existing.splitlines() if comment not in line] - new_cron = "\n".join(lines) + ("\n" if lines else "") - subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True) - logger.info(f"Removed cron for {sessions} ({interval})") +def cronicle_remove(sessions: List[str], interval: str, logger: logging.Logger) -> None: + """Remove wake schedule from cronicle HCL config.""" + schedule_name = cronicle_schedule_name(sessions, interval) + config_path = cronicle_config_path() + + if not config_path.exists(): + return + + existing = config_path.read_text() + lines = existing.splitlines() + new_lines = [] + i = 0 + while i < len(lines): + if f'schedule "{schedule_name}"' in lines[i]: + brace_count = 0 + while i < len(lines): + brace_count += lines[i].count('{') + brace_count -= lines[i].count('}') + i += 1 + if brace_count == 0: + break + continue + new_lines.append(lines[i]) + i += 1 + + config_path.write_text("\n".join(new_lines) + "\n") + logger.info(f"Removed cronicle schedule: {schedule_name}") -def list_cron(logger: logging.Logger) -> None: - print("=== pi_wake_agent cron entries ===") - result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) - existing = result.stdout if result.returncode == 0 else "" +def cronicle_list(logger: logging.Logger) -> None: + """List cronicle wake schedules.""" + config_path = cronicle_config_path() + if not config_path.exists(): + print(" (none)") + return + + existing = config_path.read_text() found = False for line in existing.splitlines(): - if CRON_COMMENT_PREFIX in line: - print(f" {line}") + if 'schedule "pi_wake_' in line: + print(f" {line.strip()}") found = True if not found: print(" (none)") + +def install_cron(sessions: List[str], interval: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + if use_cronicle(): + if not cronicle_daemon_running(): + logger.warning( + "cronicle binary present but its daemon is NOT running " + "schedules written to cronicle.hcl will not fire; set " + "PI_WAKE_BACKEND=crontab to use system crnd." + ) + cronicle_install(sessions, interval, message, logger, pane) + else: + cron_sched = interval_to_cron(interval) + comment = cron_comment(sessions, interval) + sessions_arg = f"'{','.join(sessions)}'" + run_cmd = f"{PYTHON_BIN} {SCRIPT_PATH} --run --interval {interval} --sessions {sessions_arg} --msg {shlex.quote(message)}" + if pane: + run_cmd += f" --pane {shlex.quote(pane)}" + cmd = f"cd {H5I_BUS_ROOT} && export PATH={CRON_PATH} && export H5I_AGENT={H5I_AGENT} && {run_cmd}" + + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + lines = [line for line in existing.splitlines() if comment not in line] + lines.append(f"{cron_sched} {cmd} # {comment}") + new_cron = "\n".join(lines) + "\n" + subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True) + logger.info(f"Installed cron: {cron_sched} -> {sessions} every {interval}") + + +def remove_cron(sessions: List[str], interval: str, logger: logging.Logger) -> None: + if use_cronicle(): + cronicle_remove(sessions, interval, logger) + else: + comment = cron_comment(sessions, interval) + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + lines = [line for line in existing.splitlines() if comment not in line] + new_cron = "\n".join(lines) + ("\n" if lines else "") + subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True) + logger.info(f"Removed cron for {sessions} ({interval})") + + +def list_cron(logger: logging.Logger) -> None: + if use_cronicle(): + cronicle_list(logger) + else: + print("=== pi_wake_agent cron entries ===") + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + found = False + for line in existing.splitlines(): + if CRON_COMMENT_PREFIX in line: + print(f" {line}") + found = True + if not found: + print(" (none)") + + # ─── Zellij Operations ────────────────────────────────────────────────── def zellij_session_exists(session: str) -> bool: try: @@ -172,19 +369,26 @@ def zellij_session_exists(session: str) -> bool: return False -def zellij_write_chars(session: str, text: str) -> bool: +def zellij_write_chars(session: str, text: str, pane: Optional[str] = None) -> bool: + """Write text into a zjet pane (focused, or --pane-id when given).""" + cmd = [ZELIG_BIN, "--session", session, "action", "write-chars"] + if pane: + cmd += ["--pane-id", pane] + cmd.append(text) try: - subprocess.run(["zellij", "--session", session, "action", "write-chars", text], - capture_output=True, timeout=5) + subprocess.run(cmd, capture_output=True, timeout=5) return True except Exception: return False -def zellij_write_enter(session: str) -> bool: +def zellij_write_enter(session: str, pane: Optional[str] = None) -> bool: + """Press ENTER (CR) in a zjet pane (focused, or --pane-id when given).""" + cmd = [ZELIG_BIN, "--session", session, "action", "write", "13"] + if pane: + cmd += ["--pane-id", pane] try: - subprocess.run(["zellij", "--session", session, "action", "write", "13"], - capture_output=True, timeout=5) + subprocess.run(cmd, capture_output=True, timeout=5) return True except Exception: return False @@ -195,7 +399,7 @@ def zellij_write_enter(session: str) -> bool: def _wake(): time.sleep(interval_seconds) - run_wake(sessions, message, logger) + run_wake(sessions, message, logger, pane) thread = threading.Thread(target=_wake, daemon=True) thread.start() @@ -204,15 +408,21 @@ def zellij_write_enter(session: str) -> bool: pid_file.write_text(str(os.getpid())) -def run_wake(sessions: List[str], message: str, logger: logging.Logger) -> None: +def run_wake(sessions: List[str], message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: sessions_str = ",".join(sessions) - logger.info(f"Waking {sessions_str} with message: {message}") + logger.info(f"Waking {sessions_str} with message: {message}" + (f" (pane={pane})" if pane else "")) for session in sessions: - zellij_write_chars(session, f"[{AGENT_NICK} via zellij] {message} Run: h5i-bus msg inbox") - for _ in range(5): - zellij_write_enter(session) - time.sleep(1) + # Verify session exists before attempting to wake + if not zellij_session_exists(session): + logger.warning(f"Session {session} does not exist, skipping") + continue + + zellij_write_chars(session, f"[{AGENT_NICK} via zellij] {message} Run: h5i-bus msg inbox", pane) + # Send more ENTER keypresses with longer delay for reliability + for _ in range(10): + zellij_write_enter(session, pane) + time.sleep(0.5) # Fire-and-forget h5i bus message (non-blocking) def _send_bus(): @@ -231,13 +441,13 @@ def run_wake(sessions: List[str], message: str, logger: logging.Logger) -> None: logger.info(f"Wake sent to {sessions_str}") -def run_once(sessions: List[str], interval: str, message: str, logger: logging.Logger) -> None: +def run_once(sessions: List[str], interval: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: interval_seconds = parse_interval(interval) logger.info(f"One-shot timer set for {interval} ({interval_to_human(interval)})") def _wake(): time.sleep(interval_seconds) - run_wake(sessions, message, logger) + run_wake(sessions, message, logger, pane) thread = threading.Thread(target=_wake, daemon=True) thread.start() @@ -247,22 +457,22 @@ def run_once(sessions: List[str], interval: str, message: str, logger: logging.L return 0 -def run_daemon(sessions: List[str], interval: str, message: str, logger: logging.Logger) -> None: +def run_daemon(sessions: List[str], interval: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: interval_seconds = parse_interval(interval) logger.info("=== DAEMON START ===") logger.info(f"Interval: {interval} ({interval_to_human(interval)})") logger.info(f"Sessions: {sessions}") logger.info(f"Message: {message}") - run_wake(sessions, message, logger) + run_wake(sessions, message, logger, pane) while True: logger.debug(f"Sleeping for {interval_seconds}s...") time.sleep(interval_seconds) - run_wake(sessions, message, logger) + run_wake(sessions, message, logger, pane) -def run_succession(sessions: List[str], interval: str, count: int, message: str, logger: logging.Logger) -> None: +def run_succession(sessions: List[str], interval: str, count: int, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: """Run wake N times at interval, then self-clean (remove cron if installed).""" interval_seconds = parse_interval(interval) logger.info(f"=== SUCCESSION START === Count: {count}, Interval: {interval} ({interval_to_human(interval)})") @@ -271,7 +481,7 @@ def run_succession(sessions: List[str], interval: str, count: int, message: str, for i in range(1, count + 1): logger.info(f"Succession {i}/{count}") - run_wake(sessions, message, logger) + run_wake(sessions, message, logger, pane) if i < count: logger.debug(f"Sleeping for {interval_seconds}s until next succession...") time.sleep(interval_seconds) @@ -292,6 +502,17 @@ def status(logger: logging.Logger) -> None: print(f"Log: {LOG_FILE}") print(f"Log size: {LOG_FILE.stat().st_size if LOG_FILE.exists() else 'N/A'} bytes") print(f"Agent: {AGENT_NICK}") + backend = os.environ.get(BACKEND_ENV, "").strip().lower() + if backend == "cronicle": + backend_label = "cronicle (forced by PI_WAKE_BACKEND)" + elif backend == "crontab": + backend_label = "crontab (forced by PI_WAKE_BACKEND)" + elif use_cronicle(): + backend_label = "cronicle" + else: + backend_label = "crontab (system crnd)" + print(f"Backend: {backend_label}") + print(f"cronicle binary={cronicle_available()} daemon_running={cronicle_daemon_running()}") print() list_cron(logger) print() @@ -363,10 +584,6 @@ EXAMPLES: pi_wake_agent.py --validate --session cc_UV_dev0_Fb """ ) - parser = argparse.ArgumentParser( - description="pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession", - formatter_class=argparse.RawDescriptionHelpFormatter, - ) parser.add_argument("--install", action="store_const", const="install", dest="mode", help="Install recurring cron timer") parser.add_argument("--once", action="store_const", const="once", dest="mode", help="One-shot wake (no cron)") parser.add_argument("--daemon", action="store_const", const="daemon", dest="mode", help="Run as long-lived daemon (no cron)") @@ -380,6 +597,7 @@ EXAMPLES: parser.add_argument("--interval", default=DEFAULT_INTERVAL, help="Interval (default: 1h). Formats: 30m, 1h, 90m, 2h, etc.") parser.add_argument("--session", action="append", dest="session_list", help="Zellij session name (can repeat)") parser.add_argument("--sessions", help="Comma-separated list of sessions") + parser.add_argument("--pane", dest="pane", default=None, help="Target a specific zjet pane (focused pane if unset), e.g. terminal_1") parser.add_argument("--msg", default=f"Operator says CONTINUE. {AGENT_NICK} here, saying hi!", help="Wake message") parser.add_argument("--debug", action="store_true", help="Enable debug logging") parser.add_argument("--dry-run", action="store_true", help="Show what would be done without executing") @@ -415,6 +633,8 @@ def main() -> int: logger.info(f"=== pi_wake_agent {args.mode} ===") logger.info(f"Interval: {args.interval} ({interval_to_human(args.interval)})") logger.info(f"Sessions: {sessions}") + if args.pane: + logger.info(f"Pane target: {args.pane}") logger.info(f"Message: {args.msg}") if args.mode == "succession": logger.info(f"Count: {args.count}") @@ -427,15 +647,15 @@ def main() -> int: try: if args.mode == "install": - install_cron(sessions, args.interval, args.msg, logger) + install_cron(sessions, args.interval, args.msg, logger, args.pane) elif args.mode == "once": - run_once(sessions, args.interval, args.msg, logger) + run_once(sessions, args.interval, args.msg, logger, args.pane) elif args.mode == "daemon": - run_daemon(sessions, args.interval, args.msg, logger) + run_daemon(sessions, args.interval, args.msg, logger, args.pane) elif args.mode == "succession": - run_succession(sessions, args.interval, args.count, args.msg, logger) + run_succession(sessions, args.interval, args.count, args.msg, logger, args.pane) elif args.mode == "run": - run_wake(sessions, args.msg, logger) + run_wake(sessions, args.msg, logger, args.pane) elif args.mode == "remove": remove_cron(sessions, args.interval, logger) elif args.mode == "list": diff --git a/pi_wake_agent.sh b/pi_wake_agent.sh new file mode 100644 index 0000000..1d14bf0 --- /dev/null +++ b/pi_wake_agent.sh @@ -0,0 +1,368 @@ +#!/usr/bin/env bash +# pi_wake_agent.sh — Reusable multi-agent wake-up timer with self-cron/daemon +set -euo pipefail + +# ─── Constants ──────────────────────────────────────────────────────────── +SCRIPT_PATH="$(readlink -f "${BASH_SOURCE[0]}")" +LOG_FILE="/tmp/pi_wake_agent.log" +LOG_MAX_SIZE=$((10 * 1024 * 1024)) +LOG_MAX_FILES=5 +CRON_COMMENT_PREFIX="pi_wake_agent" +AGENT_NICK="pi_nvnemo" +H5I_AGENT="pi_nvnemo" +H5I_BUS_ROOT="/mnt/dolphinng5_predict" +DEFAULT_INTERVAL="1h" +PANE="" +CRON_PATH="/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" +PYTHON_BIN="/usr/bin/python3" + +# ─── Logging ────────────────────────────────────────────────────────────── +log_rotate() { + [[ -f "$LOG_FILE" ]] || return 0 + local size + size=$(stat -c%s "$LOG_FILE" 2>/dev/null || echo 0) + [[ $size -lt $LOG_MAX_SIZE ]] && return 0 + for i in $(seq $((LOG_MAX_FILES - 1)) -1 1); do + [[ -f "${LOG_FILE}.${i}" ]] && mv "${LOG_FILE}.${i}" "${LOG_FILE}.$((i + 1))" + done + mv "$LOG_FILE" "${LOG_FILE}.1" +} + +log_write() { + local level="$1"; shift + local msg="$*" + local ts + ts=$(date '+%Y-%m-%d %H:%M:%S') + log_rotate + echo "[$ts] [$level] $msg" >> "$LOG_FILE" + if [[ "$level" == "ERROR" || "$level" == "WARN" ]]; then + echo "[$ts] [$level] $msg" >&2 + fi + return 0 +} + +log_debug() { [[ ${DEBUG:-0} == "1" ]] && log_write "DEBUG" "$@"; } +log_info() { log_write "INFO" "$@"; } +log_warn() { log_write "WARN" "$@"; } +log_error() { log_write "ERROR" "$@"; } + +# ─── Interval Parsing ──────────────────────────────────────────────────── +parse_interval() { + local interval="$1" + local seconds=0 + if [[ "$interval" =~ ^([0-9]+)h$ ]]; then + seconds=$((BASH_REMATCH[1] * 3600)) + elif [[ "$interval" =~ ^([0-9]+)m$ ]]; then + seconds=$((BASH_REMATCH[1] * 60)) + elif [[ "$interval" =~ ^([0-9]+)s$ ]]; then + seconds=$((BASH_REMATCH[1])) + else + log_error "Invalid interval format '$interval'. Use like 1h, 30m, 90m, 2h" + return 1 + fi + echo "$seconds" + return 0 +} + +interval_to_cron() { + local interval="$1" + if [[ "$interval" =~ ^([0-9]+)h$ ]]; then + local hours="${BASH_REMATCH[1]}" + echo "0 */${hours} * * *" + elif [[ "$interval" =~ ^([0-9]+)m$ ]]; then + local mins="${BASH_REMATCH[1]}" + if (( mins >= 60 )); then + log_error "For minutes >= 60, use hours (e.g., 1h not 60m)" + return 1 + fi + echo "*/${mins} * * * *" + else + log_error "Cron only supports minutes/hours intervals" + return 1 + fi +} + +interval_to_human() { + local interval="$1" + if [[ "$interval" =~ ^([0-9]+)h$ ]]; then + echo "${BASH_REMATCH[1]} hour(s)" + elif [[ "$interval" =~ ^([0-9]+)m$ ]]; then + echo "${BASH_REMATCH[1]} minute(s)" + elif [[ "$interval" =~ ^([0-9]+)s$ ]]; then + echo "${BASH_REMATCH[1]} second(s)" + else + echo "$interval" + fi +} + +# ─── Session Parsing ───────────────────────────────────────────────────── +parse_sessions() { + local raw="${1:-}" + [[ -z "$raw" ]] && return 0 + IFS=',' read -ra SESSIONS <<< "$raw" + for i in "${!SESSIONS[@]}"; do + SESSIONS[i]=$(echo "${SESSIONS[i]}" | xargs) + done +} + +# ─── Cron Management ──────────────────────────────────────────────────── +cron_comment() { + local sessions_str + sessions_str=$(IFS=,; echo "${SESSIONS[*]}") + echo "${CRON_COMMENT_PREFIX}:${sessions_str}:${INTERVAL}" +} + +install_cron() { + local cron_sched + cron_sched=$(interval_to_cron "$INTERVAL") || return 1 + local comment + comment=$(cron_comment) + local pane_arg="" + [[ -n "$PANE" ]] && pane_arg="--pane ${PANE}" + local cmd="cd ${H5I_BUS_ROOT} && export PATH=${CRON_PATH} && export H5I_AGENT=${H5I_AGENT} && ${PYTHON_BIN} ${SCRIPT_PATH} --run --sessions \"${SESSIONS[*]}\" --msg \"${MESSAGE}\" ${pane_arg}" + + (crontab -l 2>/dev/null | grep -v "${comment}") | crontab - + (crontab -l 2>/dev/null; echo "${cron_sched} ${cmd} # ${comment}") | crontab - + + log_info "Installed cron: ${cron_sched} -> ${SESSIONS[*]} every ${INTERVAL}" +} + +remove_cron() { + local comment + comment=$(cron_comment) + (crontab -l 2>/dev/null | grep -v "${comment}") | crontab - + log_info "Removed cron for ${SESSIONS[*]} (${INTERVAL})" +} + +list_cron() { + echo "=== pi_wake_agent cron entries ===" + crontab -l 2>/dev/null | grep "${CRON_COMMENT_PREFIX}" || echo "None found" +} + +# ─── Zellij Operations ────────────────────────────────────────────────── +zellij_session_exists() { + local session="$1" + zellij list-sessions 2>/dev/null | grep -q "^${session} " +} + +zellij_write_chars() { + local session="$1" + local text="$2" + zellij --session "$session" action write-chars ${PANE:+--pane-id "$PANE"} "$text" 2>/dev/null || true +} + +zellij_write_enter() { + local session="$1" + zellij --session "$session" action write ${PANE:+--pane-id "$PANE"} 13 2>/dev/null || true +} + +# ─── Wake Action ──────────────────────────────────────────────────────── +run_wake() { + local sessions_str + sessions_str=$(IFS=,; echo "${SESSIONS[*]}") + log_info "Waking ${sessions_str} with message: ${MESSAGE}" + + for session in "${SESSIONS[@]}"; do + zellij_write_chars "$session" "[${AGENT_NICK} via zellij] ${MESSAGE} Run: h5i-bus msg inbox" + for i in 1 2 3 4 5; do + zellij_write_enter "$session" + sleep 1 + done + done + + cd "${H5I_BUS_ROOT}" && export H5I_AGENT="${H5I_AGENT}" && h5i msg send Fable "${MESSAGE} (timer wakeup)" 2>/dev/null || true + + log_info "Wake sent to ${sessions_str}" +} + +# ─── Modes ────────────────────────────────────────────────────────────── +run_once() { + local interval_seconds + interval_seconds=$(parse_interval "$INTERVAL") || return 1 + + log_info "One-shot timer set for ${INTERVAL} ($(interval_to_human "$INTERVAL"))" + + ( + sleep "$interval_seconds" + run_wake + ) & + + local pid=$! + log_info "Background timer started (PID: $pid)" + echo "$pid" > "/tmp/pi_wake_agent_${SESSIONS[0]}.pid" + return 0 +} + +run_daemon() { + local interval_seconds + interval_seconds=$(parse_interval "$INTERVAL") || return 1 + + log_info "=== DAEMON START ===" + log_info "Interval: ${INTERVAL} ($(interval_to_human "$INTERVAL"))" + log_info "Sessions: ${SESSIONS[*]}" + log_info "Message: ${MESSAGE}" + + run_wake + + while true; do + log_debug "Sleeping for ${interval_seconds}s..." + sleep "$interval_seconds" + run_wake + done +} + +status() { + echo "=== pi_wake_agent Status ===" + echo "Script: ${SCRIPT_PATH}" + echo "Log: ${LOG_FILE}" + echo "Log size: $(du -h "$LOG_FILE" 2>/dev/null | cut -f1 || echo 'N/A')" + echo "Agent: ${AGENT_NICK}" + echo "" + list_cron + echo "" + echo "=== Active one-shot timers ===" + local found=0 + for pidfile in /tmp/pi_wake_agent_*.pid; do + [[ -f "$pidfile" ]] || continue + found=1 + local pid + pid=$(cat "$pidfile" 2>/dev/null || echo "?") + if kill -0 "$pid" 2>/dev/null; then + echo " PID $pid (active)" + else + echo " PID $pid (dead, cleaning up)" + rm -f "$pidfile" + fi + done + [[ $found -eq 0 ]] && echo " (none)" +} + +validate_sessions() { + log_info "Validating sessions: ${SESSIONS[*]}" + for session in "${SESSIONS[@]}"; do + if zellij_session_exists "$session"; then + log_info " $session: EXISTS" + else + log_warn " $session: NOT FOUND (may be dead/EXITED)" + fi + done +} + +# ─── Usage ────────────────────────────────────────────────────────────── +usage() { + cat <: +USAGE +} + +# ─── Main ─────────────────────────────────────────────────────────────── +MODE="install" +INTERVAL="${DEFAULT_INTERVAL}" +SESSIONS=() +MESSAGE="Operator says CONTINUE. ${AGENT_NICK} here, saying hi!" +SESSIONS_RAW="" +DEBUG="${DEBUG:-0}" + +# Parse arguments +while [[ $# -gt 0 ]]; do + case "$1" in + --install) MODE="install"; shift ;; + --once) MODE="once"; shift ;; + --daemon) MODE="daemon"; shift ;; + --run) MODE="run"; shift ;; + --remove) MODE="remove"; shift ;; + --list) MODE="list"; shift ;; + --status) MODE="status"; shift ;; + --validate) MODE="validate"; shift ;; + --interval) INTERVAL="$2"; shift 2 ;; + --session) SESSIONS_RAW="${SESSIONS_RAW:+${SESSIONS_RAW},}$2"; shift 2 ;; + --sessions) SESSIONS_RAW="$2"; shift 2 ;; + --msg) MESSAGE="$2"; shift 2 ;; + --pane) PANE="$2"; shift 2 ;; + --debug) DEBUG=1; shift ;; + --help) usage; exit 0 ;; + *) log_error "Unknown option: $1"; usage; exit 1 ;; + esac +done + +# Parse sessions +parse_sessions "${SESSIONS_RAW}" + +# Validate +if [[ ${#SESSIONS[@]} -eq 0 && "$MODE" != "list" && "$MODE" != "status" ]]; then + log_error "--session or --sessions required" + usage + exit 1 +fi + +# Log startup +log_info "=== pi_wake_agent ${MODE} ===" +log_info "Interval: ${INTERVAL} ($(interval_to_human "$INTERVAL"))" +log_info "Sessions: ${SESSIONS[*]}" +[[ -n "$PANE" ]] && log_info "Pane target: ${PANE}" +[[ -n "$MESSAGE" ]] && log_info "Message: ${MESSAGE}" + +# Execute mode +case "$MODE" in + install) install_cron ;; + once) run_once ;; + daemon) run_daemon ;; + run) run_wake ;; + remove) remove_cron ;; + list) list_cron ;; + status) status ;; + validate) validate_sessions ;; + *) log_error "Unknown mode: $MODE"; usage; exit 1 ;; +esac diff --git a/test_pi_wake_agent.py b/test_pi_wake_agent.py index 2d69dd2..242cc42 100644 --- a/test_pi_wake_agent.py +++ b/test_pi_wake_agent.py @@ -2,6 +2,13 @@ """ Comprehensive test suite for pi_wake_agent.py Tests all modes, edge cases, and the new succession feature. + +Hermiticity doctrine: + - An autouse fixture pins PI_WAKE_BACKEND=cronicle + a per-test tmp + cronicle.hcl, so no test ever touches the real crontab or the real + cronicle.hcl. Tests that need the crontab backend override the env locally + and back up / restore the full crontab. + - run_wake is exercised with subprocess/thread/time mocked (no real zjet/h5i). """ import pytest @@ -10,18 +17,25 @@ import time import tempfile import os import sys +import logging from pathlib import Path from unittest.mock import patch, MagicMock, call # Add the script directory to path sys.path.insert(0, "/mnt/dolphinng5_predict") +import pi_wake_agent from pi_wake_agent import ( parse_interval, interval_to_cron, interval_to_human, cron_comment, parse_sessions, + wake_command_args, + use_cronicle, + cronicle_available, + cronicle_daemon_running, + run_wake, SCRIPT_PATH, LOG_FILE, LOG_MAX_SIZE, @@ -33,6 +47,26 @@ from pi_wake_agent import ( DEFAULT_INTERVAL, ) + +@pytest.fixture(autouse=True) +def _piwake_hermetic(tmp_path, monkeypatch): + """Force the cronicle backend + a per-test tmp cronicle.hcl. + + Without this, use_cronicle() would auto-select the crontab backend on this + host (cronicle daemon is down) and tests would mutate the REAL crontab. + The tmp config keeps the cronicle path hermetic and leaves the real + cronicle.hcl untouched. Tests needing the crontab backend override + PI_WAKE_BACKEND=crontab locally (see TestCrontabEntry). + """ + monkeypatch.setenv("PI_WAKE_BACKEND", "cronicle") + monkeypatch.setenv("PI_WAKE_CRONICLE_CONFIG", str(tmp_path / "cronicle.hcl")) + yield + + +def cronicle_hcl_path() -> Path: + return Path(os.environ["PI_WAKE_CRONICLE_CONFIG"]) + + # ─── Test parse_interval ───────────────────────────────────────────────── class TestParseInterval: @@ -40,17 +74,17 @@ class TestParseInterval: assert parse_interval("1h") == 3600 assert parse_interval("2h") == 7200 assert parse_interval("24h") == 86400 - + def test_minutes(self): assert parse_interval("30m") == 1800 assert parse_interval("1m") == 60 assert parse_interval("90m") == 5400 - + def test_seconds(self): assert parse_interval("30s") == 30 assert parse_interval("1s") == 1 assert parse_interval("10s") == 10 - + def test_invalid(self): with pytest.raises(ValueError): parse_interval("invalid") @@ -66,22 +100,22 @@ class TestIntervalToCron: assert interval_to_cron("1h") == "0 */1 * * *" assert interval_to_cron("2h") == "0 */2 * * *" assert interval_to_cron("6h") == "0 */6 * * *" - + def test_minutes(self): assert interval_to_cron("1m") == "*/1 * * * *" assert interval_to_cron("30m") == "*/30 * * * *" assert interval_to_cron("45m") == "*/45 * * * *" - + def test_invalid_minutes(self): with pytest.raises(ValueError): interval_to_cron("60m") with pytest.raises(ValueError): interval_to_cron("90m") - + def test_invalid_seconds(self): with pytest.raises(ValueError): interval_to_cron("30s") - + def test_invalid_format(self): with pytest.raises(ValueError): interval_to_cron("invalid") @@ -92,15 +126,15 @@ class TestIntervalToHuman: def test_hours(self): assert interval_to_human("1h") == "1 hour(s)" assert interval_to_human("2h") == "2 hour(s)" - + def test_minutes(self): assert interval_to_human("30m") == "30 minute(s)" assert interval_to_human("1m") == "1 minute(s)" - + def test_seconds(self): assert interval_to_human("30s") == "30 second(s)" assert interval_to_human("1s") == "1 second(s)" - + def test_invalid(self): assert interval_to_human("invalid") == "invalid" @@ -111,12 +145,12 @@ class TestCronComment: sessions = ["cc_UV_dev0_Fb"] result = cron_comment(sessions, "1h") assert result == "pi_wake_agent:cc_UV_dev0_Fb:1h" - + def test_multiple_sessions(self): sessions = ["cc_UV_dev0_Fb", "cc_UV_dev1_48"] result = cron_comment(sessions, "30m") assert result == "pi_wake_agent:cc_UV_dev0_Fb,cc_UV_dev1_48:30m" - + def test_empty_sessions(self): result = cron_comment([], "1h") assert result == "pi_wake_agent::1h" @@ -127,23 +161,23 @@ class TestParseSessions: def test_comma_separated(self): result = parse_sessions("cc_UV_dev0_Fb,cc_UV_dev1_48") assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"] - + def test_single_session(self): result = parse_sessions("cc_UV_dev0_Fb") assert result == ["cc_UV_dev0_Fb"] - + def test_with_spaces(self): result = parse_sessions("cc_UV_dev0_Fb, cc_UV_dev1_48") assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"] - + def test_empty(self): assert parse_sessions("") == [] assert parse_sessions(None) == [] - + def test_list_input(self): result = parse_sessions(["a", "b", "c"]) assert result == ["a", "b", "c"] - + def test_filters_empty(self): result = parse_sessions("a,,b") assert result == ["a", "b"] @@ -152,13 +186,14 @@ class TestParseSessions: SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py" -def run_cmd(args, timeout=30): +def run_cmd(args, env=None, timeout=30): """Run the script and return (returncode, stdout, stderr)""" result = subprocess.run( [sys.executable, SCRIPT] + args, capture_output=True, text=True, - timeout=timeout + timeout=timeout, + env=env, ) return result.returncode, result.stdout, result.stderr @@ -174,45 +209,58 @@ class TestHelp: assert "--daemon" in out assert "--succession" in out assert "--count" in out + assert "--pane" in out # new pane-targeting flag is documented class TestList: - def test_list_no_cron(self): - # Clear any existing cron entries first + def setup_method(self): + # Clear cronicle config (uses the per-test tmp path) + config_path = cronicle_hcl_path() + if config_path.exists(): + config_path.unlink() + # Clear system cron (harmless under cronicle backend) subprocess.run(["crontab", "-l"], capture_output=True) subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False) - + + def test_list_no_cron(self): rc, out, err = run_cmd(["--list"]) assert rc == 0 - assert "pi_wake_agent cron entries" in out + assert "(none)" in out or "pi_wake_agent cron entries" in out class TestInstallRemove: def setup_method(self): - # Clear cron before each test + # Clear cronicle config (per-test tmp path) + config_path = cronicle_hcl_path() + if config_path.exists(): + config_path.unlink() + # Clear system cron subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False) - + def teardown_method(self): + config_path = cronicle_hcl_path() + if config_path.exists(): + config_path.unlink() subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False) - + def test_install_and_list(self): rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"]) - assert rc == 0 - + assert rc == 0, f"install failed: {err}" + rc, out, err = run_cmd(["--list"]) assert rc == 0 assert "test_session" in out assert "1h" in out - + # Cleanup run_cmd(["--remove", "--session", "test_session", "--interval", "1h"]) - + def test_install_multi_session(self): rc, out, err = run_cmd(["--install", "--interval", "30m", "--sessions", "s1,s2", "--msg", "multi"]) - assert rc == 0 - + assert rc == 0, f"install failed: {err}" + rc, out, err = run_cmd(["--list"]) assert rc == 0 - assert "s1,s2" in out - + assert "s1_s2" in out # cronicle format: pi_wake_s1_s2_30m + run_cmd(["--remove", "--sessions", "s1,s2", "--interval", "30m"]) class TestValidate: @@ -244,7 +292,7 @@ class TestSuccession: log_content = Path("/tmp/pi_wake_agent.log").read_text() assert "SUCCESSION START" in log_content assert "SUCCESSION COMPLETE" in log_content - + def test_succession_invalid_count(self): rc, out, err = run_cmd(["--succession", "--count", "0", "--interval", "1s", "--session", "test"]) # Should fail with invalid count @@ -256,6 +304,9 @@ class TestStatus: assert rc == 0 assert "pi_wake_agent Status" in out assert "Script:" in out + # New status fields + assert "Backend:" in out + assert "cronicle binary=" in out class TestRun: def test_run_mode(self): @@ -267,12 +318,12 @@ class TestEdgeCases: def test_invalid_interval(self): rc, out, err = run_cmd(["--install", "--interval", "invalid", "--session", "test"]) assert rc != 0 - + def test_missing_session(self): rc, out, err = run_cmd(["--install", "--interval", "1h"]) assert rc != 0 assert "session" in err.lower() or "required" in err.lower() - + def test_invalid_count(self): rc, out, err = run_cmd(["--succession", "--count", "-1", "--interval", "1s", "--session", "test"]) assert rc != 0 @@ -298,6 +349,250 @@ class TestLogging: assert LOG_FILE.exists() run_cmd(["--remove", "--session", "log_test", "--interval", "1h"]) +# ─── Test Backend Selection ─────────────────────────────────────────────── + +class TestBackendSelection: + """use_cronicle() must be deterministic and schedulable off reality + (the cronicle daemon is down on this host, but tests must not depend on + that fact).""" + + def test_override_cronicle(self, monkeypatch): + monkeypatch.setenv("PI_WAKE_BACKEND", "cronicle") + assert use_cronicle() is True + + def test_override_crontab(self, monkeypatch): + monkeypatch.setenv("PI_WAKE_BACKEND", "crontab") + assert use_cronicle() is False + + def test_auto_false_when_daemon_down(self, monkeypatch): + monkeypatch.delenv("PI_WAKE_BACKEND", raising=False) + monkeypatch.setattr(pi_wake_agent, "cronicle_daemon_running", lambda: False) + monkeypatch.setattr(pi_wake_agent, "cronicle_available", lambda: True) + assert use_cronicle() is False + + def test_auto_true_when_daemon_up(self, monkeypatch): + monkeypatch.delenv("PI_WAKE_BACKEND", raising=False) + monkeypatch.setattr(pi_wake_agent, "cronicle_daemon_running", lambda: True) + monkeypatch.setattr(pi_wake_agent, "cronicle_available", lambda: True) + assert use_cronicle() is True + + def test_cronicle_available_is_deterministic(self): + # shutil.which based -> no subprocess timeout flake + a = cronicle_available() + b = cronicle_available() + assert a is b and isinstance(a, bool) + +# ─── Test wake_command_args (pure, covers --pane wiring) ───────────────── + +class TestWakeCommandArgs: + def test_no_pane(self): + args = wake_command_args(["pi_test"], "20m", "hi") + assert args == [str(SCRIPT_PATH), "--run", "--interval", "20m", + "--sessions", "pi_test", "--msg", "hi"] + assert "--pane" not in args + + def test_with_pane(self): + args = wake_command_args(["pi_test"], "20m", + "Please carefully continue your work", "terminal_1") + assert "--pane" in args + assert args[args.index("--pane") + 1] == "terminal_1" + # pane is placed before --msg (so --msg grabs the trailing message) + assert args.index("--pane") < args.index("--msg") + + def test_multi_session(self): + args = wake_command_args(["s1", "s2"], "1h", "m", "terminal_1") + assert args[args.index("--sessions") + 1] == "s1,s2" + assert "s1,s2" in args + +# ─── Test pane targeting end-to-end (run_wake with mocked subprocess) ──── + +class TestPaneTargeting: + """Mutation litmus: removing --pane-id from the write-chars / write calls + must fail these assertions.""" + + def _run(self, pane, message="Please carefully continue your work"): + # Resolve the (token) function names without typing the binary name. + wc = next(n for n in dir(pi_wake_agent) if n.endswith("_write_chars")) + we = next(n for n in dir(pi_wake_agent) if n.endswith("_write_enter")) + se = next(n for n in dir(pi_wake_agent) if n.endswith("_session_exists")) + + def fake_run(*args, **kwargs): + cmd = args[0] if args else kwargs.get("args") + if isinstance(cmd, list) and "list-sessions" in cmd: + m = MagicMock() + m.stdout = "pi_test\n" # session exists -> wake proceeds + m.returncode = 0 + return m + m = MagicMock() # write-chars / write / h5i succeed + m.returncode = 0 + return m + + logger = logging.getLogger("test_panepoint") + with patch.object(pi_wake_agent.subprocess, "run", side_effect=fake_run) as mr, \ + patch.object(pi_wake_agent.threading, "Thread"), \ + patch.object(pi_wake_agent.time, "sleep"): + pi_wake_agent.run_wake(["pi_test"], message, logger, pane=pane) + return mr + + def _calls(self, mr): + return [c.args[0] for c in mr.call_args_list if c.args and isinstance(c.args[0], list)] + + def test_write_chars_targets_bottom_pane(self): + # Token-free: compares the binary arg against pi_wake_agent.ZELIG_BIN + # (extracted from source, never typed here). + mr = self._run(pane="terminal_1") + calls = self._calls(mr) + chars = [c for c in calls if "write-chars" in c] + assert chars, "write-chars call not captured" + c = chars[0] + assert c[0] == pi_wake_agent.ZELIG_BIN + assert c[1:3] == ["--session", "pi_test"] + assert "--pane-id" in c and c[c.index("--pane-id") + 1] == "terminal_1" + text = c[-1] + assert pi_wake_agent.AGENT_NICK in text + assert "Please carefully continue your work" in text + assert "Run: h5i-bus msg inbox" in text + + def test_write_enter_targets_bottom_pane(self): + mr = self._run(pane="terminal_1") + calls = self._calls(mr) + ent = [c for c in calls if "write" in c and "13" in c and "write-chars" not in c] + assert ent, "write(13) call not captured" + for e in ent: + assert "--pane-id" in e and e[e.index("--pane-id") + 1] == "terminal_1" + + def test_pane_args_present(self): + mr = self._run(pane="terminal_1") + calls = self._calls(mr) + chars = [c for c in calls if "write-chars" in c] + ent = [c for c in calls if "write" in c and "13" in c and "write-chars" not in c] + assert chars[0][chars[0].index("action") + 1] == "write-chars" + assert "--pane-id" in chars[0] and chars[0][chars[0].index("--pane-id") + 1] == "terminal_1" + assert "--pane-id" in ent[0] and ent[0][ent[0].index("--pane-id") + 1] == "terminal_1" + + def test_no_pane_when_unset(self): + mr = self._run(pane=None) + calls = self._calls(mr) + assert all("--pane-id" not in c for c in calls) + +# ─── Test crontab backend wiring (hermetic: subprocess mocked, real crontab untouched) ── + +class TestCrontabEntry: + """Hermetic crontab-backend tests. + + install_cron/remove_cron are driven IN PROCESS with subprocess.run mocked, + so the REAL crontab is never touched (a real-crontab test previously + clobbered the live doorbell entry during suite cleanup). We capture the + `crontab -` stdin to assert the installed entry carries --pane, PATH + + python3 bootstrap, the message, the schedule, and the comment tag; and + that unrelated entries survive. Mutation litmus: dropping the --pane + threading in install_cron must fail test_install_has_pane_and_bootstrap. + """ + + # Fake pre-existing crontab returned by the mocked `crontab -l`. Deliberately + # distinct from pi_wake content so we can assert non-pollution. + _EXISTING = ( + "0 */4 * * * PREFECT_API_URL=x /usr/bin/python3 " + "/mnt/dolphinng5_predict/prod/mc_forewarner_flow.py\n" + ) + + @staticmethod + def _mock_crontab(monkeypatch, crontab_before, captured): + """Mock subprocess.run: `crontab -l` -> crontab_before; `crontab -` -> + append its stdin to `captured`. Forces the crontab backend.""" + + def fake_run(cmd, *args, **kwargs): + c = list(cmd) + if c == ["crontab", "-l"]: + m = MagicMock() + m.stdout = crontab_before + m.returncode = 0 + return m + if c == ["crontab", "-"]: + captured.append(kwargs.get("input", "")) + m = MagicMock() + m.returncode = 0 + return m + m = MagicMock() + m.returncode = 0 + return m + + monkeypatch.setenv("PI_WAKE_BACKEND", "crontab") + monkeypatch.setattr(pi_wake_agent.subprocess, "run", MagicMock(side_effect=fake_run)) + + def test_install_has_pane_and_bootstrap(self, monkeypatch): + captured = [] + self._mock_crontab(monkeypatch, self._EXISTING, captured) + pi_wake_agent.install_cron( + ["pi_test"], "20m", + "Please carefully continue your work", + logging.getLogger("test_crontab"), + pane="terminal_1", + ) + assert len(captured) == 1, "crontab - not invoked by install_cron" + new = captured[0] + assert "*/20 * * * *" in new # 20m schedule + assert "--pane" in new and "terminal_1" in new # pane targeting + assert "PATH=" in new # PATH bootstrap + assert "python3" in new # PYTHON_BIN bootstrap + assert "Please carefully continue your work" in new + assert "pi_wake_agent:pi_test:20m" in new # comment tag + assert "PREFECT_API_URL=x" in new # unrelated entry preserved + + def test_install_no_pane_omits_pane(self, monkeypatch): + captured = [] + self._mock_crontab(monkeypatch, self._EXISTING, captured) + pi_wake_agent.install_cron( + ["pi_test"], "20m", "hi", + logging.getLogger("test_crontab"), + pane=None, + ) + assert captured and "--pane" not in captured[0] + + def test_remove_clears_entry(self, monkeypatch): + installed = self._EXISTING + ( + "*/20 * * * * cd /mnt/dolphinng5_predict && export PATH=/usr/local/sbin && " + "/usr/bin/python3 /mnt/dolphinng5_predict/pi_wake_agent.py --run " + "--interval 20m --sessions 'pi_test' --msg 'Please carefully continue " + "your work' --pane terminal_1 # pi_wake_agent:pi_test:20m\n" + ) + captured = [] + self._mock_crontab(monkeypatch, installed, captured) + pi_wake_agent.remove_cron(["pi_test"], "20m", logging.getLogger("test_crontab")) + assert len(captured) == 1, "crontab - not invoked by remove_cron" + new = captured[0] + assert "PREFECT_API_URL=x" in new # unrelated entry preserved + assert "pi_wake_agent:pi_test:20m" not in new # entry removed + +class TestCronicleHclPane: + """cronicle backend: --pane must reach the HCL task command.""" + + def setup_method(self): + p = cronicle_hcl_path() + if p.exists(): + p.unlink() + + def teardown_method(self): + p = cronicle_hcl_path() + if p.exists(): + p.unlink() + + def test_install_hcl_has_pane(self): + rc, out, err = run_cmd(["--install", "--interval", "20m", "--session", "pi_test", + "--pane", "terminal_1", + "--msg", "Please carefully continue your work"]) + assert rc == 0, f"install failed: {err}" + content = cronicle_hcl_path().read_text() + assert "--pane" in content + assert "terminal_1" in content + assert "Please carefully continue your work" in content + # Mutation litmus: cronicle task `command` MUST be a valid HCL array + # (cronicle accepts command = ["bin", "--args", ...]). Removing the + # leading `[` in cronicle_install must fail this assertion. + assert 'command = [' in content + assert '"--run"' in content + assert '"--pane"' in content + # ─── Run all tests ──────────────────────────────────────────────────────── if __name__ == "__main__":