From e58805b1ff3f2065dc0b68b36a6bf486784e92ee Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 5 Jul 2026 13:03:57 +0200 Subject: [PATCH] feat(broker): add message priority, operator messages surface first --- hive-c0re/src/stores/broker.rs | 72 ++++++++++++++++++++++++++++++---- 1 file changed, 65 insertions(+), 7 deletions(-) diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index c816029e..cfff76b0 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -22,10 +22,11 @@ CREATE TABLE IF NOT EXISTS messages ( body TEXT NOT NULL, sent_at INTEGER NOT NULL, delivered_at INTEGER, - in_reply_to INTEGER + in_reply_to INTEGER, + priority INTEGER NOT NULL DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_messages_undelivered - ON messages (recipient, id) WHERE delivered_at IS NULL; + ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL; CREATE TABLE IF NOT EXISTS reminders ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -222,9 +223,21 @@ impl Broker { pub fn send(&self, message: &Message) -> Result<()> { let conn = self.conn.lock().unwrap(); let now = now_unix(); + // Operator messages get elevated priority so they surface before + // queued wakes (bash completions, forge events, etc.) when the + // harness pops the next turn driver. All other senders stay at 0. + let priority: i64 = i64::from(message.from == "operator"); conn.execute( - "INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to) VALUES (?1, ?2, ?3, ?4, ?5)", - params![message.from, message.to, message.body, now, message.in_reply_to], + "INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to, priority) \ + VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + message.from, + message.to, + message.body, + now, + message.in_reply_to, + priority + ], )?; let row_id = conn.last_insert_rowid(); drop(conn); @@ -532,7 +545,10 @@ impl Broker { /// per message regardless of batch size. /// /// `max == 0` short-circuits to an empty vec (no DB hit); any - /// positive value caps the batch at `max`. FIFO ordering. + /// positive value caps the batch at `max`. Ordering: higher + /// priority first; FIFO within the same priority level. Operator + /// messages (priority = 1) therefore surface before queued bash + /// completions and forge events (priority = 0). /// /// Lock order: `inflight` FIRST, then `conn`. `requeue_inflight` /// and `ack_turn` follow the same order so a concurrent pop can't @@ -553,7 +569,7 @@ impl Broker { "SELECT id, sender, recipient, body, in_reply_to FROM messages WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL - ORDER BY id ASC + ORDER BY priority DESC, id ASC LIMIT ?2", )?; let rows: Vec<(i64, String, String, String, Option)> = stmt @@ -1118,6 +1134,22 @@ fn ensure_message_columns(conn: &Connection) -> Result<()> { // No backfill needed — existing messages simply have NULL here, // meaning "root of a new thread", which is correct. } + let has_priority: bool = conn + .prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = 'priority'")? + .exists([])?; + if !has_priority { + // Add column + rebuild the index with priority as the second key so + // recv_batch can serve operator messages ahead of queued wakes without + // a separate sort pass. + conn.execute_batch( + "ALTER TABLE messages ADD COLUMN priority INTEGER NOT NULL DEFAULT 0; + DROP INDEX IF EXISTS idx_messages_undelivered; + CREATE INDEX idx_messages_undelivered + ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;", + ) + .context("add messages.priority column and update delivery index")?; + // No backfill needed — all existing rows are correctly at priority 0. + } Ok(()) } @@ -1429,13 +1461,39 @@ mod tests { broker.send(&msg("matrix", "b", "matrix wake")).unwrap(); let batch = broker.recv_batch("b", 10).unwrap(); assert_eq!(batch.len(), 2); - // FIFO: mail was inserted first, comes out first. + // FIFO within same priority: mail was inserted first, comes out first. assert_eq!(batch[0].message.body, "mail"); assert_eq!(batch[1].message.body, "matrix wake"); assert!(batch[0].id > 0); assert!(batch[1].id > 0); } + /// Operator messages (priority = 1) surface before queued bash + /// completions and forge events (priority = 0) even when inserted + /// later, so an interrupt arrives before the task wake drives a turn. + #[test] + fn operator_message_jumps_queue_over_lower_priority() { + let h = open_broker(); + let broker = &h.broker; + // Low-priority wake arrives first. + broker + .send(&msg("bash-task-123", "agent", "task done")) + .unwrap(); + broker.send(&msg("forge", "agent", "new pr")).unwrap(); + // Operator interrupt arrives after both. + broker + .send(&msg("operator", "agent", "stop what you're doing")) + .unwrap(); + let batch = broker.recv_batch("agent", 10).unwrap(); + assert_eq!(batch.len(), 3); + // Operator message surfaces first despite arriving last. + assert_eq!(batch[0].message.body, "stop what you're doing"); + assert_eq!(batch[0].message.from, "operator"); + // Remaining two in FIFO order. + assert_eq!(batch[1].message.body, "task done"); + assert_eq!(batch[2].message.body, "new pr"); + } + /// `recv_batch` with no pending traffic returns an empty vec /// (the "(empty)" path), not an error. #[test]