diff --git a/agent/conversation_compression.py b/agent/conversation_compression.py index f2b3a6a00357..697258ebf748 100644 --- a/agent/conversation_compression.py +++ b/agent/conversation_compression.py @@ -3304,6 +3304,12 @@ def _release_lock() -> None: migrate_goal_to_session(old_session_id, agent.session_id, reason="compression") except Exception as _goal_err: logger.debug("Could not migrate goal on compression: %s", _goal_err) + # Same boundary hazard for /heartbeat state — carry it too. + try: + from hermes_cli.heartbeat import migrate_heartbeat_to_session + migrate_heartbeat_to_session(old_session_id, agent.session_id) + except Exception as _hb_err: + logger.debug("Could not migrate heartbeat on compression: %s", _hb_err) # Auto-number the title for the continuation session if old_title: try: diff --git a/cli.py b/cli.py index 9583383f61d9..08ec446ddaa4 100644 --- a/cli.py +++ b/cli.py @@ -10300,6 +10300,8 @@ def process_command(self, command: str) -> bool: _cprint(f" No agent running; queued as next turn: {payload[:80]}{'...' if len(payload) > 80 else ''}") elif canonical == "goal": self._handle_goal_command(cmd_original) + elif canonical == "heartbeat": + self._handle_heartbeat_command(cmd_original) elif canonical == "moa": # /moa is one-shot sugar only: run a single prompt through the # default MoA preset, then restore the prior model. To *switch* to a @@ -10578,6 +10580,72 @@ def _get_goal_manager(self): self._goal_manager = mgr return mgr + def _get_heartbeat_manager(self): + """Return the HeartbeatManager bound to the current session_id. + + Cached on ``self._heartbeat_manager`` and rebound lazily when + ``session_id`` changes (mirrors ``_get_goal_manager``). + """ + try: + from hermes_cli.heartbeat import HeartbeatManager + except Exception as exc: + logging.debug("heartbeat manager unavailable: %s", exc) + return None + + sid = getattr(self, "session_id", None) or "" + if not sid: + return None + + existing = getattr(self, "_heartbeat_manager", None) + if existing is not None and getattr(existing, "session_id", None) == sid: + return existing + + mgr = HeartbeatManager(session_id=sid) + self._heartbeat_manager = mgr + return mgr + + def _start_heartbeat_watchdog(self): + """Start the idle-poll thread that fires due heartbeats. + + Same pattern as the wake-word watchdog: a daemon thread polls a few + times a minute; when the session is idle (no agent running, empty + input queue) and the heartbeat is due, its prompt is injected into + ``_pending_input`` as a normal user turn. Missed ticks coalesce — + the anchor resets on fire, so a busy hour yields ONE heartbeat turn, + not a backlog. Idempotent; safe to call on every /heartbeat set. + """ + if getattr(self, "_heartbeat_watchdog_started", False): + return + self._heartbeat_watchdog_started = True + + from hermes_cli.heartbeat import POLL_SECONDS + + def _loop(): + try: + while not getattr(self, "_should_exit", False): + time.sleep(POLL_SECONDS) + try: + mgr = self._get_heartbeat_manager() + if mgr is None or not mgr.is_active(): + continue + busy = ( + self._agent_running + or getattr(self, "_voice_recording", False) + or getattr(self, "_voice_processing", False) + or not self._pending_input.empty() + ) + if busy: + continue + prompt = mgr.due_prompt() + if prompt: + self._pending_input.put(prompt) + except Exception as exc: + logging.debug("heartbeat watchdog tick failed: %s", exc) + finally: + self._heartbeat_watchdog_started = False + + threading.Thread(target=_loop, daemon=True, name="heartbeat-watchdog").start() + def _owns_process_notification(self, event: dict) -> bool: diff --git a/gateway/run.py b/gateway/run.py index 24d501b5b752..8f1edd9c5435 100644 --- a/gateway/run.py +++ b/gateway/run.py @@ -14025,6 +14025,7 @@ async def _dispatch_busy_slash_command( "background": self._handle_background_command, "kanban": self._handle_kanban_command, "subgoal": self._handle_subgoal_command, + "heartbeat": self._handle_heartbeat_command, "yolo": self._handle_yolo_command, "verbose": self._handle_verbose_command, "footer": self._handle_footer_command, @@ -15230,6 +15231,9 @@ async def _do_undo(): if canonical == "goal": return await self._handle_goal_command(event) + if canonical == "heartbeat": + return await self._handle_heartbeat_command(event) + if canonical == "moa": # /moa is one-shot sugar only: run a single prompt through the # default MoA preset, then restore the prior model. To *switch* to a @@ -18562,6 +18566,99 @@ async def _get_goal_manager_for_event(self, event: "MessageEvent"): max_turns = self._goal_max_turns_from_config() return GoalManager(session_id=sid, default_max_turns=max_turns), session_entry + async def _get_heartbeat_manager_for_event(self, event: "MessageEvent"): + """Return a HeartbeatManager bound to the session for this event. + + Returns ``(manager, session_entry)`` or ``(None, None)``. + """ + try: + from hermes_cli.heartbeat import HeartbeatManager + except Exception as exc: + logger.debug("heartbeat manager unavailable: %s", exc) + return None, None + try: + session_entry = await self.async_session_store.get_or_create_session(event.source) + except Exception as exc: + logger.debug("heartbeat manager: session lookup failed: %s", exc) + return None, None + sid = getattr(session_entry, "session_id", None) or "" + if not sid: + return None, None + return HeartbeatManager(session_id=sid), session_entry + + def _register_heartbeat_watch(self, quick_key: str, source: Any, session_id: str) -> None: + """Track a session with an active heartbeat and start the poller. + + The registry maps ``quick_key`` → ``(source, session_id)`` so the + poller can rebuild a MessageEvent and enqueue via the adapter FIFO. + In-memory by design: heartbeat STATE survives restarts in SessionDB, + but firing resumes when the user touches /heartbeat again in the new + gateway process (documented; durable schedules belong to cron). + """ + watch = getattr(self, "_heartbeat_watch", None) + if watch is None: + watch = {} + self._heartbeat_watch = watch + watch[quick_key] = (source, session_id) + self._start_heartbeat_poller() + + def _unregister_heartbeat_watch(self, quick_key: str) -> None: + watch = getattr(self, "_heartbeat_watch", None) + if watch: + watch.pop(quick_key, None) + + def _start_heartbeat_poller(self) -> None: + """Start the single gateway-wide heartbeat poll task (idempotent).""" + existing = getattr(self, "_heartbeat_poll_task", None) + if existing is not None and not existing.done(): + return + + from hermes_cli.heartbeat import POLL_SECONDS + + async def _poll_loop(): + while True: + await asyncio.sleep(POLL_SECONDS) + watch = getattr(self, "_heartbeat_watch", None) + if not watch: + continue + for quick_key, (source, session_id) in list(watch.items()): + try: + # Busy sessions coalesce their tick to the next idle poll. + if quick_key in self._running_agents: + continue + from hermes_cli.heartbeat import HeartbeatManager + + mgr = HeartbeatManager(session_id=session_id) + if not mgr.has_heartbeat(): + watch.pop(quick_key, None) + continue + prompt = mgr.due_prompt() + if not prompt: + continue + adapter = self._adapter_for_source(source) + if adapter is None: + continue + hb_event = MessageEvent( + text=prompt, + message_type=MessageType.TEXT, + source=source, + message_id=None, + channel_prompt=None, + ) + self._enqueue_fifo(quick_key, hb_event, adapter) + except Exception as exc: + logger.debug("heartbeat poll for %s failed: %s", quick_key, exc) + + try: + task = asyncio.create_task(_poll_loop()) + self._heartbeat_poll_task = task + _bg = getattr(self, "_background_tasks", None) + if _bg is not None: + _bg.add(task) + task.add_done_callback(_bg.discard) + except Exception: + logger.debug("Failed to start heartbeat poller", exc_info=True) + async def _send_goal_status_notice(self, source: Any, message: str) -> None: diff --git a/gateway/slash_commands.py b/gateway/slash_commands.py index 7b87e435055c..762d01d898cd 100644 --- a/gateway/slash_commands.py +++ b/gateway/slash_commands.py @@ -2740,6 +2740,78 @@ async def _handle_goal_command(self, event: "MessageEvent") -> str: return f"{base}\n(Couldn't draft a contract — running as a free-form goal.)" return base + async def _handle_heartbeat_command(self, event: "MessageEvent") -> str: + """Handle /heartbeat for gateway platforms (mirror of CLI handler). + + Sets/manages the session's one recurring re-entry prompt. The + gateway-wide poller injects due heartbeats through the adapter FIFO + as ordinary user turns, so alternation and caching are untouched. + """ + from hermes_cli.heartbeat import parse_interval, format_interval, MIN_INTERVAL_SECONDS + + args = (event.get_command_args() or "").strip() + lower = args.lower() + + mgr, session_entry = await self._get_heartbeat_manager_for_event(event) + if mgr is None: + return "Heartbeats unavailable (no session)." + + quick_key = self._session_key_for_source(event.source) if event.source else None + + if not args or lower == "status": + return mgr.status_line() + + if lower == "pause": + state = mgr.pause() + return f"⏸ Heartbeat paused: {state.prompt}" if state else "No heartbeat set." + + if lower == "resume": + state = mgr.resume() + if state is None: + return "No heartbeat to resume." + if quick_key and event.source is not None: + self._register_heartbeat_watch(quick_key, event.source, mgr.session_id) + return f"▶ Heartbeat resumed (every {format_interval(state.interval_seconds)}): {state.prompt}" + + if lower in {"clear", "stop", "off"}: + had = mgr.clear() + if quick_key: + self._unregister_heartbeat_watch(quick_key) + return "✓ Heartbeat cleared." if had else "No heartbeat set." + + # Set: `/heartbeat every 10m ` (also accepts `10m `). + tokens = args.split(None, 2) + interval = None + prompt = "" + if tokens and tokens[0].lower() == "every" and len(tokens) >= 2: + interval = parse_interval(f"every {tokens[1]}") + prompt = tokens[2] if len(tokens) > 2 else "" + elif tokens: + interval = parse_interval(tokens[0]) + prompt = args[len(tokens[0]):].strip() if interval and interval > 0 else "" + + if interval is None: + return ( + "Usage: /heartbeat every (e.g. /heartbeat every 10m Check CI)\n" + "Also: /heartbeat status | pause | resume | clear" + ) + if interval < 0: + return f"Interval too small — minimum is {MIN_INTERVAL_SECONDS}s." + if not prompt.strip(): + return "Usage: /heartbeat every — the prompt is required." + + try: + state = mgr.set(prompt, interval) + except ValueError as exc: + return f"Invalid heartbeat: {exc}" + if quick_key and event.source is not None: + self._register_heartbeat_watch(quick_key, event.source, mgr.session_id) + return ( + f"♥ Heartbeat set (every {format_interval(state.interval_seconds)}): {state.prompt}\n" + "Fires as a normal turn whenever this session is idle and the interval has " + "elapsed. Lives while the gateway runs — use `hermes cron` for durable schedules." + ) + async def _handle_subgoal_command(self, event: "MessageEvent") -> str: """Handle /subgoal for gateway platforms (mirror of CLI handler). diff --git a/hermes_cli/cli_commands_mixin.py b/hermes_cli/cli_commands_mixin.py index 9f9071ea94ef..a0fb6aa15d75 100644 --- a/hermes_cli/cli_commands_mixin.py +++ b/hermes_cli/cli_commands_mixin.py @@ -2368,6 +2368,91 @@ def _handle_browser_command(self, cmd: str): print(" status Show current browser mode") print() + def _handle_heartbeat_command(self, cmd: str) -> None: + """Dispatch /heartbeat: set / status / pause / resume / clear. + + ``/heartbeat every 10m Check the deployment`` sets the session's one + recurring instruction; the idle watchdog injects it as a normal user + turn whenever due. Session-scoped and in-process — for durable + cross-process schedules use `hermes cron`. + """ + from cli import _DIM, _RST, _cprint + from hermes_cli.heartbeat import parse_interval, format_interval + + parts = (cmd or "").strip().split(None, 1) + arg = parts[1].strip() if len(parts) > 1 else "" + lower = arg.lower() + + mgr = self._get_heartbeat_manager() + if mgr is None: + _cprint(f" {_DIM}Heartbeats unavailable (no active session).{_RST}") + return + + if not arg or lower == "status": + _cprint(f" {mgr.status_line()}") + return + + if lower == "pause": + state = mgr.pause() + if state is None: + _cprint(f" {_DIM}No heartbeat set.{_RST}") + else: + _cprint(f" ⏸ Heartbeat paused: {state.prompt}") + return + + if lower == "resume": + state = mgr.resume() + if state is None: + _cprint(f" {_DIM}No heartbeat to resume.{_RST}") + else: + self._start_heartbeat_watchdog() + _cprint(f" ▶ Heartbeat resumed (every {format_interval(state.interval_seconds)}): {state.prompt}") + return + + if lower in {"clear", "stop", "off"}: + if mgr.clear(): + _cprint(" ✓ Heartbeat cleared.") + else: + _cprint(f" {_DIM}No heartbeat set.{_RST}") + return + + # Set: `/heartbeat every 10m ` (also accepts `10m `). + tokens = arg.split(None, 2) + interval = None + prompt = "" + if tokens and tokens[0].lower() == "every" and len(tokens) >= 2: + interval = parse_interval(f"every {tokens[1]}") + prompt = tokens[2] if len(tokens) > 2 else "" + elif tokens: + interval = parse_interval(tokens[0]) + prompt = arg[len(tokens[0]):].strip() if interval and interval > 0 else "" + + if interval is None: + _cprint(" Usage: /heartbeat every (e.g. /heartbeat every 10m Check CI)") + _cprint(f" {_DIM}Also: /heartbeat status | pause | resume | clear{_RST}") + return + if interval < 0: + from hermes_cli.heartbeat import MIN_INTERVAL_SECONDS + _cprint(f" Interval too small — minimum is {MIN_INTERVAL_SECONDS}s.") + return + if not prompt.strip(): + _cprint(" Usage: /heartbeat every — the prompt is required.") + return + + try: + state = mgr.set(prompt, interval) + except ValueError as exc: + _cprint(f" Invalid heartbeat: {exc}") + return + self._start_heartbeat_watchdog() + _cprint(f" ♥ Heartbeat set (every {format_interval(state.interval_seconds)}): {state.prompt}") + _cprint( + f" {_DIM}Fires as a normal turn whenever the session is idle and the " + f"interval has elapsed. /heartbeat pause | resume | clear to manage; " + f"lives only while this Hermes process runs — use `hermes cron` for " + f"durable schedules.{_RST}" + ) + def _handle_goal_command(self, cmd: str) -> None: """Dispatch /goal subcommands: set / draft / show / status / pause / resume / clear.""" from cli import _DIM, _RST, _cprint diff --git a/hermes_cli/commands.py b/hermes_cli/commands.py index a803ee925f30..b836bd66583c 100644 --- a/hermes_cli/commands.py +++ b/hermes_cli/commands.py @@ -159,6 +159,10 @@ class CommandDef: CommandDef("goal", "Set a standing goal Hermes works on across turns until achieved", "Session", args_hint="[text | draft | show | pause | resume | clear | status | wait | unwait]", busy_policy="dispatch", busy_handler="goal"), + CommandDef("heartbeat", "Set a recurring prompt that re-enters this session when idle", "Session", + aliases=("hb",), args_hint="[every | status | pause | resume | clear]", + subcommands=("status", "pause", "resume", "clear"), + busy_policy="dispatch"), CommandDef("moa", "Run one prompt through the default Mixture of Agents preset, then restore your model", "Session", args_hint="", busy_policy="reject", busy_handler="moa"), CommandDef("subgoal", "Add or manage extra criteria on the active goal", "Session", @@ -1261,7 +1265,9 @@ def discord_skill_commands_by_category( # /hermes update on Slack. Demoted to free the native slot /approvals now # claims — without this entry /approvals tips the registry past the 50-cap # and silently clamps /update off, breaking Telegram parity. -_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff", "update"}) +# - heartbeat: session heartbeat management; reached via /hermes heartbeat +# on Slack. Added at the 50-cap — a native slot would clamp /insights. +_SLACK_VIA_HERMES_ONLY = frozenset({"topup", "moa", "debug", "egress", "init", "version", "diff", "update", "heartbeat"}) def _sanitize_slack_name(raw: str) -> str: @@ -2188,8 +2194,12 @@ def get_suggestion(self, buffer, document): if len(parts) == 1 and not text.endswith(" "): # Still typing the command name: /upd → suggest "ate" + # Prefer the SHORTEST matching command so a short, high-frequency + # command keeps its ghost text when a longer command shares its + # prefix (e.g. /he → "lp" for /help, not "artbeat" for + # /heartbeat; type one more letter to steer). word = text[1:].lower() - for cmd in COMMANDS: + for cmd in sorted(COMMANDS, key=len): if self._completer is not None and not self._completer._command_allowed(cmd): continue cmd_name = cmd[1:] # strip leading / diff --git a/hermes_cli/heartbeat.py b/hermes_cli/heartbeat.py new file mode 100644 index 000000000000..bf0df9f5b686 --- /dev/null +++ b/hermes_cli/heartbeat.py @@ -0,0 +1,332 @@ +"""Session heartbeats — recurring re-entry prompts for the current session. + +A heartbeat is one user-owned recurring instruction bound to a session +(`/heartbeat every 10m Check the deployment and report meaningful changes`). +When due AND the session is idle, the prompt is injected as a normal user +turn — same mechanism as a /goal continuation, so message-role alternation +and prompt caching are untouched. If the agent is busy at the due moment, +the tick coalesces: it fires once when the session next goes idle, never +stacking a backlog. + +This is deliberately session-scoped and in-process (CLI process or gateway +process must be running) — the durable cross-process scheduling surface +remains ``hermes cron`` / the ``cronjob`` tool, which runs in isolated +sessions. A heartbeat is for "keep re-entering THIS conversation", the +cron system is for "run this job on a schedule". Distinct by design. + +State is persisted in SessionDB ``state_meta`` keyed by +``heartbeat:`` so ``/resume`` picks it up. + +Invariants (mirrors goals.py): +- Injection is a plain user message. No system-prompt mutation, no toolset + swap — prompt caching stays intact. +- A real user message always wins: heartbeats only fire into an idle + session with an empty input queue. +- Failures are contained: any DB/import error degrades to "no heartbeat", + never to a crashed input loop. +""" + +from __future__ import annotations + +import json +import logging +import re +import time +from dataclasses import dataclass, asdict +from typing import Any, Dict, Optional + +logger = logging.getLogger(__name__) + + +# Floor: a heartbeat that re-enters the session more often than once a +# minute is a busy-loop, not a heartbeat. (Prime-Agent uses a similar floor.) +MIN_INTERVAL_SECONDS = 60 +# How often drivers poll for due heartbeats. Not user-facing. +POLL_SECONDS = 5.0 + +HEARTBEAT_PROMPT_TEMPLATE = ( + "[Heartbeat — recurring instruction, fires every {interval}]\n" + "{prompt}\n\n" + "If there is nothing meaningful to do or report for this instruction " + "right now, reply briefly that nothing has changed and stop — do not " + "invent work." +) + +_INTERVAL_RE = re.compile( + r"^\s*(?:every\s+)?(\d+(?:\.\d+)?)\s*(s|sec|secs|seconds?|m|min|mins|minutes?|h|hr|hrs|hours?|d|days?)\s*$", + re.IGNORECASE, +) + +_UNIT_SECONDS = { + "s": 1, "sec": 1, "secs": 1, "second": 1, "seconds": 1, + "m": 60, "min": 60, "mins": 60, "minute": 60, "minutes": 60, + "h": 3600, "hr": 3600, "hrs": 3600, "hour": 3600, "hours": 3600, + "d": 86400, "day": 86400, "days": 86400, +} + + +def parse_interval(text: str) -> Optional[int]: + """Parse ``10m`` / ``every 2h`` / ``every 90 minutes`` into seconds. + + Returns None when the text is not an interval. Values below + ``MIN_INTERVAL_SECONDS`` are rejected (returns -1 so callers can + distinguish "not an interval" from "too small"). + """ + if not text: + return None + m = _INTERVAL_RE.match(text) + if not m: + return None + value = float(m.group(1)) + unit = m.group(2).lower() + seconds = int(value * _UNIT_SECONDS[unit]) + if seconds < MIN_INTERVAL_SECONDS: + return -1 + return seconds + + +def format_interval(seconds: int) -> str: + """Human-readable interval (``600`` → ``10m``).""" + seconds = int(seconds) + if seconds % 86400 == 0: + return f"{seconds // 86400}d" + if seconds % 3600 == 0: + return f"{seconds // 3600}h" + if seconds % 60 == 0: + return f"{seconds // 60}m" + return f"{seconds}s" + + +@dataclass +class HeartbeatState: + """Serializable per-session heartbeat.""" + + prompt: str + interval_seconds: int + status: str = "active" # active | paused | cleared + created_at: float = 0.0 + last_fired_at: float = 0.0 + fire_count: int = 0 + + def to_json(self) -> str: + return json.dumps(asdict(self), ensure_ascii=False) + + @classmethod + def from_json(cls, raw: str) -> "HeartbeatState": + data = json.loads(raw) + return cls( + prompt=str(data.get("prompt") or ""), + interval_seconds=int(data.get("interval_seconds", 0) or 0), + status=str(data.get("status") or "active"), + created_at=float(data.get("created_at", 0.0) or 0.0), + last_fired_at=float(data.get("last_fired_at", 0.0) or 0.0), + fire_count=int(data.get("fire_count", 0) or 0), + ) + + def is_due(self, now: Optional[float] = None) -> bool: + if self.status != "active" or not self.prompt or self.interval_seconds <= 0: + return False + now = now if now is not None else time.time() + anchor = self.last_fired_at or self.created_at + return (now - anchor) >= self.interval_seconds + + def render_prompt(self) -> str: + return HEARTBEAT_PROMPT_TEMPLATE.format( + interval=format_interval(self.interval_seconds), + prompt=self.prompt, + ) + + +# ────────────────────────────────────────────────────────────────────── +# Persistence (SessionDB state_meta) — same pattern as goals.py +# ────────────────────────────────────────────────────────────────────── + + +def _meta_key(session_id: str) -> str: + return f"heartbeat:{session_id}" + + +def _get_session_db() -> Optional[Any]: + # Reuse the goals module's per-HERMES_HOME cached SessionDB so both + # features share one connection instead of thrashing the file. + try: + from hermes_cli.goals import _get_session_db as _goals_db + + return _goals_db() + except Exception as exc: # pragma: no cover + logger.debug("HeartbeatManager: SessionDB bootstrap failed (%s)", exc) + return None + + +def load_heartbeat(session_id: str) -> Optional[HeartbeatState]: + if not session_id: + return None + db = _get_session_db() + if db is None: + return None + try: + raw = db.get_meta(_meta_key(session_id)) + except Exception as exc: + logger.debug("HeartbeatManager: get_meta failed: %s", exc) + return None + if not raw: + return None + try: + state = HeartbeatState.from_json(raw) + except Exception as exc: + logger.warning("HeartbeatManager: could not parse stored heartbeat for %s: %s", session_id, exc) + return None + return None if state.status == "cleared" else state + + +def save_heartbeat(session_id: str, state: HeartbeatState) -> None: + if not session_id: + return + db = _get_session_db() + if db is None: + return + try: + db.set_meta(_meta_key(session_id), state.to_json()) + except Exception as exc: + logger.debug("HeartbeatManager: set_meta failed: %s", exc) + + +# ────────────────────────────────────────────────────────────────────── +# Manager — the surface CLI + gateway talk to +# ────────────────────────────────────────────────────────────────────── + + +class HeartbeatManager: + """Per-session heartbeat state + due-tick decisions. + + Drivers (CLI thread / gateway task) call :meth:`due_prompt` on a poll + cadence while the session is idle; a non-None return is the user-role + message to inject. Firing is recorded immediately so a slow turn can't + double-fire. + """ + + def __init__(self, session_id: str): + self.session_id = session_id + self._state: Optional[HeartbeatState] = load_heartbeat(session_id) + + @property + def state(self) -> Optional[HeartbeatState]: + return self._state + + def has_heartbeat(self) -> bool: + return self._state is not None and self._state.status in {"active", "paused"} + + def is_active(self) -> bool: + return self._state is not None and self._state.status == "active" + + def status_line(self) -> str: + s = self._state + if s is None: + return "No heartbeat. Set one with /heartbeat every ." + every = format_interval(s.interval_seconds) + fired = f", fired {s.fire_count}×" if s.fire_count else "" + if s.status == "active": + anchor = s.last_fired_at or s.created_at + next_in = max(0, int(anchor + s.interval_seconds - time.time())) + return f"♥ Heartbeat (every {every}, next in ~{next_in}s{fired}): {s.prompt}" + if s.status == "paused": + return f"⏸ Heartbeat (paused, every {every}{fired}): {s.prompt}" + return f"Heartbeat ({s.status}, every {every}{fired}): {s.prompt}" + + # --- mutation ----------------------------------------------------- + + def set(self, prompt: str, interval_seconds: int) -> HeartbeatState: + prompt = (prompt or "").strip() + if not prompt: + raise ValueError("heartbeat prompt is empty") + interval_seconds = int(interval_seconds) + if interval_seconds < MIN_INTERVAL_SECONDS: + raise ValueError(f"interval must be at least {MIN_INTERVAL_SECONDS}s") + state = HeartbeatState( + prompt=prompt, + interval_seconds=interval_seconds, + status="active", + created_at=time.time(), + ) + self._state = state + save_heartbeat(self.session_id, state) + return state + + def pause(self) -> Optional[HeartbeatState]: + if not self._state: + return None + self._state.status = "paused" + save_heartbeat(self.session_id, self._state) + return self._state + + def resume(self) -> Optional[HeartbeatState]: + if not self._state: + return None + self._state.status = "active" + # Re-anchor so resuming doesn't instantly fire a stale tick. + self._state.last_fired_at = time.time() + save_heartbeat(self.session_id, self._state) + return self._state + + def clear(self) -> bool: + if self._state is None: + return False + self._state.status = "cleared" + save_heartbeat(self.session_id, self._state) + self._state = None + return True + + # --- driver entry point -------------------------------------------- + + def due_prompt(self, now: Optional[float] = None) -> Optional[str]: + """Return the injection prompt if the heartbeat is due, else None. + + Records the fire immediately (before the turn runs) so overlapping + polls or a long turn can never double-fire the same tick. Missed + ticks coalesce into one — the anchor resets to NOW, not to the + theoretical schedule. + """ + s = self._state + if s is None or not s.is_due(now): + return None + s.last_fired_at = now if now is not None else time.time() + s.fire_count += 1 + save_heartbeat(self.session_id, s) + return s.render_prompt() + + +def migrate_heartbeat_to_session(old_session_id: str, new_session_id: str) -> bool: + """Carry a heartbeat across a compression session rotation. + + Same shape as ``goals.migrate_goal_to_session`` — copy to the child, + archive the parent row, never raise. + """ + if not old_session_id or not new_session_id or old_session_id == new_session_id: + return False + try: + state = load_heartbeat(old_session_id) + if state is None: + return False + if load_heartbeat(new_session_id) is not None: + return False + save_heartbeat(new_session_id, state) + state.status = "cleared" + save_heartbeat(old_session_id, state) + return True + except Exception as exc: # pragma: no cover - defensive + logger.debug("HeartbeatManager: migration failed: %s", exc) + return False + + +__all__ = [ + "HeartbeatState", + "HeartbeatManager", + "parse_interval", + "format_interval", + "load_heartbeat", + "save_heartbeat", + "migrate_heartbeat_to_session", + "HEARTBEAT_PROMPT_TEMPLATE", + "MIN_INTERVAL_SECONDS", + "POLL_SECONDS", +] diff --git a/tests/hermes_cli/test_heartbeat.py b/tests/hermes_cli/test_heartbeat.py new file mode 100644 index 000000000000..4d3496b00888 --- /dev/null +++ b/tests/hermes_cli/test_heartbeat.py @@ -0,0 +1,187 @@ +"""Tests for /heartbeat (hermes_cli/heartbeat.py).""" + +import time + +import pytest + +from hermes_cli.heartbeat import ( + HeartbeatManager, + HeartbeatState, + MIN_INTERVAL_SECONDS, + format_interval, + load_heartbeat, + migrate_heartbeat_to_session, + parse_interval, + save_heartbeat, +) + + +# ────────────────────────────────────────────────────────────────────── +# interval parsing +# ────────────────────────────────────────────────────────────────────── + + +@pytest.mark.parametrize( + "text,expected", + [ + ("10m", 600), + ("every 10m", 600), + ("2h", 7200), + ("every 2 hours", 7200), + ("1d", 86400), + ("90 minutes", 5400), + ("600s", 600), + ], +) +def test_parse_interval_valid(text, expected): + assert parse_interval(text) == expected + + +@pytest.mark.parametrize("text", ["", "banana", "check CI", "every", "m10"]) +def test_parse_interval_not_an_interval(text): + assert parse_interval(text) is None + + +def test_parse_interval_too_small_is_rejected(): + assert parse_interval("5s") == -1 + assert parse_interval("30s") == -1 + # Exactly the floor is allowed. + assert parse_interval(f"{MIN_INTERVAL_SECONDS}s") == MIN_INTERVAL_SECONDS + + +def test_format_interval(): + assert format_interval(600) == "10m" + assert format_interval(7200) == "2h" + assert format_interval(86400) == "1d" + assert format_interval(90) == "90s" + + +# ────────────────────────────────────────────────────────────────────── +# state + due logic +# ────────────────────────────────────────────────────────────────────── + + +def test_state_roundtrip(): + s = HeartbeatState(prompt="check CI", interval_seconds=600, created_at=time.time()) + loaded = HeartbeatState.from_json(s.to_json()) + assert loaded.prompt == "check CI" + assert loaded.interval_seconds == 600 + assert loaded.status == "active" + + +def test_is_due_anchors_on_created_then_last_fired(): + now = time.time() + s = HeartbeatState(prompt="p", interval_seconds=600, created_at=now) + assert s.is_due(now + 1) is False + assert s.is_due(now + 601) is True + s.last_fired_at = now + 601 + assert s.is_due(now + 700) is False + assert s.is_due(now + 1300) is True + + +def test_paused_never_due(): + now = time.time() + s = HeartbeatState(prompt="p", interval_seconds=60, created_at=now - 3600, status="paused") + assert s.is_due(now) is False + + +def test_render_prompt_contains_instruction_and_interval(): + s = HeartbeatState(prompt="check the deploy", interval_seconds=600) + rendered = s.render_prompt() + assert "check the deploy" in rendered + assert "10m" in rendered + assert "Heartbeat" in rendered + + +# ────────────────────────────────────────────────────────────────────── +# manager +# ────────────────────────────────────────────────────────────────────── + + +def test_manager_set_pause_resume_clear(): + mgr = HeartbeatManager(session_id="hb-lifecycle-sid") + state = mgr.set("watch CI", 600) + assert state.status == "active" + assert mgr.is_active() + + mgr.pause() + assert not mgr.is_active() + assert mgr.has_heartbeat() + + mgr.resume() + assert mgr.is_active() + + assert mgr.clear() is True + assert not mgr.has_heartbeat() + # Cleared rows don't resurrect on reload. + assert load_heartbeat("hb-lifecycle-sid") is None + + +def test_manager_rejects_bad_input(): + mgr = HeartbeatManager(session_id="hb-bad-sid") + with pytest.raises(ValueError): + mgr.set("", 600) + with pytest.raises(ValueError): + mgr.set("ok", 5) + + +def test_manager_persists_across_instances(): + mgr = HeartbeatManager(session_id="hb-persist-sid") + mgr.set("persisted prompt", 600) + again = HeartbeatManager(session_id="hb-persist-sid") + assert again.has_heartbeat() + assert again.state.prompt == "persisted prompt" + + +def test_due_prompt_fires_once_and_reanchors(): + mgr = HeartbeatManager(session_id="hb-due-sid") + mgr.set("tick", 600) + # Not due immediately after set. + assert mgr.due_prompt() is None + # Force due by rewinding the anchor. + mgr.state.created_at = time.time() - 700 + prompt = mgr.due_prompt() + assert prompt is not None and "tick" in prompt + assert mgr.state.fire_count == 1 + # Immediately after firing it re-anchors — not due again. + assert mgr.due_prompt() is None + + +def test_missed_ticks_coalesce(): + mgr = HeartbeatManager(session_id="hb-coalesce-sid") + mgr.set("tick", 600) + # Simulate 5 missed intervals: exactly ONE fire results. + mgr.state.created_at = time.time() - 600 * 5 - 10 + assert mgr.due_prompt() is not None + assert mgr.due_prompt() is None + assert mgr.state.fire_count == 1 + + +def test_resume_reanchors_instead_of_instant_fire(): + mgr = HeartbeatManager(session_id="hb-resume-sid") + mgr.set("tick", 600) + mgr.state.created_at = time.time() - 3600 + mgr.pause() + mgr.resume() + assert mgr.due_prompt() is None + + +# ────────────────────────────────────────────────────────────────────── +# compression migration +# ────────────────────────────────────────────────────────────────────── + + +def test_migrate_heartbeat_to_session(): + save_heartbeat( + "hb-parent-sid", + HeartbeatState(prompt="carry me", interval_seconds=600, created_at=time.time()), + ) + assert migrate_heartbeat_to_session("hb-parent-sid", "hb-child-sid") is True + child = load_heartbeat("hb-child-sid") + assert child is not None and child.prompt == "carry me" + assert load_heartbeat("hb-parent-sid") is None + + +def test_migrate_noop_without_source(): + assert migrate_heartbeat_to_session("hb-none-a", "hb-none-b") is False + assert migrate_heartbeat_to_session("same", "same") is False diff --git a/website/docs/reference/slash-commands.md b/website/docs/reference/slash-commands.md index 860d2c8d1a9e..4502a7982da6 100644 --- a/website/docs/reference/slash-commands.md +++ b/website/docs/reference/slash-commands.md @@ -53,6 +53,7 @@ Type `/` in the CLI to open the autocomplete menu. Built-in commands are case-in | `/steer ` | Inject a mid-run note that arrives at the agent **after the next tool call** — no interrupt, no new user turn. The text is appended to the last tool result's content once the current tool completes, giving the agent new context without breaking the current tool-calling loop. Use this to nudge direction mid-task (e.g. "focus on the auth module" while the agent is running tests). | | `/goal ` | Set a standing goal Hermes works toward across turns — our take on the Ralph loop. After each turn an auxiliary judge model decides whether the goal is done; if not, Hermes auto-continues. Subcommands: `/goal status`, `/goal pause`, `/goal resume`, `/goal clear`. Budget defaults to 20 turns (`goals.max_turns`); any real user message preempts the continuation loop, and state survives `/resume`. See [Persistent Goals](/user-guide/features/goals) for the full walkthrough. | | `/subgoal ` | Append a user-supplied criterion to the active goal mid-loop. The continuation prompt surfaces all subgoals to the agent verbatim, and the judge factors them into its DONE/CONTINUE verdict — so the goal isn't marked done until the original goal **and** every subgoal are met. Subcommands: `/subgoal` (list), `/subgoal remove `, `/subgoal clear`. Requires an active `/goal`. | +| `/heartbeat every ` (alias: `/hb`) | Set a recurring prompt that re-enters **this session** as a normal user turn whenever it's idle and the interval has elapsed (min 60s; missed ticks coalesce). Subcommands: `/heartbeat status`, `/heartbeat pause`, `/heartbeat resume`, `/heartbeat clear`. Session-scoped and in-process — use `hermes cron` for durable isolated schedules. See [Session Heartbeats](/user-guide/features/heartbeat). | | `/moa ` | Run a single prompt through the default [Mixture of Agents](/user-guide/features/mixture-of-agents) preset, then restore your current model. One-shot — does not change your session model. | | `/resume [name]` | Resume a previously-named session | | `/sessions` (TUI alias: `/switch`) | Classic CLI: browse and resume previous sessions in an interactive picker. TUI: open the live session switcher for currently open TUI sessions. Use `/sessions new` in the TUI to start another live session immediately. | @@ -249,6 +250,7 @@ The messaging gateway supports the following built-in commands inside Telegram, | `/steer ` | Inject a message after the next tool call without interrupting — the model picks it up on its next iteration rather than as a new turn. | | `/goal ` | Set a standing goal Hermes works toward across turns — our take on the Ralph loop. A judge model checks after each turn; if not done, Hermes auto-continues until it is, you pause/clear it, or the turn budget (default 20) is hit. Subcommands: `/goal status`, `/goal pause`, `/goal resume`, `/goal clear`. Safe to run mid-agent for status/pause/clear; setting a new goal requires `/stop` first. See [Persistent Goals](/user-guide/features/goals). | | `/subgoal ` | Append criteria to the active `/goal` mid-loop (`/subgoal`, `/subgoal remove `, `/subgoal clear`). | +| `/heartbeat every ` (alias: `/hb`) | Set a recurring prompt that re-enters this session when idle. Subcommands: `status`, `pause`, `resume`, `clear`. On Slack use `/hermes heartbeat …`. | | `/moa ` | Run one prompt through the default [Mixture of Agents](/user-guide/features/mixture-of-agents) preset, then restore the session model. | | `/branch [name]` (alias: `/fork`) | Branch the current session (explore a different path). | | `/agents` (alias: `/tasks`) | Show active agents and running tasks. | diff --git a/website/docs/user-guide/features/heartbeat.md b/website/docs/user-guide/features/heartbeat.md new file mode 100644 index 000000000000..69984a3d7fae --- /dev/null +++ b/website/docs/user-guide/features/heartbeat.md @@ -0,0 +1,65 @@ +--- +sidebar_position: 17 +title: "Session Heartbeats" +description: "A recurring prompt that re-enters your current session whenever it's idle — /heartbeat every 10m Check the deployment." +--- + +# Session Heartbeats (`/heartbeat`) + +`/heartbeat` gives the **current session** one recurring instruction. Whenever the session is idle and the interval has elapsed, the prompt fires as a normal user turn — same conversation, same context, same prompt cache. + +``` +/heartbeat every 10m Check the deployment and report meaningful changes +``` + +Inspired by Prime-Agent's `/heartbeat`. The Hermes adaptation keeps the strict message-flow invariants: the heartbeat is injected only between turns (never mid-run), as a plain user-role message. + +## Heartbeat vs cron: which one do I want? + +They look similar but serve different jobs: + +| | `/heartbeat` | [`hermes cron`](./cron) | +|---|---|---| +| Runs in | **This conversation** — full context, memory of the discussion | A fresh isolated session per tick | +| Survives process restart | State survives (SessionDB); firing resumes next time the session is driven | Yes — fully durable scheduler | +| How many | One per session | Unlimited jobs | +| Best for | "Keep an eye on X *in this thread* while we work" | Standing jobs, reports, watchdogs, deliveries | + +Rule of thumb: if the recurring prompt needs the conversation's context, use `/heartbeat`. If it's a self-contained job, use cron. + +## Commands + +| Command | What it does | +|---|---| +| `/heartbeat every ` | Set (or replace) the session's heartbeat. Intervals: `90s`, `10m`, `2h`, `1d` (minimum 60s). | +| `/heartbeat` or `/heartbeat status` | Show the heartbeat, its interval, and time to next fire. | +| `/heartbeat pause` | Stop firing without clearing. | +| `/heartbeat resume` | Resume (re-anchors the timer — no instant stale fire). | +| `/heartbeat clear` | Remove the heartbeat. | + +`/hb` is an alias. Works on the CLI and gateway platforms (on Slack, use `/hermes heartbeat …`). + +## Behavior details + +- **Idle-only.** A heartbeat never interrupts a running turn. If the agent is busy when the tick comes due, it fires at the next idle poll. +- **Missed ticks coalesce.** If the session was busy (or the process wasn't running) through several intervals, you get **one** heartbeat turn, not a backlog. The timer re-anchors on every fire. +- **User messages win.** A queued user message always takes priority; the heartbeat waits for the input queue to drain. +- **Cache-safe.** The injected prompt is an ordinary user message. No system-prompt mutation, no toolset change. +- **Persistence.** State lives in `SessionDB.state_meta` keyed by `heartbeat:` — it survives `/resume` and rides across context-compression session rotations. Firing requires the owning process (CLI session or gateway) to be running; for schedules that must survive anything, use cron. +- **Don't-invent-work guard.** The injected prompt tells the agent to reply briefly and stop when nothing meaningful changed, so an idle heartbeat doesn't generate busywork. + +## Example + +``` +You: /heartbeat every 15m Check whether the CI run for PR #1234 finished; summarize the result when it does + + ♥ Heartbeat set (every 15m): Check whether the CI run for PR #1234 finished; ... + +[15 minutes of you working on other things in the same session] + +Hermes: [Heartbeat — recurring instruction, fires every 15m] + 💻 gh pr checks 1234 (1.2s) + CI is still running (14/37 checks complete). Nothing to report yet. +``` + +When the answer stops changing, `/heartbeat clear` it — or let it keep watch. diff --git a/website/sidebars.ts b/website/sidebars.ts index 7b1e47098dfb..77bcf0f806bd 100644 --- a/website/sidebars.ts +++ b/website/sidebars.ts @@ -99,6 +99,7 @@ const sidebars: SidebarsConfig = { 'user-guide/features/kanban-tutorial', 'user-guide/features/kanban-worker-lanes', 'user-guide/features/goals', + 'user-guide/features/heartbeat', 'user-guide/features/code-execution', 'user-guide/features/hooks', 'user-guide/features/batch-processing',