feat(pi_wake_agent): --at HHMM one-off self-cleaning wake mode

pi_wake_agent.py

@@ -25,7 +25,7 @@ import sys
 import threading
 import time
 from pathlib import Path
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 # ─── Constants ────────────────────────────────────────────────────────────
 SCRIPT_PATH = Path(__file__).resolve()
@@ -187,13 +187,18 @@ def cronicle_schedule_name(sessions: List[str], interval: str) -> str:
     return f"pi_wake_{sessions_str}_{interval}"
 
 
-def wake_command_args(sessions: List[str], interval: str, message: str, pane: Optional[str] = None) -> List[str]:
+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).
 
-    Always passes --interval so the cronicle/crnd entry is self-describing.
+    ``interval`` is the recurring cadence used by cron/crontab entries; pass
-    --pane targets a specific zjet pane (e.g. the bottom Soundcheck pane).
+    ``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", "--interval", interval, "--sessions", ",".join(sessions)]
+    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]
@@ -347,9 +352,199 @@ def remove_cron(sessions: List[str], interval: str, logger: logging.Logger) -> N
         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)
@@ -609,13 +804,16 @@ def parse_sessions(raw) -> List[str]:
 # ─── 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",
+        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!"
 
@@ -639,6 +837,12 @@ EXAMPLES:
 
   # 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")
@@ -658,6 +862,13 @@ EXAMPLES:
     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
 
@@ -677,6 +888,22 @@ def main() -> int:
     # 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")
@@ -703,7 +930,9 @@ def main() -> int:
         return 0
     
     try:
-        if args.mode == "install":
+        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)
@@ -714,6 +943,9 @@ def main() -> int:
         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)

test_pi_wake_agent.py

@@ -32,6 +32,16 @@ from pi_wake_agent import (
     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,
@@ -778,6 +788,346 @@ class TestHermeticity:
         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__":