feat: /heartbeat — recurring session re-entry prompt fired when idle

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:

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:

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:

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 <prompt>` (also accepts `10m <prompt>`).
+        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 <interval> <prompt>  (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 <interval> <prompt> — 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).
 

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 <prompt>` (also accepts `10m <prompt>`).
+        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 <interval> <prompt>   (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 <interval> <prompt> — 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