reminder: add background scheduler loop - checks & delivers due reminders every 5s

This reverts commit 68a9b8575b1647643c87bd753767acabf96528c3.

hive-c0re/src/agent_server.rs

@@ -188,5 +188,34 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
                 },
             }
         }
+        AgentRequest::Remind {
+            message,
+            timing,
+            file_path,
+        } => {
+            use hive_sh4re::ReminderTiming;
+            let due_at = match timing {
+                ReminderTiming::InSeconds { seconds } => {
+                    std::time::SystemTime::now()
+                        .checked_add(std::time::Duration::from_secs(*seconds))
+                        .and_then(|t| {
+                            t.duration_since(std::time::UNIX_EPOCH)
+                                .ok()
+                                .and_then(|d| i64::try_from(d.as_secs()).ok())
+                        })
+                        .unwrap_or(0)
+                }
+                ReminderTiming::At { unix_timestamp } => *unix_timestamp,
+            };
+            match broker.store_reminder(agent, message, file_path.as_deref(), due_at) {
+                Ok(id) => {
+                    tracing::info!(%id, %agent, %due_at, "reminder scheduled");
+                    AgentResponse::Ok
+                }
+                Err(e) => AgentResponse::Err {
+                    message: format!("failed to store reminder: {e:#}"),
+                },
+            }
+        }
     }
 }

hive-c0re/src/broker.rs

@@ -22,6 +22,18 @@ CREATE TABLE IF NOT EXISTS messages (
 );
 CREATE INDEX IF NOT EXISTS idx_messages_undelivered
     ON messages (recipient, id) WHERE delivered_at IS NULL;
+
+CREATE TABLE IF NOT EXISTS reminders (
+    id        INTEGER PRIMARY KEY AUTOINCREMENT,
+    agent     TEXT    NOT NULL,
+    message   TEXT    NOT NULL,
+    file_path TEXT,
+    due_at    INTEGER NOT NULL,
+    created_at INTEGER NOT NULL,
+    sent_at   INTEGER
+);
+CREATE INDEX IF NOT EXISTS idx_reminders_due
+    ON reminders (agent, due_at) WHERE sent_at IS NULL;
 ";
 
 /// Capacity of the live event channel. Slow subscribers (e.g. an idle browser)
@@ -205,6 +217,51 @@ impl Broker {
         });
         Ok(Some(Message { from, to, body }))
     }
+
+    /// Store a new reminder. Returns the reminder id.
+    pub fn store_reminder(
+        &self,
+        agent: &str,
+        message: &str,
+        file_path: Option<&str>,
+        due_at: i64,
+    ) -> Result<i64> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "INSERT INTO reminders (agent, message, file_path, due_at, created_at) VALUES (?1, ?2, ?3, ?4, ?5)",
+            params![agent, message, file_path, due_at, now_unix()],
+        )?;
+        let id = conn.last_insert_rowid();
+        Ok(id)
+    }
+
+    /// Get all reminders for an agent that are due now or in the past.
+    /// Returns (id, message, file_path) tuples.
+    pub fn get_due_reminders(&self, agent: &str) -> Result<Vec<(i64, String, Option<String>)>> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare(
+            "SELECT id, message, file_path FROM reminders WHERE agent = ?1 AND due_at <= ?2 AND sent_at IS NULL ORDER BY due_at ASC"
+        )?;
+        let rows = stmt.query_map(params![agent, now_unix()], |row| {
+            Ok((
+                row.get::<_, i64>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, Option<String>>(2)?,
+            ))
+        })?;
+        rows.collect::<rusqlite::Result<Vec<_>>>()
+            .context("query reminders")
+    }
+
+    /// Mark a reminder as sent (delivered).
+    pub fn mark_reminder_sent(&self, id: i64) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "UPDATE reminders SET sent_at = ?1 WHERE id = ?2",
+            params![now_unix(), id],
+        )?;
+        Ok(())
+    }
 }
 
 fn now_unix() -> i64 {

hive-c0re/src/coordinator.rs

@@ -118,6 +118,10 @@ impl Coordinator {
         }
     }
 
+    pub fn list_agents(&self) -> Vec<String> {
+        self.agents.lock().unwrap().keys().cloned().collect()
+    }
+
     /// Mark an agent as in-progress (only one state per agent for now).
     pub fn set_transient(&self, name: &str, kind: TransientKind) {
         self.transient.lock().unwrap().insert(

hive-c0re/src/main.rs

@@ -157,6 +157,52 @@ async fn main() -> Result<()> {
             // when a previously-running container goes away without an
             // operator-initiated transient state.
             crash_watch::spawn(coord.clone());
+            // Reminder scheduler: checks for due reminders every 5 seconds,
+            // delivers them as inbox messages from "reminder".
+            let reminder_coord = coord.clone();
+            tokio::spawn(async move {
+                use hive_sh4re::Message;
+                loop {
+                    // Get all agents currently registered
+                    let agents = reminder_coord.list_agents();
+                    for agent in agents {
+                        match reminder_coord.broker.get_due_reminders(&agent) {
+                            Ok(reminders) => {
+                                for (id, message, _file_path) in reminders {
+                                    // Deliver as inbox message from "reminder"
+                                    if let Err(e) = reminder_coord.broker.send(&Message {
+                                        from: "reminder".to_owned(),
+                                        to: agent.clone(),
+                                        body: message.clone(),
+                                    }) {
+                                        tracing::warn!(
+                                            reminder_id = id,
+                                            %agent,
+                                            error = ?e,
+                                            "failed to deliver reminder"
+                                        );
+                                        continue;
+                                    }
+                                    // Mark as sent
+                                    if let Err(e) = reminder_coord.broker.mark_reminder_sent(id) {
+                                        tracing::warn!(
+                                            reminder_id = id,
+                                            error = ?e,
+                                            "failed to mark reminder sent"
+                                        );
+                                    }
+                                }
+                            }
+                            Err(e) => tracing::warn!(
+                                %agent,
+                                error = ?e,
+                                "failed to query due reminders"
+                            ),
+                        }
+                    }
+                    tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+                }
+            });
             let dash_coord = coord.clone();
             tokio::spawn(async move {
                 if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {

hive-sh4re/src/lib.rs

@@ -169,6 +169,17 @@ pub struct InboxRow {
     pub at: i64,
 }
 
+/// Reminder timing: either relative (wait N seconds) or absolute (at unix
+/// timestamp).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "timing_type", rename_all = "snake_case")]
+pub enum ReminderTiming {
+    /// Remind after this many seconds from now.
+    InSeconds { seconds: u64 },
+    /// Remind at this unix timestamp (seconds since epoch).
+    At { unix_timestamp: i64 },
+}
+
 /// Requests on a per-agent socket. The agent's identity is the socket
 /// it came in on; `Send.from` is filled in by the server, not the client.
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -218,6 +229,18 @@ pub enum AgentRequest {
         #[serde(default)]
         ttl_seconds: Option<u64>,
     },
+    /// Schedule a reminder message to be delivered to this agent at a
+    /// future time. The reminder lands in the agent's inbox as an auto-sent
+    /// message from `"reminder"`. Use for agent follow-ups (e.g. check task
+    /// status, retry failed operation). Message length is limited; pass
+    /// `file_path` to store in a file and get a path-reference message
+    /// instead.
+    Remind {
+        message: String,
+        timing: ReminderTiming,
+        #[serde(default)]
+        file_path: Option<String>,
+    },
 }
 
 /// Responses on a per-agent socket.