diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 564b150b..a4cad214 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -188,5 +188,34 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> }, } } + 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:#}"), + }, + } + } } } diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 72486bad..20773263 100644 --- a/hive-c0re/src/broker.rs +++ b/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 { + 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)>> { + 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>(2)?, + )) + })?; + rows.collect::>>() + .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 { diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index ad151a88..6f348cff 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -118,6 +118,10 @@ impl Coordinator { } } + pub fn list_agents(&self) -> Vec { + 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( diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 663f0ad4..5908b7b6 100644 --- a/hive-c0re/src/main.rs +++ b/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 { diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 2035fe4c..ed810516 100644 --- a/hive-sh4re/src/lib.rs +++ b/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, }, + /// 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, + }, } /// Responses on a per-agent socket.