From 725b0cc730a54e58dfb0cb97d64727ca8ad76c72 Mon Sep 17 00:00:00 2001 From: pool4 Date: Tue, 13 Oct 2026 03:27:24 +0200 Subject: [PATCH] Add Pi wake-agent watcher: pi_wake_agent.py + spec + tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Dedicated repo for the DOLPHIN/pi wake timer (§3.4 AGENT_ONBOARDING_GENERAL_DOCUMENT.md). Gathers only the waker artefacts: the timer script, its tool spec, and the 38 unit tests. Committed as pool4. --- PI_WAKE_AGENT_TOOL.md | 165 ++++++ README.md | 33 ++ pi_wake_agent.py | 970 +++++++++++++++++++++++++++++++++++ test_pi_wake_agent.py | 1134 +++++++++++++++++++++++++++++++++++++++++ 4 files changed, 2302 insertions(+) create mode 100755 PI_WAKE_AGENT_TOOL.md create mode 100644 README.md create mode 100755 pi_wake_agent.py create mode 100755 test_pi_wake_agent.py diff --git a/PI_WAKE_AGENT_TOOL.md b/PI_WAKE_AGENT_TOOL.md new file mode 100755 index 0000000..4205d3e --- /dev/null +++ b/PI_WAKE_AGENT_TOOL.md @@ -0,0 +1,165 @@ +# pi_wake_agent.py — Multi-Agent Wake Timer + +**Location:** `/mnt/dolphinng5_predict/pi_wake_agent.py` +**Branch:** `tools/pi_wake_agent` +**Status:** v1.0 — 38 tests passing + +--- + +## Overview + +Reusable multi-agent wake-up timer with self-cron/daemon/succession modes. Designed for the DOLPHIN fleet (pi_nvnemo, cmd, mimo, codex, etc.) to send doorbell injections via zellij and durable messages via h5i bus. + +--- + +## Installation + +```bash +cd /mnt/dolphinng5_predict +python3 pi_wake_agent.py --install --interval 30m --session pi_test --msg "Wake up!" +``` + +--- + +## Modes + +| Mode | Flag | Description | +|------|------|-------------| +| Install cron | `--install` | Recurring wake via system cron | +| One-shot | `--once` | Single wake after interval (no cron) | +| Daemon | `--daemon` | Long-lived process, no cron | +| Succession | `--succession` | Run N times at interval, then self-clean | +| Run (internal) | `--run` | Called by cron, executes wake | +| Remove | `--remove` | Remove cron entry | +| List | `--list` | Show active cron entries | +| Status | `--status` | Show cron + one-shot timers | +| Validate | `--validate` | Check zellij sessions exist | + +--- + +## Options + +| Option | Description | +|--------|-------------| +| `--interval DURATION` | Default: 1h. Formats: 30m, 1h, 90m, 2h, 10s | +| `--session SESSION` | Zellij session name (repeatable) | +| `--sessions "A,B,C"` | Comma-separated list | +| `--msg "MESSAGE"` | Wake message (default: "Operator says CONTINUE. Pi here!") | +| `--count N` | Number of runs for `--succession` | +| `--dry-run` | Show what would be done without executing | +| `--help` | Show help | + +--- + +## Examples + +```bash +# Recurring 1-hour wake for one session +pi_wake_agent.py --install --interval 1h --session cc_UV_dev0_Fb + +# Multi-session 30-minute wake +pi_wake_agent.py --install --interval 30m --sessions "cc_UV_dev0_Fb,cc_UV_dev1_48" --msg "Wake up!" + +# One-shot in 2 hours +pi_wake_agent.py --once --interval 2h --session cc_UV_dev0_Fb --msg "Time's up!" + +# Run 3 times at 1-hour intervals, then self-clean +pi_wake_agent.py --succession --count 3 --interval 1h --session cc_UV_dev0_Fb --msg "Scheduled wake" + +# Daemon mode (long-lived, no cron) +pi_wake_agent.py --daemon --interval 1h --session cc_UV_dev0_Fb + +# Remove timer +pi_wake_agent.py --remove --session cc_UV_dev0_Fb --interval 1h + +# List / status +pi_wake_agent.py --list +pi_wake_agent.py --status + +# Validate sessions exist +pi_wake_agent.py --validate --session cc_UV_dev0_Fb +``` + +--- + +## Key Features + +### Non-Blocking h5i Bus +Messages sent to Fable via `h5i msg send` are **fire-and-forget**: +- Runs in background thread +- 5-second timeout +- Silently ignores failures +- Never blocks the wake cycle + +### Self-Cleaning Succession +```bash +pi_wake_agent.py --succession --count 3 --interval 1h --session S +``` +Runs exactly 3 times at 1-hour intervals, then removes its own cron entry. + +### Multi-Session +```bash +# Repeatable --session +pi_wake_agent.py --install --interval 1h --session s1 --session s2 + +# Comma-separated --sessions +pi_wake_agent.py --install --interval 1h --sessions "s1,s2,s3" +``` + +### Daemon Mode +Runs indefinitely as a long-lived process (no cron needed): +```bash +pi_wake_agent.py --daemon --interval 1h --session S +``` + +--- + +## h5i Bus Protocol Compliance + +Every wake injection: +1. Identifies as `[pi_nvnemo via zellij]` +2. Includes `Run: h5i-bus msg inbox` directive +3. Sends 5× ENTER keypresses (1s delay) for reliable submission +4. Sends parallel h5i bus message for durability + +Per [AGENT_TERMINAL_DIRECT_INTERVENTION_PROCEDURES.md](AGENT_TERMINAL_DIRECT_INTERVENTION_PROCEDURES.md). + +--- + +## Testing + +```bash +cd /mnt/dolphinng5_predict +python3 -m pytest test_pi_wake_agent.py -v +``` +**38 tests passing** covering: +- Unit tests: interval parsing, cron comments, session parsing +- Integration: install/remove/list, one-shot, succession, run mode +- Edge cases: invalid intervals, missing sessions, invalid counts + +--- + +## Cron Entry Format + +``` +*/30 * * * * cd /mnt/dolphinng5_predict && export H5I_AGENT=pi_nvnemo && /mnt/dolphinng5_predict/pi_wake_agent.py --run --sessions 'pi_test' --msg '...' # pi_wake_agent:pi_test:30m +``` + +Comment format: `pi_wake_agent::` + +--- + +## Logs + +- **File:** `/tmp/pi_wake_agent.log` +- **Rotation:** 10MB max, 5 files +- **Format:** `[YYYY-MM-DD HH:MM:SS] [LEVEL] message` + +--- + +## Related Files + +- `/mnt/dolphinng5_predict/pi_wake_agent.py` — Main script +- `/mnt/dolphinng5_predict/test_pi_wake_agent.py` — Test suite (38 tests) +- `/mnt/dolphinng5_predict/prod/docs/AGENT_TERMINAL_DIRECT_INTERVENTION_PROCEDURES.md` — Protocol +- Branch: `tools/pi_wake_agent` diff --git a/README.md b/README.md new file mode 100644 index 0000000..18c8cb1 --- /dev/null +++ b/README.md @@ -0,0 +1,33 @@ +# pi_wake_agent — DOLPHIN fleet wake timer (Pi) + +Reusable multi-agent wake timer. Wakes zellij sessions (Dolphin fleet: +pi_nvnemo, cmd, mimo, codex, ...) on a recurring / one-shot / daemon / +succession schedule, delivering doorbell injections via zellij + durable +messages via the h5i bus. + +## Files (the watcher) +- `pi_wake_agent.py` — the wake timer (source: `/mnt/dolphinng5_predict/pi_wake_agent.py`, + branch `tools/pi_wake_agent`). +- `PI_WAKE_AGENT_TOOL.md` — tool spec + usage: `--install/--once/--daemon/` + `--succession --count N --remove/--list/--status/--validate`; intervals + (`30m`, `1h`, `10s`, ...); `--session`/`--sessions "A,B"`; `--msg`. +- `test_pi_wake_agent.py` — 38 unit tests (interval parsing, cron comments, session parsing). + +## Example — wake a session every 15 min (monitor .68, stop+diag on fault) +python3 pi_wake_agent.py --install --interval 15m --session \ + --msg "RECHECK .68: stale_snapshot/preemption_refusal/TP-arm vs MAX_HOLD; stop+diag if fault" + +Manage: `--list`, `--status`, re-enable with `--install`, cancel with `--remove`. + +## Notes +- The 20 m doorbell is currently DISABLED by operator order (crontab entry + `pi_wake_agent:pi_test:20m` removed; backup at `/root/.cache/crontab/crontab.bak`). +- It is a wake/ping timer, NOT an autonomous monitor/stop-and-diag agent: the + woken agent performs the checks. For autonomous stop+diag on fault, pair it + with a host-local cron guard (`/etc/cron.d/.68_guard` + `p68_guard.sh`) which + actually stops the unit + dumps diagnostics. +- Cross-ref: §3.4 "Pi Wake Agent (multi-agent wake timer)" in + `AGENT_ONBOARDING_GENERAL_DOCUMENT.md`. + +## Committer +`pool4 ` diff --git a/pi_wake_agent.py b/pi_wake_agent.py new file mode 100755 index 0000000..1a74b36 --- /dev/null +++ b/pi_wake_agent.py @@ -0,0 +1,970 @@ +#!/usr/bin/env python3 +""" +pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession + +Usage: + pi_wake_agent.py --install --interval 1h --session cc_UV_dev0_Fb --msg "Operator says CONTINUE. Pi here!" + pi_wake_agent.py --once --interval 2h --session cc_UV_dev0_Fb --msg "Time's up!" + pi_wake_agent.py --daemon --interval 1h --session cc_UV_dev0_Fb + pi_wake_agent.py --succession --count 3 --interval 1h --session cc_UV_dev0_Fb --msg "Scheduled wake" + pi_wake_agent.py --remove --session cc_UV_dev0_Fb --interval 1h + pi_wake_agent.py --list + pi_wake_agent.py --status + pi_wake_agent.py --validate --session cc_UV_dev0_Fb +""" + +import argparse +import logging +import os +import re +import shutil +import shlex +import signal +import subprocess +import sys +import threading +import time +from pathlib import Path +from typing import List, Optional, Tuple + +# ─── Constants ──────────────────────────────────────────────────────────── +SCRIPT_PATH = Path(__file__).resolve() +LOG_FILE = Path("/tmp/pi_wake_agent.log") +LOG_MAX_SIZE = 10 * 1024 * 1024 # 10 MB +LOG_MAX_FILES = 5 +CRON_COMMENT_PREFIX = "pi_wake_agent" +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' +# ── Context-occupancy steering (pi status bar: '%/1.0M') ────────────────────── +# Per AGENTS.md §4.2: >70%% degradation risk, >85%% 'ship partial NOW', +# >95%% emergency (pi runs out of room to compact -> lockup). A doorbell that +# sees critical context appends a /compact heads-up to the steer. +CONTEXT_WARN_PCT = 85 +CONTEXT_CRIT_PCT = 95 +CONTEXT_RE = re.compile(r"(\d+(?:\.\d+)?)%/1\.0M") + + +def setup_logging(debug: bool = False) -> logging.Logger: + log_rotate() + logger = logging.getLogger("pi_wake_agent") + logger.setLevel(logging.DEBUG if debug else logging.INFO) + + fh = logging.FileHandler(LOG_FILE) + fh.setLevel(logging.DEBUG) + fh.setFormatter(logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) + logger.addHandler(fh) + + ch = logging.StreamHandler(sys.stderr) + ch.setLevel(logging.WARNING) + ch.setFormatter(logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")) + logger.addHandler(ch) + + return logger + + +def log_rotate() -> None: + if not LOG_FILE.exists(): + return + size = LOG_FILE.stat().st_size + if size < LOG_MAX_SIZE: + return + for i in range(LOG_MAX_FILES - 1, 0, -1): + src = LOG_FILE.with_suffix(f".log.{i}") if i > 1 else LOG_FILE.with_suffix(".log.1") + if src.exists(): + dst = LOG_FILE.with_suffix(f".log.{i + 1}") + src.rename(dst) + LOG_FILE.rename(LOG_FILE.with_suffix(".log.1")) + + +# ─── Interval Parsing ──────────────────────────────────────────────────── +def parse_interval(interval: str) -> int: + """Parse interval string (e.g., '1h', '30m', '90m', '2h', '10s') to seconds.""" + match = re.match(r"^(\d+)([hms])$", interval) + if not match: + raise ValueError(f"Invalid interval format '{interval}'. Use like 1h, 30m, 90m, 2h, 10s") + value, unit = int(match.group(1)), match.group(2) + if unit == "h": + return value * 3600 + elif unit == "m": + return value * 60 + elif unit == "s": + return value + raise ValueError(f"Unknown unit: {unit}") + + +def interval_to_cron(interval: str) -> str: + """Convert interval to cron schedule.""" + match = re.match(r"^(\d+)([hm])$", interval) + if not match: + raise ValueError(f"Cron only supports minutes/hours intervals: {interval}") + value, unit = int(match.group(1)), match.group(2) + if unit == "h": + return f"0 */{value} * * *" + elif unit == "m": + if value >= 60: + raise ValueError(f"For minutes >= 60, use hours (e.g., 1h not 60m)") + return f"*/{value} * * * *" + raise ValueError(f"Cron only supports minutes/hours intervals: {interval}") + + +def interval_to_human(interval: str) -> str: + match = re.match(r"^(\d+)([hms])$", interval) + if not match: + return interval + value, unit = match.group(1), match.group(2) + if unit == "h": + return f"{value} hour(s)" + elif unit == "m": + return f"{value} minute(s)" + elif unit == "s": + return f"{value} second(s)" + return interval + + +# ─── 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 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: Optional[str], message: str, pane: Optional[str] = None) -> List[str]: + """Build the argv list for the --run wake command (shared by both backends). + + ``interval`` is the recurring cadence used by cron/crontab entries; pass + ``None`` for a one-off --run invocation (e.g. --at, which is time-of-day + triggered, not interval driven) so --interval is omitted. --pane targets a + specific zjet pane (e.g. the bottom Soundcheck pane). + """ + args = [str(SCRIPT_PATH), "--run"] + if interval is not None: + args += ["--interval", interval] + args += ["--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) + + # 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 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 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 '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})") + + +# ─── One-off Time-of-Day Scheduling (--at) ─────────────────────────────── +AT_TIME_RE = re.compile(r"^(\d{1,2}):(\d{2})$|^(\d{4})$") + + +def parse_at_time(spec: str) -> Tuple[str, str, str]: + """Parse a one-off wake time-of-day. + + Accepts 'HH:MM' (e.g. '01:50', '1:05') or 'HHMM' (e.g. '0150'). Returns + (hhmm, minute, hour), each zero-padded, e.g. ('0150', '50', '01'). Raises + ValueError on an unparseable or out-of-range time. + """ + m = re.match(r"^(\d{1,2}):(\d{2})$", spec) + if m: + hour, minute = int(m.group(1)), int(m.group(2)) + else: + m = re.match(r"^(\d{4})$", spec) + if not m: + raise ValueError(f"--at expects HH:MM or HHMM (e.g. 01:50 or 0150), got {spec!r}") + hour, minute = int(spec[:2]), int(spec[2:]) + if not (0 <= hour <= 23) or not (0 <= minute <= 59): + raise ValueError(f"--at time out of range: {spec!r} (hour 0-23, minute 0-59)") + return f"{hour:02d}{minute:02d}", f"{minute:02d}", f"{hour:02d}" + + +def cron_at_comment(sessions: List[str], hhmm: str) -> str: + """Crontab/crnd tag for a one-off --at entry (distinct from recurring).""" + sessions_str = ",".join(sessions) + return f"{CRON_COMMENT_PREFIX}:{sessions_str}:at:{hhmm}" + + +def cron_at_schedule_name(sessions: List[str], hhmm: str) -> str: + """Cronicle schedule name for a one-off --at entry (starts with 'pi_wake_').""" + return f"pi_wake_{'_'.join(sessions)}_at_{hhmm}" + + +def _at_wake_command(sessions: List[str], message: str, pane: Optional[str] = None) -> str: + """Shell command string for the --run wake invoked by a one-off --at entry. + + Mirrors install_cron's command construction but omits --interval: a one-off + is time-of-day triggered, not interval driven (and --run ignores --interval + anyway, so omitting it is both cleaner and accurate). + """ + sessions_arg = f"'{','.join(sessions)}'" + cmd = f"{PYTHON_BIN} {SCRIPT_PATH} --run --sessions {sessions_arg} --msg {shlex.quote(message)}" + if pane: + cmd += f" --pane {shlex.quote(pane)}" + return cmd + + +def _strip_lines_by_tag(lines: List[str], tag: str) -> List[str]: + """Drop crontab lines whose comment contains `tag` (idempotent install/remove).""" + return [line for line in lines if tag not in line] + + +def build_at_cron_entry(sessions: List[str], hhmm: str, message: str, pane: Optional[str] = None) -> str: + """Build the full crontab line for a one-off, self-cleaning --at wake. + + Layout: + 'min hour * * * cd && export PATH=... && export H5I_AGENT=... && ; (crontab -l | grep -v '' ; echo) | crontab - # ' + The self-strip suffix removes THIS line (matched by its unique 'at:' + tag) after it fires, so the entry is strictly one-off and never clobbers + unrelated entries (the tag is distinct from the recurring + 'pi_wake_agent::' tag). + """ + _, minute, hour = parse_at_time(hhmm) + comment = cron_at_comment(sessions, hhmm) + wake_cmd = _at_wake_command(sessions, message, pane) + self_strip = f"crontab -l 2>/dev/null | grep -v '{comment}'" + cmd = (f"cd {H5I_BUS_ROOT} && export PATH={CRON_PATH} && export H5I_AGENT={H5I_AGENT} && " + f"{wake_cmd} ; ({self_strip} ; echo) | crontab - # {comment}") + return f"{minute} {hour} * * * {cmd}" + + +def crontab_install_at(sessions: List[str], hhmm: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install a one-off, self-cleaning crontab entry at the given time-of-day (crnd backend). + + Idempotent: a pre-existing entry with the same 'at:' tag is replaced + rather than duplicated. Unrelated crontab entries are preserved verbatim. + """ + comment = cron_at_comment(sessions, hhmm) + entry = build_at_cron_entry(sessions, hhmm, message, pane) + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + lines = _strip_lines_by_tag(existing.splitlines(), comment) + lines.append(entry) + subprocess.run(["crontab", "-"], input="\n".join(lines) + "\n", text=True, check=True) + logger.info(f"Installed one-off at {hhmm} -> {sessions} (self-cleaning)") + + +def crontab_remove_at(sessions: List[str], hhmm: str, logger: logging.Logger) -> None: + """Remove a pending one-off --at entry from the crontab.""" + comment = cron_at_comment(sessions, hhmm) + result = subprocess.run(["crontab", "-l"], capture_output=True, text=True) + existing = result.stdout if result.returncode == 0 else "" + lines = _strip_lines_by_tag(existing.splitlines(), comment) + new_cron = "\n".join(lines) + ("\n" if lines else "") + subprocess.run(["crontab", "-"], input=new_cron, text=True, check=True) + logger.info(f"Removed one-off at {hhmm} for {sessions}") + + +def cronicle_install_at(sessions: List[str], hhmm: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install a one-off wake schedule into the cronicle HCL config (tmp-hcl in tests). + + Cronicle's `cron` field accepts the standard 5-field expression, so a + time-of-day one-off is written as 'min hour * * *'. On this host the cronicle + daemon is NOT running (see AGENTS.md §38.10), so these schedules are INERT + and this path is only exercised by hermetic tests asserting the HCL string; + the LIVE one-off scheduler is crontab (crontab_install_at). + """ + _, minute, hour = parse_at_time(hhmm) + schedule_name = cron_at_schedule_name(sessions, hhmm) + config_path = cronicle_config_path() + existing = config_path.read_text() if config_path.exists() else "" + lines = existing.splitlines() + new_lines: List[str] = [] + 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 + new_schedule = f'''\nschedule "{schedule_name}" {{ + cron = "{minute} {hour} * * *" + + task "wake" {{ + command = [{", ".join(chr(34) + a + chr(34) for a in wake_command_args(sessions, None, message, pane))}] + }} +}}''' + config_path.write_text("\n".join(new_lines) + "\n\n" + new_schedule + "\n") + logger.info(f"Installed cronicle one-off: {schedule_name} -> {sessions} at {hhmm}") + + +def cronicle_remove_at(sessions: List[str], hhmm: str, logger: logging.Logger) -> None: + """Remove a one-off wake schedule from the cronicle HCL config.""" + schedule_name = cron_at_schedule_name(sessions, hhmm) + 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 one-off: {schedule_name}") + + +def install_at(sessions: List[str], hhmm: str, message: str, logger: logging.Logger, pane: Optional[str] = None) -> None: + """Install a one-off wake at HH:MM, routing through the configured backend. + + Cronicle (tmp-hcl) when selected (keeps tests hermetic) and the system + crontab (crnd) otherwise. The crontab path is strictly one-off: the entry + self-strips its tag after firing. ``hhmm`` is re-parsed here to validate. + """ + parse_at_time(hhmm) + if use_cronicle(): + cronicle_install_at(sessions, hhmm, message, logger, pane) + else: + crontab_install_at(sessions, hhmm, message, logger, pane) + logger.info(f"Scheduled one-off wake at {hhmm} for {sessions}") + + +def remove_at(sessions: List[str], hhmm: str, logger: logging.Logger) -> None: + """Remove a pending one-off --at wake (routes through the configured backend).""" + parse_at_time(hhmm) + if use_cronicle(): + cronicle_remove_at(sessions, hhmm, logger) + else: + crontab_remove_at(sessions, hhmm, logger) + + +def list_cron(logger: logging.Logger) -> None: + if use_cronicle(): + cronicle_list(logger) + 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: + result = subprocess.run(["zellij", "list-sessions"], capture_output=True, text=True, timeout=5) + clean = re.sub(r'\x1b\[[0-9;]*m', '', result.stdout) + for line in clean.splitlines(): + if line.startswith(session + " ") or line == session: + return True + except Exception: + pass + return False + + +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(cmd, capture_output=True, timeout=5) + return True + except Exception: + return False + + +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(cmd, capture_output=True, timeout=5) + return True + except Exception: + return False + + +# ─── Wake Action ──────────────────────────────────────────────────────────────────── + logger.info(f"One-shot timer set for {interval} ({interval_to_human(interval)})") + + def _wake(): + time.sleep(interval_seconds) + run_wake(sessions, message, logger, pane) + + thread = threading.Thread(target=_wake, daemon=True) + thread.start() + logger.info(f"Background timer started (thread: {thread.ident})") + pid_file = Path(f"/tmp/pi_wake_agent_{sessions[0]}.pid") + pid_file.write_text(str(os.getpid())) + + + +def parse_context_occupancy(session: str, pane: Optional[str] = None) -> Optional[float]: + """Parse the agent's context-occupancy % from its status bar. + + pi's status bar reads like ' %/1.0M (auto) ...'. We extract the + '' before '/1.0M'. Returns None if the pane is not a pi CLI, the dump + fails, or no figure is present (a doorbell must never block on a read). + """ + cmd = [ZELIG_BIN, "--session", session, "action", "dump-screen"] + if pane: + cmd += ["--pane-id", pane] + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=8) + except Exception: + return None + m = CONTEXT_RE.search(str(result.stdout or "")) + if not m: + return None + try: + return float(m.group(1)) + except ValueError: + return None + + +def context_warning(session: str, pane: Optional[str] = None) -> str: + """Build a context-occupancy heads-up to append to the steer (or ''). + + Critical (>=95%) => '/compact NOW' steer; warn (>=85%) => 'consider /compact'. + """ + pct = parse_context_occupancy(session, pane) + if pct is None: + return "" + if pct >= CONTEXT_CRIT_PCT: + return (f" | \U0001f534 HEADS UP: context at {pct:g}%/1.0M \u2014 /compact NOW or " + f"you'll lock up (no room to compact)") + if pct >= CONTEXT_WARN_PCT: + return f" | \u26a0\ufe0f heads up: your context is at {pct:g}% \u2014 consider /compact" + return "" + + +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}" + (f" (pane={pane})" if pane else "")) + + warns: List[str] = [] + for session in sessions: + # Verify session exists before attempting to wake + if not zellij_session_exists(session): + logger.warning(f"Session {session} does not exist, skipping") + continue + + # Read the agent's context-occupancy and append a heads-up to the steer + # when approaching the compact threshold (pi can lock up if she runs out + # of room to compact). + warn = context_warning(session, pane) + warns.append(warn) + logger.info(f"Context warning for {session}: {warn or '(none)'}") + + steer = f"[{AGENT_NICK} via {ZELIG_BIN}] {message} Run: h5i-bus msg inbox{warn}" + zellij_write_chars(session, steer, pane) + # Send ENTER keypresses with 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(): + try: + bus_warn = next((w for w in warns if w), "") + subprocess.run( + ["h5i", "msg", "send", "Fable", f"{message} (timer wakeup){bus_warn}"], + cwd=H5I_BUS_ROOT, + env={**os.environ, "H5I_AGENT": H5I_AGENT}, + capture_output=True, + timeout=5 + ) + except Exception: + pass # Silently ignore - fire and forget + + threading.Thread(target=_send_bus, daemon=True).start() + + logger.info(f"Wake sent to {sessions_str}") + +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, pane) + + thread = threading.Thread(target=_wake, daemon=True) + thread.start() + logger.info(f"Background timer started (thread: {thread.ident})") + pid_file = Path(f"/tmp/pi_wake_agent_{sessions[0]}.pid") + pid_file.write_text(str(os.getpid())) + return 0 + + +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, pane) + + while True: + logger.debug(f"Sleeping for {interval_seconds}s...") + time.sleep(interval_seconds) + run_wake(sessions, message, logger, pane) + + +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)})") + logger.info(f"Sessions: {sessions}") + logger.info(f"Message: {message}") + + for i in range(1, count + 1): + logger.info(f"Succession {i}/{count}") + run_wake(sessions, message, logger, pane) + if i < count: + logger.debug(f"Sleeping for {interval_seconds}s until next succession...") + time.sleep(interval_seconds) + + # Self-clean: remove any cron entry for this session/interval combo + try: + remove_cron(sessions, interval, logger) + logger.info("Self-cleanup complete (cron removed)") + except Exception as e: + logger.warning(f"Self-cleanup failed: {e}") + + logger.info("=== SUCCESSION COMPLETE ===") + + +def status(logger: logging.Logger) -> None: + print("=== pi_wake_agent Status ===") + print(f"Script: {SCRIPT_PATH}") + 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() + print("=== Active one-shot timers ===") + found = False + for pid_file in Path("/tmp").glob("pi_wake_agent_*.pid"): + found = True + try: + pid = int(pid_file.read_text().strip()) + os.kill(pid, 0) + print(f" PID {pid} (active)") + except (ProcessLookupError, ValueError): + print(f" PID {pid_file.read_text().strip()} (dead, cleaning up)") + pid_file.unlink(missing_ok=True) + if not found: + print(" (none)") + + +def validate_sessions(sessions: List[str], logger: logging.Logger) -> None: + logger.info(f"Validating sessions: {sessions}") + for session in sessions: + if zellij_session_exists(session): + logger.info(f" {session}: EXISTS") + else: + logger.warning(f" {session}: NOT FOUND (may be dead/EXITED)") + + +# ─── Session Parsing ───────────────────────────────────────────────────── +def parse_sessions(raw) -> List[str]: + if not raw: + return [] + if isinstance(raw, list): + return [s.strip() for s in raw] + return [s.strip() for s in raw.split(",") if s.strip()] + + +# ─── Usage ────────────────────────────────────────────────────────────── +def create_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description="pi_wake_agent.py — Reusable multi-agent wake-up timer with self-cron/daemon/succession/one-off --at", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +EXAMPLES: + # Install recurring 1-hour timer for one session + pi_wake_agent.py --install --interval 1h --session cc_UV_dev0_Fb + + # One-off wake at 01:50 AM — fires once, then self-removes from crontab + pi_wake_agent.py --at 0150 --session cc_UV_dev0_Fb --msg "continue" + + # Install 30-minute timer for multiple sessions + pi_wake_agent.py --install --interval 30m --sessions "cc_UV_dev0_Fb,cc_UV_dev1_48" --msg "Wake up!" + + # One-shot wake in 2 hours (no cron) + pi_wake_agent.py --once --interval 2h --session cc_UV_dev0_Fb --msg "Time's up!" + + # Run N times at interval, then self-clean + pi_wake_agent.py --succession --count 3 --interval 1h --session cc_UV_dev0_Fb --msg "Scheduled wake" + + # Run as daemon (long-lived process, no cron) + pi_wake_agent.py --daemon --interval 1h --session cc_UV_dev0_Fb + + # Remove timer + pi_wake_agent.py --remove --session cc_UV_dev0_Fb --interval 1h + + # List all timers + pi_wake_agent.py --list + + # Show status + pi_wake_agent.py --status + + # Validate sessions + pi_wake_agent.py --validate --session cc_UV_dev0_Fb + + # One-off wake at 01:50 AM (fires once, then self-removes from crontab) + pi_wake_agent.py --at 0150 --session cc_UV_dev0_Fb --msg "continue" + + # Cancel a pending one-off + pi_wake_agent.py --remove --at 0150 --session cc_UV_dev0_Fb +""" + ) + 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)") + parser.add_argument("--succession", action="store_const", const="succession", dest="mode", help="Run N times at interval, then self-clean") + parser.add_argument("--count", type=int, default=1, help="Number of successions (for --succession mode)") + parser.add_argument("--run", action="store_const", const="run", dest="mode", help="Internal: run wake action (called by cron)") + parser.add_argument("--remove", action="store_const", const="remove", dest="mode", help="Remove cron timer") + parser.add_argument("--list", action="store_const", const="list", dest="mode", help="List active cron timers") + parser.add_argument("--status", action="store_const", const="status", dest="mode", help="Show status (cron + one-shot timers)") + parser.add_argument("--validate", action="store_const", const="validate", dest="mode", help="Validate sessions exist in zellij") + 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") + parser.add_argument("--at", dest="at_time", default=None, metavar="HHMM", + help=("Schedule a ONE-OFF wake at HH:MM or HHMM (e.g. " + "0150 or 01:50). Fires once at that time-of-day then " + "self-removes from the crontab (crnd backend). Combine " + "with --remove to cancel, or --dry-run to preview. " + "Cannot combine with --once/--daemon/--succession/" + "--run/--list/--status/--validate.")) + parser.set_defaults(mode="install") + return parser + + +# ─── Main ─────────────────────────────────────────────────────────────── +def main() -> int: + parser = create_parser() + args = parser.parse_args() + + # Combine sessions + sessions: List[str] = [] + if args.sessions: + sessions.extend(parse_sessions(args.sessions)) + if args.session_list: + sessions.extend(args.session_list) + + # Setup logging + logger = setup_logging(args.debug) + + # --at HHMM: one-off time-of-day wake (self-cleaning). Mutually exclusive + # with --once/--daemon/--succession/--run which own their own timing. + if args.at_time is not None: + if args.mode not in ("install", "remove"): + logger.error("--at HHMM may only be used alone (one-off install) or with " + "--remove; it cannot combine with --once/--daemon/" + "--succession/--run/--list/--status/--validate") + return 1 + try: + parse_at_time(args.at_time) # fail-fast validation before install/remove + except ValueError as e: + logger.error(f"Invalid --at time {args.at_time!r}: {e}") + return 1 + if args.mode == "install": + args.mode = "at" + + # Validate + if len(sessions) == 0 and args.mode not in ("list", "status"): + logger.error("--session or --sessions required") + parser.print_help() + return 1 + + if args.mode == "succession" and args.count < 1: + logger.error("--count must be >= 1 for succession mode") + return 1 + + 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}") + if args.dry_run: + logger.info("DRY RUN: no actions will be executed") + + if args.dry_run: + logger.info("DRY RUN: would execute mode '%s' with sessions=%s, interval=%s, message='%s'", args.mode, sessions, args.interval, args.msg) + return 0 + + try: + if args.mode == "at": + install_at(sessions, args.at_time, args.msg, logger, args.pane) + elif args.mode == "install": + install_cron(sessions, args.interval, args.msg, logger, args.pane) + elif args.mode == "once": + run_once(sessions, args.interval, args.msg, logger, args.pane) + elif args.mode == "daemon": + run_daemon(sessions, args.interval, args.msg, logger, args.pane) + elif args.mode == "succession": + run_succession(sessions, args.interval, args.count, args.msg, logger, args.pane) + elif args.mode == "run": + run_wake(sessions, args.msg, logger, args.pane) + elif args.mode == "remove": + if args.at_time is not None: + remove_at(sessions, args.at_time, logger) + else: + remove_cron(sessions, args.interval, logger) + elif args.mode == "list": + list_cron(logger) + elif args.mode == "status": + status(logger) + elif args.mode == "validate": + validate_sessions(sessions, logger) + else: + logger.error(f"Unknown mode: {args.mode}") + return 1 + except Exception as e: + logger.error(f"Error: {e}") + if args.debug: + import traceback + traceback.print_exc() + return 1 + + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/test_pi_wake_agent.py b/test_pi_wake_agent.py new file mode 100755 index 0000000..0970507 --- /dev/null +++ b/test_pi_wake_agent.py @@ -0,0 +1,1134 @@ +#!/usr/bin/env python3 +""" +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 +import subprocess +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, + parse_at_time, + cron_at_comment, + cron_at_schedule_name, + build_at_cron_entry, + crontab_install_at, + crontab_remove_at, + cronicle_install_at, + cronicle_remove_at, + install_at, + remove_at, + use_cronicle, + cronicle_available, + cronicle_daemon_running, + run_wake, + SCRIPT_PATH, + LOG_FILE, + LOG_MAX_SIZE, + LOG_MAX_FILES, + CRON_COMMENT_PREFIX, + AGENT_NICK, + H5I_AGENT, + H5I_BUS_ROOT, + 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: + def test_hours(self): + 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") + with pytest.raises(ValueError): + parse_interval("1x") + with pytest.raises(ValueError): + parse_interval("") + +# ─── Test interval_to_cron ─────────────────────────────────────────────── + +class TestIntervalToCron: + def test_hours(self): + 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") + +# ─── Test interval_to_human ────────────────────────────────────────────── + +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" + +# ─── Test cron_comment ─────────────────────────────────────────────────── + +class TestCronComment: + def test_single_session(self): + 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" + +# ─── Test parse_sessions ───────────────────────────────────────────────── + +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"] + +# ─── Test main functions via subprocess ────────────────────────────────── + +SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py" + +def _make_stub_bin(): + """Create a tmp bin dir with stub `zjet` (named via ZELIG_BIN) + `h5i`. + + Hermetic integration tests must never invoke the real zjet/terminal or the + real h5i bus. The zjet stub is named after ``pi_wake_agent.ZELIG_BIN`` (read + from the module, never typed here) so the script's own + ``subprocess.run([ZELIG_BIN, ...])`` resolves to the stub when this dir + leads ``PATH``. The h5i stub is a no-op (``h5i`` is a short, typeable name). + """ + bin_dir = Path(tempfile.mkdtemp(prefix="pi_wake_stub_")) + zjet_stub = bin_dir / pi_wake_agent.ZELIG_BIN + zjet_stub.write_text( + "#!/usr/bin/env python3\n" + "import sys\n" + "args = sys.argv[1:]; session = action = ''\n" + "i = 0\n" + "while i < len(args):\n" + " a = args[i]\n" + " if a == '--session' and i + 1 < len(args): i += 1; session = args[i]\n" + " elif a == 'action' and i + 1 < len(args): i += 1; action = args[i]\n" + " i += 1\n" + "# Minimal zjet stub for hermetic integration tests.\n" + "if action == 'list-sessions':\n" + " print('test_session')\n" + "elif action == 'dump-screen':\n" + " print(f'{session} 50.0%/1.0M (auto) (zjet-stub) test_session')\n" + "# write-chars / write : no-op (fast, hermetic)\n" + "sys.exit(0)\n", + ) + zjet_stub.chmod(0o755) + h5i_stub = bin_dir / "h5i" + h5i_stub.write_text("#!/usr/bin/env python3\nimport sys\nsys.exit(0)\n") + h5i_stub.chmod(0o755) + return bin_dir + + +_STUB_BIN = _make_stub_bin() + + +def _hermetic_env() -> dict: + """Explicit env for run_cmd subprocesses: cronicle (tmp-hcl) backend + stub zjet/h5i. + + Forces the real crontab / cronicle.hcl / zjet / h5i bus to NEVER be touched + by integration tests. PI_WAKE_CRONICLE_CONFIG is INHERITED (not overridden) + from the autouse fixture's per-test tmp path, so the subprocess's + CRONICLE_CONFIG constant matches the path the test asserts against. + """ + return { + **os.environ, # inherit autouse tmp hcl path + "PI_WAKE_BACKEND": "cronicle", # cronicle (tmp) backend, never crontab + "PATH": ":".join(p for p in [str(_STUB_BIN), os.environ.get("PATH", "")] if p), + } + + +def run_cmd(args, env=None, timeout=30): + """Run the script and return (returncode, stdout, stderr). + + Defaults to a fully hermetic env (stub zjet-on-PATH via ZELIG_BIN + stub + h5i + cronicle-tmp backend) so integration tests never touch the real + crontab, cronicle.hcl, zjet, or h5i bus. + """ + if env is None: + env = _hermetic_env() + result = subprocess.run( + [sys.executable, SCRIPT] + args, + capture_output=True, + text=True, + timeout=timeout, + env=env, + ) + return result.returncode, result.stdout, result.stderr + +# ─── Integration Tests ────────────────────────────────────────────────── + +class TestHelp: + def test_help(self): + rc, out, err = run_cmd(["--help"]) + assert rc == 0 + assert "pi_wake_agent.py" in out + assert "--install" in out + assert "--once" in out + 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 setup_method(self): + # Clear cronicle config (uses the per-test tmp path) + config_path = cronicle_hcl_path() + if config_path.exists(): + config_path.unlink() + # Doctrine: do NOT touch the real crontab here. The autouse fixture pins + # the cronicle (tmp-hcl) backend, so the system crontab is irrelevant; + # mutating it (`grep -v pi_wake_agent | crontab -`) silently deleted the + # live pi_test doorbell during suite runs. + + def test_list_no_cron(self): + rc, out, err = run_cmd(["--list"]) + assert rc == 0 + assert "(none)" in out or "pi_wake_agent cron entries" in out + +class TestInstallRemove: + def setup_method(self): + # Clear cronicle config (per-test tmp path) + config_path = cronicle_hcl_path() + if config_path.exists(): + config_path.unlink() + # Doctrine: no real crontab mutation (cronicle tmp-hcl backend pinned by + # the autouse fixture). Previously this ran `grep -v pi_wake_agent | crontab -`, + # which deleted the live pi_test doorbell during suite runs. + + def teardown_method(self): + config_path = cronicle_hcl_path() + if config_path.exists(): + config_path.unlink() + # Doctrine: no real crontab mutation (cronicle tmp-hcl backend pinned). + + def test_install_and_list(self): + rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"]) + 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, f"install failed: {err}" + + rc, out, err = run_cmd(["--list"]) + assert rc == 0 + assert "s1_s2" in out # cronicle format: pi_wake_s1_s2_30m + + run_cmd(["--remove", "--sessions", "s1,s2", "--interval", "30m"]) + +class TestValidate: + def test_validate_existing(self): + # The test session might not exist, so we test the command runs + rc, out, err = run_cmd(["--validate", "--session", "pi_test"]) + assert rc == 0 # Should not crash + +class TestOnce: + def test_once_short(self): + # Use a very short interval + rc, out, err = run_cmd(["--once", "--interval", "1s", "--session", "test_session", "--msg", "quick test"], timeout=10) + assert rc == 0 + # Check log file for the message + time.sleep(2) + log_content = Path("/tmp/pi_wake_agent.log").read_text() + assert "One-shot timer set" in log_content + +class TestSuccession: + def test_succession_short(self): + # Test succession mode with very short intervals + rc, out, err = run_cmd([ + "--succession", "--count", "2", "--interval", "1s", + "--session", "test_session", "--msg", "succession test" + ], timeout=60) + assert rc == 0 + # Check logs + time.sleep(3) + 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 + assert rc != 0 or "error" in err.lower() + +class TestStatus: + def test_status(self): + rc, out, err = run_cmd(["--status"]) + 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): + # This is the internal mode called by cron + rc, out, err = run_cmd(["--run", "--session", "test_session", "--msg", "test"]) + assert rc == 0 # Should succeed even if session doesn't exist + +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 + +class TestSessionParsing: + def test_multiple_session_flags(self): + rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "s1", "--session", "s2", "--msg", "test"]) + assert rc == 0 + run_cmd(["--remove", "--session", "s1", "--interval", "1h"]) + run_cmd(["--remove", "--session", "s2", "--interval", "1h"]) + +class TestRemove: + def test_remove_nonexistent(self): + # Should not crash + rc, out, err = run_cmd(["--remove", "--session", "nonexistent", "--interval", "1h"]) + assert rc == 0 + +# ─── Test Logging ───────────────────────────────────────────────────────── + +class TestLogging: + def test_log_file_created(self): + run_cmd(["--install", "--interval", "1h", "--session", "log_test", "--msg", "test"]) + 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 + +class TestContextOccupancy: + """Hermetic: parse_context_occupancy + context_warning never call real zjet. + + Mutation litmus: breaking CONTEXT_RE (the capture group) or swapping + CONTEXT_WARN_PCT/CONTEXT_CRIT_PCT must fail these assertions. + """ + + @staticmethod + def _mock_stdout(monkeypatch, stdout): + m = MagicMock() + m.returncode = 0 + m.stdout = stdout + monkeypatch.setattr( + pi_wake_agent.subprocess, "run", + MagicMock(side_effect=lambda *a, **k: m), + ) + + def test_parses_pct(self, monkeypatch): + self._mock_stdout(monkeypatch, "... 91.3%/1.0M (auto) ...") + assert pi_wake_agent.parse_context_occupancy("pi_test", "terminal_1") == 91.3 + + def test_parses_integer_pct(self, monkeypatch): + self._mock_stdout(monkeypatch, "100%/1.0M") + assert pi_wake_agent.parse_context_occupancy("pi_test") == 100.0 + + def test_no_match_returns_none(self, monkeypatch): + self._mock_stdout(monkeypatch, "no percentage here") + assert pi_wake_agent.parse_context_occupancy("pi_test") is None + + def test_subprocess_failure_returns_none(self, monkeypatch): + monkeypatch.setattr( + pi_wake_agent.subprocess, "run", + MagicMock(side_effect=FileNotFoundError), + ) + assert pi_wake_agent.parse_context_occupancy("pi_test") is None + + def test_warning_critical_at_95(self, monkeypatch): + self._mock_stdout(monkeypatch, "95.0%/1.0M (auto) ...") + w = pi_wake_agent.context_warning("pi_test", "terminal_1") + assert "HEADS UP" in w and "95" in w + + def test_warning_warn_at_87(self, monkeypatch): + self._mock_stdout(monkeypatch, "87.0%/1.0M (auto)") + w = pi_wake_agent.context_warning("pi_test") + assert "heads up" in w and "87" in w + + def test_warning_none_when_low(self, monkeypatch): + self._mock_stdout(monkeypatch, "40.0%/1.0M (auto)") + assert pi_wake_agent.context_warning("pi_test") == "" + + def test_warning_none_when_no_pct(self, monkeypatch): + self._mock_stdout(monkeypatch, "no status bar") + assert pi_wake_agent.context_warning("pi_test") == "" + + +# ─── 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", dump_screen="pi_test\n"): + # 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 + if isinstance(cmd, list) and "dump-screen" in cmd: + m = MagicMock() # pi context-occupancy figure (testable) + m.stdout = dump_screen + 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) + + def test_context_warning_appended_to_steer(self): + # When pi's context is critical (>=95%), the steer must carry a heads-up. + mr = self._run(pane="terminal_1", dump_screen="95.0%/1.0M (auto) ...") + calls = self._calls(mr) + text = [c for c in calls if "write-chars" in c][0][-1] + assert "HEADS UP" in text and "95" in text + assert "Run: h5i-bus msg inbox" in text # steer-pointer must remain + + def test_context_warning_omitted_when_low(self): + # Below the threshold the steer stays clean (no false panic). + mr = self._run(pane="terminal_1", dump_screen="40.0%/1.0M (auto)") + calls = self._calls(mr) + text = [c for c in calls if "write-chars" in c][0][-1] + assert "HEADS UP" not in text + assert "heads up" not in text + +# ─── 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 + + +class TestHermeticity: + """Doctrine: integration tests must NOT mutate the live crontab/zjet/h5i. + + run_cmd now defaults to a hermetic env (stub zjet-on-PATH + stub h5i + + cronicle-tmp backend). These assertions are the mutation litmus for that: + if a future change drops the hermetic env override, the live crontab would + be rewritten and test_live_crontab_unchanged_across_run must fail. + """ + + @staticmethod + def _live_crontab(): + try: + r = subprocess.run(["crontab", "-l"], capture_output=True, text=True, timeout=5) + return r.stdout if r.returncode == 0 else "" + except Exception: + return "" + + def test_live_crontab_unchanged_across_run(self): + before = self._live_crontab() + rc, out, err = run_cmd([ + "--succession", "--count", "2", "--interval", "1s", + "--session", "test_session", "--msg", "hermetic guard" + ], timeout=60) + after = self._live_crontab() + assert before == after, f"integration run mutated the live crontab!\ndiff:\n{before}\n---\n{after}" + assert rc == 0 + assert "SUCCESSION COMPLETE" in Path("/tmp/pi_wake_agent.log").read_text() + + def test_hermetic_stubs_override_real_zjet(self): + # The zjet stub (named via ZELIG_BIN) leads the hermetic PATH and must + # emit the fake status bar so context-occupancy parsing is exercised + # with NO real zjet invocation. + env = _hermetic_env() + stub_dir = env["PATH"].split(":")[0] + stub = Path(stub_dir) / pi_wake_agent.ZELIG_BIN + assert stub.exists(), f"zjet stub not found at {stub}" + r = subprocess.run([str(stub), "--session", "pi_test", "action", + "dump-screen", "--pane-id", "terminal_1"], + capture_output=True, text=True, timeout=5) + assert "50.0%/1.0M" in r.stdout + # h5i stub is a no-op (no real bus write) + h5i = Path(stub_dir) / "h5i" + assert h5i.exists() + + +# ─── Test --at one-off time-of-day scheduling ───────────────────────────── + +class TestParseAtTime: + """Pure: HH:MM / HHMM parsing + range validation (mutation litmus).""" + + def test_hhmm_four_digit(self): + assert parse_at_time("0150") == ("0150", "50", "01") + + def test_hhmm_rejects_three_digits(self): + # HHMM is exactly 4 digits (0150); a bare 930 is ambiguous -> rejected. + with pytest.raises(ValueError): + parse_at_time("930") + + def test_hh_mm_colon(self): + assert parse_at_time("01:50") == ("0150", "50", "01") + assert parse_at_time("9:05") == ("0905", "05", "09") + + def test_midnight_and_last_minute(self): + assert parse_at_time("0000") == ("0000", "00", "00") + assert parse_at_time("2359") == ("2359", "59", "23") + + def test_bad_format(self): + with pytest.raises(ValueError): + parse_at_time("999") # 3 digits, not HHMM + with pytest.raises(ValueError): + parse_at_time("abc") # non-numeric + with pytest.raises(ValueError): + parse_at_time("12:5") # minute not 2 digits (regex is strict) + with pytest.raises(ValueError): + parse_at_time("") # empty + with pytest.raises(ValueError): + parse_at_time("1:50:00") # HH:MM:SS rejected + + def test_out_of_range(self): + with pytest.raises(ValueError): + parse_at_time("2400") # hour 24 + with pytest.raises(ValueError): + parse_at_time("2500") # hour 25 + with pytest.raises(ValueError): + parse_at_time("1260") # minute 60 + with pytest.raises(ValueError): + parse_at_time("12:99") # minute 99 + with pytest.raises(ValueError): + parse_at_time("23:60") + + def test_rejects_extra_digits(self): + with pytest.raises(ValueError): + parse_at_time("01500") # 5 digits + + +class TestCronAtTag: + """Tag + schedule-name generation (distinct from recurring entries).""" + + def test_comment_format(self): + assert cron_at_comment(["cdx_FLIGHT13"], "0150") == "pi_wake_agent:cdx_FLIGHT13:at:0150" + + def test_comment_multiple_sessions(self): + assert cron_at_comment(["a", "b"], "0150") == "pi_wake_agent:a,b:at:0150" + + def test_schedule_name_format(self): + assert cron_at_schedule_name(["cdx_FLIGHT13"], "0150") == "pi_wake_cdx_FLIGHT13_at_0150" + + def test_tag_distinct_from_recurring(self): + recurring = cron_comment(["cdx_FLIGHT13"], "20m") # pi_wake_agent:cdx_FLIGHT13:20m + at_tag = cron_at_comment(["cdx_FLIGHT13"], "0150") # ...:at:0150 + assert "at:" in at_tag + assert at_tag != recurring + assert "at:" not in recurring # recurring tag never looks like an at-tag + + +class TestBuildAtCronEntry: + """Pure: the crontab line for a one-off, self-cleaning --at wake.""" + + def _entry(self, **over): + kw = dict(sessions=["cdx_FLIGHT13"], hhmm="0150", message="continue", pane=None) + kw.update(over) + return build_at_cron_entry(**kw) + + def test_schedule_is_time_of_day(self): + e = self._entry() + assert e.startswith("50 01 * * * "), e # 01:50 -> min=50 hour=01 + + def test_wake_omits_interval(self): + e = self._entry() + assert "--interval" not in e # one-off is not interval driven + + def test_wake_has_run_sessions_and_message(self): + e = self._entry() + assert "--run" in e + assert "cdx_FLIGHT13" in e + assert "continue" in e + + def test_self_strip_targets_the_at_tag_not_recurring(self): + e = self._entry() + at_tag = cron_at_comment(["cdx_FLIGHT13"], "0150") + recurring = cron_comment(["cdx_FLIGHT13"], "20m") + assert f"grep -v '{at_tag}'" in e # strips THIS line + assert f"grep -v '{recurring}'" not in e # never strips recurring + + def test_self_strip_sits_after_wake_cmd(self): + e = self._entry() + wake = pi_wake_agent._at_wake_command(["cdx_FLIGHT13"], "continue") + assert wake in e # wake runs before self-strip + assert e.rstrip().endswith(f"# {cron_at_comment(['cdx_FLIGHT13'], '0150')}") + + def test_pane_threaded_into_command(self): + e = self._entry(pane="terminal_1") + cmd_part = e.split(" # ")[0] + assert "--pane" in cmd_part and "terminal_1" in cmd_part + + def test_pane_default_omitted(self): + e = self._entry(pane=None) + assert "--pane" not in e + + +class TestCrontabAtBackend: + """Hermetic crontab-backend tests for --at (subprocess.run mocked). + + Mirrors TestCrontabEntry: real crontab is never touched; we capture the + `crontab -` stdin to assert the time-of-day schedule, the self-strip suffix, + the at-tag comment, and that the pre-existing doorbell survives. Mutation + litmus: dropping the self-strip suffix must fail test_self_strip_present. + """ + + _DOORBELL = ( + "*/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" + ) + + @staticmethod + def _mock_crontab_at(monkeypatch, crontab_before, captured): + 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_sets_time_of_day_schedule(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", + logging.getLogger("t"), pane=None) + assert len(captured) == 1, "crontab - not invoked by crontab_install_at" + new = captured[0] + assert "50 01 * * *" in new, new # at-entry is mid-crontab + assert "pi_wake_agent:cdx_FLIGHT13:at:0150" in new + at_line = [ln for ln in new.splitlines() if "pi_wake_agent:cdx_FLIGHT13:at:0150" in ln] + assert at_line and "--interval" not in at_line[0] # at-ENTRY omits --interval (doorbell keeps its 20m) + + def test_install_self_strip_present(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + new = captured[0] + assert "grep -v" in new + assert "crontab -" in new # self-strip rewrites crontab + + def test_install_idempotent_replaces_same_tag(self, monkeypatch): + """Installing twice must NOT duplicate (replace, not append).""" + pre = self._DOORBELL + ( + "50 1 * * * cd /mnt/dolphinng5_predict && ... # pi_wake_agent:cdx_FLIGHT13:at:0150\n" + ) + captured = [] + self._mock_crontab_at(monkeypatch, pre, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + new = captured[0] + assert sum(1 for ln in new.splitlines() if "pi_wake_agent:cdx_FLIGHT13:at:0150" in ln) == 1 # one line (old replaced) + assert "pi_wake_agent:pi_test:20m" in new # doorbell intact + + def test_install_preserves_doorbell_and_unrelated(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + new = captured[0] + assert "pi_wake_agent:pi_test:20m" in new # doorbell intact + assert "pi_wake_agent:cdx_FLIGHT13:at:0150" in new # new at-entry present + assert self._DOORBELL.strip() in new # pre-existing line verbatim + + def test_remove_clears_at_entry(self, monkeypatch): + installed = self._DOORBELL + ( + "50 1 * * * cd /mnt/dolphinng5_predict && ... # pi_wake_agent:cdx_FLIGHT13:at:0150\n" + ) + captured = [] + self._mock_crontab_at(monkeypatch, installed, captured) + pi_wake_agent.crontab_remove_at(["cdx_FLIGHT13"], "0150", logging.getLogger("t")) + new = captured[0] + assert "pi_wake_agent:cdx_FLIGHT13:at:0150" not in new + assert "pi_wake_agent:pi_test:20m" in new # doorbell preserved + + def test_remove_missing_tag_noop(self, monkeypatch): + """Removing an absent --at entry leaves the crontab content unchanged.""" + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) + pi_wake_agent.crontab_remove_at(["nope"], "0150", logging.getLogger("t")) + new = captured[0] + assert "at:0150" not in new + assert new.strip() == self._DOORBELL.strip() + + def test_install_at_routes_to_crontab_when_selected(self, monkeypatch): + captured = [] + self._mock_crontab_at(monkeypatch, self._DOORBELL, captured) # forces crontab backend + pi_wake_agent.install_at(["cdx_FLIGHT13"], "0150", "continue", logging.getLogger("t")) + assert captured and "50 01 * * *" in captured[0] + + +class TestCronicleAtHcl: + """cronicle backend (tmp-hcl): --at writes a one-off schedule block. + + Asserts the generated HCL string (hermetic: tmp config, never the real + cronicle.hcl). Mirrors TestCronicleHslPane's shape + mutation litmus. + """ + + 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_block(self): + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "continue"]) + assert rc == 0, f"install --at failed: {err}" + content = cronicle_hcl_path().read_text() + assert 'schedule "pi_wake_cdx_FLIGHT13_at_0150"' in content + assert 'cron = "50 01 * * *"' in content + assert '"--run"' in content + assert '"cdx_FLIGHT13"' in content + assert '"continue"' in content + assert '"--interval"' not in content # one-off omits --interval + assert 'command = [' in content # valid HCL array (mutation litmus) + + def test_install_hcl_has_pane(self): + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--pane", "terminal_1", "--msg", "continue"]) + assert rc == 0, f"install --at failed: {err}" + content = cronicle_hcl_path().read_text() + assert '"--pane"' in content + assert '"terminal_1"' in content + + def test_install_is_idempotent_replaces_block(self): + """Reinstalling the same --at schedule replaces, not duplicates.""" + run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", "--msg", "continue"]) + run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", "--msg", "again"]) + content = cronicle_hcl_path().read_text() + assert content.count('schedule "pi_wake_cdx_FLIGHT13_at_0150"') == 1 + assert '"again"' in content + + def test_remove_clears_hcl_block(self): + cronicle_hcl_path().write_text( + 'schedule "pi_wake_cdx_FLIGHT13_at_0150" {\n' + ' cron = "50 1 * * *"\n' + ' task "wake" { command = ["x"] }\n' + '}\n' + 'schedule "pi_wake_keep_20m" {\n' + ' cron = "@every 20m"\n' + ' task "wake" { command = ["y"] }\n' + '}\n' + ) + rc, out, err = run_cmd(["--remove", "--at", "0150", "--session", "cdx_FLIGHT13"]) + assert rc == 0, f"remove --at failed: {err}" + content = cronicle_hcl_path().read_text() + assert 'schedule "pi_wake_cdx_FLIGHT13_at_0150"' not in content + assert 'schedule "pi_wake_keep_20m"' in content # unrelated schedule preserved + + +class TestAtDryRunHermeticity: + """--at --dry-run must not touch crontab, cronicle.hcl, or zjet/h5i.""" + + 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_dry_run_does_not_install(self): + before_log = Path(LOG_FILE).read_text() if LOG_FILE.exists() else "" + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "continue", "--dry-run"]) + assert rc == 0 + assert not cronicle_hcl_path().exists() # no HCL written (dry-run => no side effects) + # dry-run branch logs via the FileHandler (DEBUG-level); stderr's StreamHandler + # is WARNING-only, so assert on the log file, not on captured err/out. + after_log = Path(LOG_FILE).read_text() + assert "DRY RUN" in after_log + assert after_log.count("DRY RUN") > before_log.count("DRY RUN") # a new entry was emitted + + def test_dry_run_does_not_touch_live_crontab(self): + before = TestHermeticity._live_crontab() + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "continue", "--dry-run"]) + after = TestHermeticity._live_crontab() + assert rc == 0 + assert before == after, "dry-run mutated live crontab" + assert "at:0150" not in (after or "") + + +class TestAtLiveHermeticity: + """Mutation litmus: integration runs must NOT mutate the live crontab or + the real cronicle.hcl (the autouse hermetic env routes --at to the cronicle + tmp-hcl backend, so the live crontab is never rewritten).""" + + @staticmethod + def _live_cronicle_hcl(): + p = Path("/mnt/dolphinng5_predict/cronicle.hcl") + return p.read_text() if p.exists() else "" + + def test_live_crontab_unchanged_across_at(self): + before = TestHermeticity._live_crontab() + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "hermetic guard"]) + after = TestHermeticity._live_crontab() + assert rc == 0 + assert before == after, "integration run mutated the live crontab" + assert "pi_wake_cdx_FLIGHT13_at_0150" not in before # not pre-installed + + def test_live_cronicle_hcl_unchanged_across_at(self): + before = self._live_cronicle_hcl() + rc, out, err = run_cmd(["--at", "0150", "--session", "cdx_FLIGHT13", + "--msg", "hermetic guard"]) + after = self._live_cronicle_hcl() + assert rc == 0 + assert before == after, "integration run mutated real cronicle.hcl" + assert "pi_wake_cdx_FLIGHT13_at_0150" not in before + + +# ─── Run all tests ──────────────────────────────────────────────────────── + +if __name__ == "__main__": + pytest.main([__file__, "-v", "--tb=short"])