diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index a4cad214..564b150b 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -188,34 +188,5 @@ 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 20773263..72486bad 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -22,18 +22,6 @@ 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) @@ -217,51 +205,6 @@ 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 6f348cff..ad151a88 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -118,10 +118,6 @@ 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 5908b7b6..663f0ad4 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -157,52 +157,6 @@ 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 ed810516..2035fe4c 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -169,17 +169,6 @@ 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)] @@ -229,18 +218,6 @@ 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.