feat(broker): add message priority, operator messages surface first

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)",
+            "INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to, priority) \
-            params![message.from, message.to, message.body, now, message.in_reply_to],
+             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<i64>)> = 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]