Add Pi wake-agent watcher: pi_wake_agent.py + spec + tests

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

@@ -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:<sessions>:<interval>`
+
+---
+
+## 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`

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 <zellij-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 <pool4@normey.net>`

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: '<pct>%/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 <root> && export PATH=... && export H5I_AGENT=... && <wake_cmd> ; (crontab -l | grep -v '<tag>' ; echo) | crontab - # <tag>'
+    The self-strip suffix removes THIS line (matched by its unique 'at:<HHMM>'
+    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:<s>:<interval>' 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:<HHMM>' 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 '<cost> <pct>%/1.0M (auto) ...'. We extract the
+    '<pct>' 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())