Add heartbeat module

Author: Snehal-Reddy

src/agent.rs

@@ -197,6 +197,37 @@ pub async fn process_message(
     Ok(final_content)
 }
 
+// ---------------------------------------------------------------------------
+// Heartbeat agent entry point (one-shot, no session)
+// ---------------------------------------------------------------------------
+
+/// One-shot run for heartbeat: same context as `process_message` but with empty
+/// history and summary.  No session load or save.
+pub async fn process_heartbeat_message(
+    llm: &HttpProvider,
+    registry: &ToolRegistry,
+    workspace_path: &Path,
+    model: &str,
+    chat_id: &str,
+    user_message: &str,
+    tool_ctx: &ToolCtx,
+) -> Result<String, AgentError> {
+    let skills_summary = skills::build_skills_summary(workspace_path)?;
+    let tool_summaries = registry.summaries();
+    let today = crate::workspace::today_yyyymmdd();
+    let messages = build_messages(
+        workspace_path,
+        &[],
+        "",
+        user_message,
+        Some(chat_id),
+        &skills_summary,
+        &tool_summaries,
+        Some(&today),
+    );
+    run_agent_loop(llm, registry, messages, tool_ctx, model, MAX_ITERATIONS).await
+}
+
 // ---------------------------------------------------------------------------
 // Subagent runner (background; called by SubagentManager::spawn)
 // ---------------------------------------------------------------------------

src/heartbeat.rs

@@ -1 +1,202 @@
-//! Timer loop: read workspace/HEARTBEAT.md, push content as one-shot user message to agent.
+//! Timer loop: read workspace/HEARTBEAT.md, push one InboundMsg per task to agent.
+//!
+//! Each markdown bullet (`- `) in HEARTBEAT.md becomes its own agent run (one-shot, no session).
+//! Heartbeat pushes onto the same `inbound_tx` as Telegram and cron; the main loop branches on
+//! `channel == "heartbeat"` to call `process_heartbeat_message` instead of `process_message`.
+
+use std::path::{Path, PathBuf};
+use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::mpsc;
+
+use crate::telegram::InboundMsg;
+
+/// Parse markdown bullet tasks from HEARTBEAT.md content.
+///
+/// Lines whose trimmed form starts with `"- "` are tasks; everything else is ignored.
+/// Inner whitespace around the task text is trimmed; blank tasks are dropped.
+pub fn parse_tasks(content: &str) -> Vec<String> {
+    content
+        .lines()
+        .filter_map(|line| {
+            line.trim()
+                .strip_prefix("- ")
+                .map(|task| task.trim().to_string())
+        })
+        .filter(|task| !task.is_empty())
+        .collect()
+}
+
+/// Read and parse tasks from `workspace/HEARTBEAT.md`.
+///
+/// Returns an empty vec if the file does not exist or cannot be read.
+/// Sync I/O is fine: this is called at most once per N-minute tick.
+fn read_tasks(workspace: &Path) -> Vec<String> {
+    let path = workspace.join("HEARTBEAT.md");
+    if !path.exists() {
+        return vec![];
+    }
+    let content = std::fs::read_to_string(&path).unwrap_or_default();
+    parse_tasks(&content)
+}
+
+/// Spawn the heartbeat runner.
+///
+/// Every `interval_minutes` minutes: read `HEARTBEAT.md`, and for each task push one
+/// `InboundMsg { channel: "heartbeat" }` onto `inbound_tx`.  The main loop will call
+/// `process_heartbeat_message` once per message — N agent calls per tick (N = tasks).
+///
+/// `last_chat_id` is loaded on each tick to find the current active Telegram chat.
+/// If it is `0` (no user has messaged yet) the messages are still pushed; main.rs
+/// drops the reply in that case.
+///
+/// # Panics
+/// Panics if `interval_minutes == 0` (caller must check before calling).
+pub fn spawn_heartbeat_runner(
+    workspace: PathBuf,
+    interval_minutes: u64,
+    inbound_tx: mpsc::Sender<InboundMsg>,
+    last_chat_id: Arc<AtomicI64>,
+) -> tokio::task::JoinHandle<()> {
+    assert!(interval_minutes >= 1, "heartbeat interval_minutes must be >= 1");
+    tokio::spawn(async move {
+        let mut interval = tokio::time::interval(Duration::from_secs(interval_minutes * 60));
+        // Skip the immediately-firing first tick so the first real tick is one full interval out.
+        interval.tick().await;
+        loop {
+            interval.tick().await;
+            let tasks = read_tasks(&workspace);
+            if tasks.is_empty() {
+                continue;
+            }
+            let chat_id = last_chat_id.load(Ordering::Relaxed);
+            for task in tasks {
+                let msg = InboundMsg {
+                    chat_id,
+                    user_id: 0,
+                    text: format!("[Heartbeat Task] {task}"),
+                    channel: "heartbeat".to_string(),
+                };
+                if inbound_tx.send(msg).await.is_err() {
+                    // Receiver closed (main loop exited); nothing more to do.
+                    return;
+                }
+            }
+        }
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // --- parse_tasks ---
+
+    #[test]
+    fn parse_empty() {
+        assert!(parse_tasks("").is_empty());
+    }
+
+    #[test]
+    fn parse_single_bullet() {
+        assert_eq!(parse_tasks("- Check weather"), ["Check weather"]);
+    }
+
+    #[test]
+    fn parse_multiple_bullets() {
+        let tasks = parse_tasks("- First\n- Second\n- Third");
+        assert_eq!(tasks, ["First", "Second", "Third"]);
+    }
+
+    #[test]
+    fn parse_skips_non_bullet_lines() {
+        let content = "# Heartbeat\n\nProse.\n\n- Do thing\n<!-- comment -->\n- Another";
+        assert_eq!(parse_tasks(content), ["Do thing", "Another"]);
+    }
+
+    #[test]
+    fn parse_unicode_task() {
+        assert_eq!(parse_tasks("- こんにちは 🦀"), ["こんにちは 🦀"]);
+    }
+
+    #[test]
+    fn parse_strips_inner_whitespace() {
+        assert_eq!(parse_tasks("- \t  trim me  \t"), ["trim me"]);
+    }
+
+    #[test]
+    fn parse_skips_empty_bullets() {
+        // A bare "- " with nothing after it is dropped.
+        assert_eq!(parse_tasks("- \n- real task"), ["real task"]);
+    }
+
+    #[test]
+    fn parse_mixed_indentation_ignored() {
+        // Lines that do NOT start with "- " after trimming are skipped.
+        let tasks = parse_tasks("  - indented\n- normal");
+        assert_eq!(tasks, ["indented", "normal"]);
+    }
+
+    // --- read_tasks ---
+
+    #[test]
+    fn read_tasks_returns_empty_when_file_missing() {
+        let dir = std::env::temp_dir().join("icrab_hb_no_file_test");
+        // Ensure no HEARTBEAT.md exists in this temp dir.
+        let _ = std::fs::remove_file(dir.join("HEARTBEAT.md"));
+        assert!(read_tasks(&dir).is_empty());
+    }
+
+    #[test]
+    fn read_tasks_parses_file() {
+        let dir = std::env::temp_dir().join("icrab_hb_read_test");
+        std::fs::create_dir_all(&dir).unwrap();
+        std::fs::write(dir.join("HEARTBEAT.md"), "- Alpha\n- Beta\n").unwrap();
+        let tasks = read_tasks(&dir);
+        assert_eq!(tasks, ["Alpha", "Beta"]);
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+
+    // --- message format ---
+
+    #[tokio::test]
+    async fn messages_have_correct_format_and_channel() {
+        use std::sync::atomic::AtomicI64;
+        use tokio::sync::mpsc;
+
+        let dir = std::env::temp_dir().join("icrab_hb_msg_fmt_test");
+        std::fs::create_dir_all(&dir).unwrap();
+        std::fs::write(dir.join("HEARTBEAT.md"), "- Task A\n- Task B\n").unwrap();
+
+        let last_chat_id = Arc::new(AtomicI64::new(42));
+        let tasks = read_tasks(&dir);
+        assert_eq!(tasks.len(), 2);
+
+        let (tx, mut rx) = mpsc::channel(8);
+        let chat_id = last_chat_id.load(Ordering::Relaxed);
+        for task in &tasks {
+            tx.send(InboundMsg {
+                chat_id,
+                user_id: 0,
+                text: format!("[Heartbeat Task] {task}"),
+                channel: "heartbeat".to_string(),
+            })
+            .await
+            .unwrap();
+        }
+        drop(tx);
+
+        let a = rx.recv().await.unwrap();
+        assert_eq!(a.chat_id, 42);
+        assert_eq!(a.text, "[Heartbeat Task] Task A");
+        assert_eq!(a.channel, "heartbeat");
+        assert_eq!(a.user_id, 0);
+
+        let b = rx.recv().await.unwrap();
+        assert_eq!(b.text, "[Heartbeat Task] Task B");
+
+        let _ = std::fs::remove_dir_all(&dir);
+    }
+}

src/main.rs

@@ -3,6 +3,7 @@
 //! Single binary: runs Telegram poller + agent loop. Config: `~/.icrab/config.toml` or env.
 
 use std::path::PathBuf;
+use std::sync::atomic::{AtomicI64, Ordering};
 use std::sync::Arc;
 
 use tokio::sync::mpsc;
@@ -11,6 +12,7 @@ use icrab::agent;
 use icrab::agent::subagent_manager::SubagentManager;
 use icrab::config;
 use icrab::cron_runner;
+use icrab::heartbeat;
 use icrab::llm::HttpProvider;
 use icrab::telegram::{self, OutboundMsg};
 use icrab::tools;
@@ -83,9 +85,34 @@ async fn main() {
         60,
     );
     registry.register(CronTool::new(Arc::clone(&cron_store)));
+
+    // Track the last Telegram/cron chat_id so heartbeat replies go to the right chat.
+    let last_chat_id: Arc<AtomicI64> = Arc::new(AtomicI64::new(0));
+
+    // Spawn heartbeat if configured with interval_minutes >= 1.
+    let heartbeat_interval = cfg
+        .heartbeat
+        .as_ref()
+        .and_then(|h| h.interval_minutes)
+        .unwrap_or(0);
+    if heartbeat_interval >= 1 {
+        heartbeat::spawn_heartbeat_runner(
+            workspace.clone(),
+            heartbeat_interval,
+            inbound_tx.clone(),
+            Arc::clone(&last_chat_id),
+        );
+        eprintln!("heartbeat runner started (interval: {} min)", heartbeat_interval);
+    }
+
     drop(inbound_tx);
 
     while let Some(msg) = inbound_rx.recv().await {
+        // Update last_chat_id for non-heartbeat sources so replies go to the right place.
+        if msg.channel != "heartbeat" {
+            last_chat_id.store(msg.chat_id, Ordering::Relaxed);
+        }
+
         let tool_ctx = tools::ToolCtx {
             workspace: workspace.clone(),
             restrict_to_workspace: restrict,
@@ -94,23 +121,50 @@ async fn main() {
             outbound_tx: Some(Arc::new(outbound_tx.clone())),
         };
         let chat_id_str = msg.chat_id.to_string();
-        let reply = match agent::process_message(
-            &llm,
-            &registry,
-            &workspace,
-            model,
-            &chat_id_str,
-            &msg.text,
-            &tool_ctx,
-        )
-        .await
-        {
-            Ok(r) => r,
-            Err(e) => {
-                eprintln!("agent error: {}", e);
-                format!("Error: {}.", e)
+
+        let reply = if msg.channel == "heartbeat" {
+            match agent::process_heartbeat_message(
+                &llm,
+                &registry,
+                &workspace,
+                model,
+                &chat_id_str,
+                &msg.text,
+                &tool_ctx,
+            )
+            .await
+            {
+                Ok(r) => r,
+                Err(e) => {
+                    eprintln!("heartbeat agent error: {}", e);
+                    format!("Error: {}.", e)
+                }
+            }
+        } else {
+            match agent::process_message(
+                &llm,
+                &registry,
+                &workspace,
+                model,
+                &chat_id_str,
+                &msg.text,
+                &tool_ctx,
+            )
+            .await
+            {
+                Ok(r) => r,
+                Err(e) => {
+                    eprintln!("agent error: {}", e);
+                    format!("Error: {}.", e)
+                }
             }
         };
+
+        // Heartbeat with no known chat (chat_id == 0): no user has messaged yet, drop reply.
+        if msg.channel == "heartbeat" && msg.chat_id == 0 {
+            continue;
+        }
+
         let _ = outbound_tx
             .send(OutboundMsg {
                 chat_id: msg.chat_id,