//! Sqlite-backed message broker. Survives `hive-c0re` restart, and taps every
//! send/recv onto a broadcast channel so the dashboard can stream it.

use std::collections::{HashMap, HashSet, VecDeque};
use std::path::Path;
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};

use anyhow::{Context, Result};
use hive_sh4re::{InboxRow, Message};
use rusqlite::{Connection, OptionalExtension, params};
use serde::Serialize;
use tokio::sync::broadcast;

const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS messages (
    id           INTEGER PRIMARY KEY AUTOINCREMENT,
    sender       TEXT    NOT NULL,
    recipient    TEXT    NOT NULL,
    body         TEXT    NOT NULL,
    sent_at      INTEGER NOT NULL,
    delivered_at INTEGER,
    in_reply_to  INTEGER
);
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;

CREATE TABLE IF NOT EXISTS kv (
    key   TEXT PRIMARY KEY,
    value TEXT NOT NULL
);
";

/// Capacity of the live event channel. Slow subscribers (e.g. an idle browser)
/// may drop events past this; we send a `lagged` notice in their stream.
const EVENT_CHANNEL: usize = 256;

/// Row shape returned by [`Broker::get_due_reminders`]:
/// `(agent, reminder_id, message, file_path)`. Type alias keeps
/// `clippy::type_complexity` quiet and makes the scheduler call site
/// self-documenting.
pub type DueReminder = (String, i64, String, Option<String>);

/// A single message hand-off from broker to recipient. Carries the
/// broker's row id (so the harness can drive `ack_turn` later) and
/// the redelivery flag (so the harness can prepend the
/// "may already be handled" hint to the wake prompt). The
/// `Message` itself is identical to a pristine `Send` payload.
#[derive(Debug, Clone)]
pub struct Delivery {
    pub id: i64,
    pub redelivered: bool,
    pub message: Message,
}

/// Row shape for [`Broker::list_pending_reminders`], shipped on the
/// dashboard `/api/reminders` response.
#[derive(Debug, Clone, Serialize)]
pub struct PendingReminder {
    pub id: i64,
    pub agent: String,
    pub message: String,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub file_path: Option<String>,
    #[serde(with = "hive_sh4re::wire_time::iso")]
    pub due_at: i64,
    #[serde(with = "hive_sh4re::wire_time::iso")]
    pub created_at: i64,
    /// Most recent delivery failure for this row, if any. Cleared
    /// to NULL on operator retry. Surfaced inline in the dashboard
    /// so a stuck reminder doesn't just silently retry forever.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub last_error: Option<String>,
    /// Number of failed delivery attempts since the row was
    /// created or last retried. After `MAX_REMINDER_ATTEMPTS` the
    /// scheduler stops trying (the row stays in `pending` with the
    /// error so the operator can decide between retry + cancel).
    #[serde(default)]
    pub attempt_count: u32,
}

/// Stop retrying a row after this many consecutive failures. The
/// scheduler quits scheduling it until an operator explicitly
/// retries (which resets the counter) or cancels (which deletes
/// the row). Below the cap the existing 5s tick re-attempts each
/// time the row is due.
pub const MAX_REMINDER_ATTEMPTS: u32 = 5;

/// Intra-process broker event. `recv_blocking_batch` listens on the
/// same channel as the dashboard forwarder; the forwarder re-emits
/// each event as a `DashboardEvent` with a freshly-stamped seq from
/// the Coordinator. The broker itself doesn't stamp seqs — that's a
/// wire concern, not a storage concern.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "snake_case", tag = "kind")]
pub enum MessageEvent {
    Sent {
        /// Broker row id — used by the dashboard to track thread parents.
        id: i64,
        from: String,
        to: String,
        body: String,
        at: i64,
        in_reply_to: Option<i64>,
    },
    Delivered {
        /// Broker row id — used by the dashboard to track thread parents.
        id: i64,
        from: String,
        to: String,
        body: String,
        at: i64,
        in_reply_to: Option<i64>,
    },
    /// Transient wake signal — NOT persisted to sqlite. Wakes
    /// `recv_blocking_batch` for the target agent but is not stored,
    /// not re-delivered on restart, and not shown in message history.
    /// Used for bash task completion notifications.
    Ping {
        to: String,
        from: String,
        body: String,
    },
}

/// Per-recipient in-memory bookkeeping for the deliver-then-ack
/// flow. Source of truth is the DB columns `delivered_at` +
/// `acked_at`; the in-memory state here is purely an optimisation
/// (avoids scanning the messages table on `AckTurn`) plus the
/// redelivery-hint marker.
#[derive(Default)]
struct RecipientInflight {
    /// Message ids the broker has handed to this recipient since the
    /// last `AckTurn`. Drained on `ack_turn`, which then runs a
    /// single `UPDATE … WHERE id IN (…)` to set `acked_at`.
    unacked_ids: Vec<i64>,
    /// Message ids resurfaced by the most recent `requeue_inflight`
    /// call. The next `recv_batch` pop of any id in this set tags
    /// the response with `redelivered: true` so the harness can
    /// prepend the "may already be handled" hint to the wake prompt;
    /// successful pops drain the id from the set.
    requeued_ids: HashSet<i64>,
}

/// Hard cap on buffered transient pings per recipient. Pings are
/// ephemeral wakes (bash completions, matrix events); an agent drains
/// them on its next `recv`, so the buffer is normally near-empty. The
/// cap only bounds memory if a recipient stops recv'ing entirely (e.g.
/// a stopped container) — past it the oldest ping is dropped.
const MAX_PENDING_PINGS: usize = 256;

pub struct Broker {
    conn: Mutex<Connection>,
    events: broadcast::Sender<MessageEvent>,
    /// Per-recipient deliver/ack tracking. Lost on hive-c0re restart
    /// (harmless — the harness fires `RequeueInflight` on its own
    /// boot, which rebuilds the `requeued_ids` set from the DB and
    /// clears any stale `unacked_ids`).
    inflight: Mutex<HashMap<String, RecipientInflight>>,
    /// Per-recipient buffer of transient wake signals (`ping`) that
    /// haven't been consumed by a `recv` yet. The broadcast channel only
    /// reaches a `recv_blocking_batch` that is *currently parked*; a ping
    /// fired while the harness is mid-turn (no live subscriber) would
    /// otherwise be lost. Buffering here makes a transient wake reliable
    /// — the next `recv` drains it — without sqlite persistence or
    /// redelivery-on-restart (the buffer is in-memory, dropped on
    /// hive-c0re restart, which is fine: a wake that old is stale).
    pending_pings: Mutex<HashMap<String, VecDeque<(String, String)>>>,
}

impl Broker {
    pub fn open(path: &Path) -> Result<Self> {
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)
                .with_context(|| format!("create db parent {}", parent.display()))?;
        }
        let conn =
            Connection::open(path).with_context(|| format!("open broker db {}", path.display()))?;
        conn.execute_batch(SCHEMA).context("apply broker schema")?;
        ensure_message_columns(&conn).context("migrate messages columns")?;
        ensure_reminder_columns(&conn).context("migrate reminders columns")?;
        let (events, _) = broadcast::channel(EVENT_CHANNEL);
        Ok(Self {
            conn: Mutex::new(conn),
            events,
            inflight: Mutex::new(HashMap::new()),
            pending_pings: Mutex::new(HashMap::new()),
        })
    }

    pub fn subscribe(&self) -> broadcast::Receiver<MessageEvent> {
        self.events.subscribe()
    }

    /// Set a small persistent key/value pair (upsert). The `kv` table is a
    /// general single-value store for state that must survive a hive-c0re
    /// restart but isn't worth a dedicated table — currently just the
    /// `hivectl start` running-agents snapshot (see
    /// `Coordinator::set_last_stopped_running`).
    ///
    /// # Errors
    ///
    /// Returns an error if the sqlite upsert fails.
    pub fn kv_set(&self, key: &str, value: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "INSERT INTO kv (key, value) VALUES (?1, ?2)
             ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            params![key, value],
        )?;
        Ok(())
    }

    /// Read a persistent key/value pair. `None` when the key is absent.
    ///
    /// # Errors
    ///
    /// Returns an error if the sqlite query fails.
    pub fn kv_get(&self, key: &str) -> Result<Option<String>> {
        let conn = self.conn.lock().unwrap();
        let value: Option<String> = conn
            .query_row("SELECT value FROM kv WHERE key = ?1", params![key], |row| {
                row.get(0)
            })
            .optional()?;
        Ok(value)
    }

    /// Delete a persistent key/value pair. Idempotent — deleting an absent
    /// key is a no-op.
    ///
    /// # Errors
    ///
    /// Returns an error if the sqlite delete fails.
    pub fn kv_delete(&self, key: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute("DELETE FROM kv WHERE key = ?1", params![key])?;
        Ok(())
    }

    pub fn send(&self, message: &Message) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        let now = now_unix();
        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],
        )?;
        let row_id = conn.last_insert_rowid();
        drop(conn);
        let _ = self.events.send(MessageEvent::Sent {
            id: row_id,
            from: message.from.clone(),
            to: message.to.clone(),
            body: message.body.clone(),
            at: now,
            in_reply_to: message.in_reply_to,
        });
        Ok(())
    }

    /// Deliver a transient wake signal to `to` without writing to sqlite.
    /// The signal wakes the target agent's next `recv_blocking_batch` but is
    /// not persisted, not redelivered on restart, and not shown in message
    /// history. Use for ephemeral notifications (bash task completions,
    /// matrix events) where persistence would cause duplicate delivery.
    ///
    /// The ping is buffered in `pending_pings` *before* the broadcast so a
    /// `recv` that wasn't parked at fire time still drains it on its next
    /// call — the broadcast alone reaches only a currently-parked receiver.
    pub fn ping(&self, to: &str, from: &str, body: &str) {
        {
            let mut pending = self.pending_pings.lock().unwrap();
            let queue = pending.entry(to.to_owned()).or_default();
            queue.push_back((from.to_owned(), body.to_owned()));
            // Bound memory if the recipient never recv's (stopped container):
            // drop the oldest so the most recent wakes survive.
            while queue.len() > MAX_PENDING_PINGS {
                queue.pop_front();
            }
        }
        let _ = self.events.send(MessageEvent::Ping {
            to: to.to_owned(),
            from: from.to_owned(),
            body: body.to_owned(),
        });
    }

    /// Drain up to `max` buffered transient pings for `recipient`, mapping
    /// each to a `Delivery` with the sentinel `id = 0` (never pushed to
    /// `unacked_ids`, so `ack_turn` ignores it). FIFO; leaves any remainder
    /// buffered for the next call.
    fn drain_pings(&self, recipient: &str, max: usize) -> Vec<Delivery> {
        if max == 0 {
            return Vec::new();
        }
        let mut pending = self.pending_pings.lock().unwrap();
        let Some(queue) = pending.get_mut(recipient) else {
            return Vec::new();
        };
        let take = queue.len().min(max);
        let drained: Vec<Delivery> = queue
            .drain(..take)
            .map(|(from, body)| Delivery {
                id: 0,
                redelivered: false,
                message: Message {
                    from,
                    to: recipient.to_owned(),
                    body,
                    in_reply_to: None,
                },
            })
            .collect();
        if queue.is_empty() {
            pending.remove(recipient);
        }
        drained
    }

    /// Collect a batch for `recipient`: buffered transient pings first
    /// (they're wakes — surface them ahead of queued mail), then sqlite
    /// messages up to the remaining budget. Shared by the immediate check
    /// and the post-wake path of `recv_blocking_batch`.
    fn collect_batch(&self, recipient: &str, max: usize) -> Result<Vec<Delivery>> {
        let mut batch = self.drain_pings(recipient, max);
        if batch.len() < max {
            batch.extend(self.recv_batch(recipient, max - batch.len())?);
        }
        Ok(batch)
    }

    /// Unread (unacked) messages addressed to `recipient`, newest-first.
    /// Filters to `acked_at IS NULL` so the agent inbox view clears after
    /// "mark all read" — mirroring exactly what `mark_all_read` will drain.
    pub fn recent_for(&self, recipient: &str, limit: u64) -> Result<Vec<InboxRow>> {
        let conn = self.conn.lock().unwrap();
        let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
        let mut stmt = conn.prepare(
            "SELECT id, sender, body, sent_at, in_reply_to
               FROM messages
              WHERE recipient = ?1
                AND acked_at IS NULL
              ORDER BY id DESC
              LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![recipient, limit_i], |row| {
            Ok(InboxRow {
                id: row.get(0)?,
                from: row.get(1)?,
                body: row.get(2)?,
                at: row.get(3)?,
                in_reply_to: row.get(4)?,
            })
        })?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    /// Latest `limit` messages across every recipient, newest-first.
    /// Backs the dashboard's message-flow backfill so a reload doesn't
    /// blank the operator's view of recent traffic. Returns each row as
    /// a [`MessageEvent::Sent`] so the dashboard's live renderer (which
    /// already speaks `MessageEvent`) can replay history through the
    /// same code path. We don't synthesise `Delivered` events here —
    /// the recv-side acks live in a different table column and would
    /// double-render on backfill; the live stream picks them up
    /// immediately on the first new `recv`.
    pub fn recent_all(&self, limit: u64) -> Result<Vec<MessageEvent>> {
        let conn = self.conn.lock().unwrap();
        let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
        let mut stmt = conn.prepare(
            "SELECT id, sender, recipient, body, sent_at, in_reply_to
               FROM messages
              ORDER BY id DESC
              LIMIT ?1",
        )?;
        let rows = stmt.query_map(params![limit_i], |row| {
            Ok(MessageEvent::Sent {
                id: row.get(0)?,
                from: row.get(1)?,
                to: row.get(2)?,
                body: row.get(3)?,
                at: row.get(4)?,
                in_reply_to: row.get(5)?,
            })
        })?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    /// Unacknowledged messages addressed to `recipient`, newest-first.
    /// Backs the dashboard's operator inbox: the operator never
    /// `recv`s over an agent socket, so messages to `"operator"` sit in
    /// the broker with `acked_at IS NULL` until the operator hits "mark
    /// all read" (which calls [`Broker::mark_all_read`]). This read
    /// mirrors that filter EXACTLY — `recipient = ?1 AND acked_at IS
    /// NULL`, with no `delivered_at` condition — so everything listed
    /// here is precisely what `mark_all_read` will clear (operator rows
    /// may never get `delivered_at` set). Returned as
    /// [`MessageEvent::Sent`] so the dashboard reuses its live renderer.
    ///
    /// # Errors
    ///
    /// Returns `Err` if the `SQLite` prepare or query fails.
    pub fn unread_for_recipient(&self, recipient: &str, limit: u64) -> Result<Vec<MessageEvent>> {
        let conn = self.conn.lock().unwrap();
        let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
        let mut stmt = conn.prepare(
            "SELECT id, sender, recipient, body, sent_at, in_reply_to
               FROM messages
              WHERE recipient = ?1 AND acked_at IS NULL
              ORDER BY id DESC
              LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![recipient, limit_i], |row| {
            Ok(MessageEvent::Sent {
                id: row.get(0)?,
                from: row.get(1)?,
                to: row.get(2)?,
                body: row.get(3)?,
                at: row.get(4)?,
                in_reply_to: row.get(5)?,
            })
        })?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .map_err(Into::into)
    }

    /// Number of undelivered messages addressed to `recipient`. Non-mutating
    /// — used by the harness to surface "N unread" in tool-result status
    /// lines without popping the queue.
    pub fn count_pending(&self, recipient: &str) -> Result<u64> {
        let conn = self.conn.lock().unwrap();
        // Skip rows closed by `ack_until` while still pending — they
        // will never pop, so counting them would show phantom unread.
        let n: i64 = conn.query_row(
            "SELECT COUNT(*) FROM messages
              WHERE recipient = ?1 AND delivered_at IS NULL AND acked_at IS NULL",
            params![recipient],
            |row| row.get(0),
        )?;
        Ok(u64::try_from(n.max(0)).unwrap_or(0))
    }

    /// Returns true when the recipient already has at least one
    /// undelivered message from `sender` with exactly `body` in the
    /// broker. Used by the scheduler to skip re-delivery of the same
    /// scheduled prompt without blocking distinct schedules whose
    /// bodies differ.
    pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result<bool> {
        let conn = self.conn.lock().unwrap();
        // An `ack_until`-closed pending row is dead — it must not
        // suppress a fresh scheduled delivery of the same body.
        let n: i64 = conn.query_row(
            "SELECT COUNT(*) FROM messages
              WHERE recipient = ?1 AND sender = ?2 AND body = ?3
                AND delivered_at IS NULL AND acked_at IS NULL",
            params![recipient, sender, body],
            |row| row.get(0),
        )?;
        Ok(n > 0)
    }

    /// Send a "your parent changed from X to Y" notification to `child`,
    /// coalescing with any existing undelivered one so that multiple moves
    /// while the agent is offline collapse into a single message spanning
    /// the full arc (e.g. A→B then B→C becomes "your parent changed from A to C").
    ///
    /// If an undelivered system reparent notification for `child` already
    /// exists, its body is updated in-place preserving the original "from"
    /// label. If none exists, a fresh message is inserted with `old_label`
    /// as the source.
    pub fn send_coalescing_reparent(
        &self,
        child: &str,
        old_label: &str,
        new_label: &str,
    ) -> Result<()> {
        const PREFIX: &str = "your parent changed from ";
        const SEPARATOR: &str = " to ";
        let conn = self.conn.lock().unwrap();
        let now = now_unix();
        let existing: Option<(i64, String)> = conn
            .query_row(
                "SELECT id, body FROM messages
                  WHERE recipient = ?1
                    AND sender = ?2
                    AND body LIKE ?3
                    AND delivered_at IS NULL
                    AND acked_at IS NULL
                  LIMIT 1",
                params![child, hive_sh4re::SYSTEM_SENDER, format!("{PREFIX}%")],
                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
            )
            .optional()?;
        if let Some((row_id, old_body)) = existing {
            // Preserve the original "from" label from the earlier notification.
            let original_from = old_body
                .strip_prefix(PREFIX)
                .and_then(|rest| rest.split(SEPARATOR).next())
                .unwrap_or(old_label);
            let new_body = format!("{PREFIX}{original_from}{SEPARATOR}{new_label}");
            conn.execute(
                "UPDATE messages SET body = ?1, sent_at = ?2 WHERE id = ?3",
                params![new_body, now, row_id],
            )?;
            drop(conn);
            let _ = self.events.send(MessageEvent::Sent {
                id: row_id,
                from: hive_sh4re::SYSTEM_SENDER.to_owned(),
                to: child.to_owned(),
                body: new_body,
                at: now,
                in_reply_to: None,
            });
        } else {
            let body = format!("{PREFIX}{old_label}{SEPARATOR}{new_label}");
            conn.execute(
                "INSERT INTO messages (sender, recipient, body, sent_at, in_reply_to)
                  VALUES (?1, ?2, ?3, ?4, NULL)",
                params![hive_sh4re::SYSTEM_SENDER, child, body, now],
            )?;
            let row_id = conn.last_insert_rowid();
            drop(conn);
            let _ = self.events.send(MessageEvent::Sent {
                id: row_id,
                from: hive_sh4re::SYSTEM_SENDER.to_owned(),
                to: child.to_owned(),
                body,
                at: now,
                in_reply_to: None,
            });
        }
        Ok(())
    }

    /// Long-poll variant of `recv_batch`: returns immediately if any
    /// row is pending (popping up to `max`); otherwise waits up to
    /// `timeout` for the broker to emit a `Sent { to: recipient }`
    /// event and re-tries the pop. Lets agents react to new mail
    /// without polling their socket on a fixed interval AND lets a
    /// single round-trip drain a burst of messages.
    ///
    /// **Subscribe-before-check order matters.** If we polled the
    /// sqlite row first and only then called `subscribe()`, a
    /// concurrent `send` landing in that window would commit +
    /// broadcast its event *before* our receiver existed — and we'd
    /// then sit on the long-poll until the timeout (or another,
    /// unrelated send) fired. That looked externally like "the agent
    /// processed one wake then went deaf until the operator poked it
    /// again". Subscribing first guarantees any post-subscribe send
    /// notifies us; the redundant `recv_batch()` catches the message
    /// either way.
    ///
    /// `max == 0` returns an empty vec without subscribing or waiting.
    pub async fn recv_blocking_batch(
        &self,
        recipient: &str,
        timeout: std::time::Duration,
        max: usize,
    ) -> Result<Vec<Delivery>> {
        if max == 0 {
            return Ok(Vec::new());
        }
        let mut rx = self.subscribe();
        // Immediate check: buffered pings (incl. any fired while no recv was
        // parked) + queued sqlite mail.
        let batch = self.collect_batch(recipient, max)?;
        if !batch.is_empty() {
            return Ok(batch);
        }
        let deadline = tokio::time::Instant::now() + timeout;
        loop {
            let Some(remaining) = deadline.checked_duration_since(tokio::time::Instant::now())
            else {
                return Ok(Vec::new());
            };
            match tokio::time::timeout(remaining, rx.recv()).await {
                Err(_) => return Ok(Vec::new()),
                // Channel lagged or closed — fall back to a direct collect
                // (in case we missed our notification while behind).
                Ok(Err(_)) => return self.collect_batch(recipient, max),
                // A relevant event landed (real message or transient ping).
                // Re-collect from the buffers; the ping payload on the event
                // is ignored — `drain_pings` is the source of truth, so a
                // ping can't be double-delivered.
                Ok(Ok(MessageEvent::Sent { to, .. } | MessageEvent::Ping { to, .. }))
                    if to == recipient =>
                {
                    let batch = self.collect_batch(recipient, max)?;
                    if !batch.is_empty() {
                        return Ok(batch);
                    }
                    // Lost a race (concurrent recv drained it). Keep waiting.
                }
                Ok(Ok(_)) => {}
            }
        }
    }

    /// Delete fully-acked messages older than `older_than_secs`.
    /// Unacked rows (delivered but not yet acknowledged by a clean
    /// turn-end, plus undelivered rows) are always kept regardless of
    /// age — the former because they're recoverable via
    /// `requeue_inflight`, the latter because they're still in flight
    /// from the broker's POV. Returns the number of rows removed.
    pub fn vacuum_delivered(&self, older_than_secs: i64) -> Result<u64> {
        let cutoff = now_unix() - older_than_secs;
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "DELETE FROM messages
              WHERE acked_at IS NOT NULL
                AND acked_at < ?1",
            params![cutoff],
        )?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Pop up to `max` pending messages for `recipient` in one
    /// round-trip. Every popped row is marked `delivered_at = NOW`,
    /// pushed onto the per-recipient `unacked_ids` list (so the next
    /// `ack_turn` closes them out), and tagged with
    /// `redelivered = true` if it was resurfaced by the most recent
    /// `requeue_inflight`. Emits one `MessageEvent::Delivered` per
    /// popped row so the dashboard forwarder stream sees one event
    /// 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.
    ///
    /// Lock order: `inflight` FIRST, then `conn`. `requeue_inflight`
    /// and `ack_turn` follow the same order so a concurrent pop can't
    /// race the requeue's DB update vs in-memory populate and miss
    /// the redelivered tag.
    pub fn recv_batch(&self, recipient: &str, max: usize) -> Result<Vec<Delivery>> {
        if max == 0 {
            return Ok(Vec::new());
        }
        // Same lock order as `recv` / `ack_turn` / `requeue_inflight`.
        let mut inflight = self.inflight.lock().unwrap();
        let conn = self.conn.lock().unwrap();
        let max_i = i64::try_from(max).unwrap_or(i64::MAX);
        // `acked_at IS NULL` matters for rows closed by `ack_until`
        // while still pending (never delivered): they carry an ack but
        // no `delivered_at`, and must not pop.
        let mut stmt = conn.prepare(
            "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
              LIMIT ?2",
        )?;
        let rows: Vec<(i64, String, String, String, Option<i64>)> = stmt
            .query_map(params![recipient, max_i], |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                ))
            })?
            .collect::<rusqlite::Result<_>>()?;
        drop(stmt);
        if rows.is_empty() {
            return Ok(Vec::new());
        }
        // Stamp all popped rows in a single UPDATE — under the broker
        // mutex, well within sqlite's 999-param default.
        let now = now_unix();
        let ids: Vec<i64> = rows.iter().map(|(id, _, _, _, _)| *id).collect();
        let placeholders = std::iter::repeat_n("?", ids.len())
            .collect::<Vec<_>>()
            .join(",");
        let sql = format!("UPDATE messages SET delivered_at = ? WHERE id IN ({placeholders})");
        let mut params_vec: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(ids.len() + 1);
        params_vec.push(&now);
        for id in &ids {
            params_vec.push(id);
        }
        conn.execute(&sql, params_vec.as_slice())?;
        drop(conn);
        // Bookkeeping + assemble the Delivery list. Per-row
        // `requeued_ids` lookup runs once per pop, same as `recv`.
        let slot = inflight.entry(recipient.to_owned()).or_default();
        let mut deliveries = Vec::with_capacity(rows.len());
        for (id, from, to, body, in_reply_to) in rows {
            slot.unacked_ids.push(id);
            let redelivered = slot.requeued_ids.remove(&id);
            deliveries.push(Delivery {
                id,
                redelivered,
                message: Message {
                    from,
                    to,
                    body,
                    in_reply_to,
                },
            });
        }
        drop(inflight);
        // Mirror the per-row Delivered emit `recv` does so the
        // dashboard forwarder sees one event per message regardless of
        // which surface the harness used.
        for d in &deliveries {
            let _ = self.events.send(MessageEvent::Delivered {
                id: d.id,
                from: d.message.from.clone(),
                to: d.message.to.clone(),
                body: d.message.body.clone(),
                at: now,
                in_reply_to: d.message.in_reply_to,
            });
        }
        Ok(deliveries)
    }

    /// Drain the per-recipient unacked-id list and mark every row
    /// `acked_at = NOW`. Fired by the harness after `TurnOutcome::Ok`.
    /// Returns the number of rows acked (zero is normal — claude
    /// may have not called recv during the turn). Tolerant of ids
    /// that no longer exist in the DB (vacuumed, manually deleted)
    /// — `UPDATE … WHERE id IN (…)` simply matches zero rows.
    pub fn ack_turn(&self, recipient: &str) -> Result<u64> {
        // Same lock order as `recv` and `requeue_inflight`.
        let mut inflight = self.inflight.lock().unwrap();
        let ids: Vec<i64> = inflight
            .get_mut(recipient)
            .map(|s| std::mem::take(&mut s.unacked_ids))
            .unwrap_or_default();
        if ids.is_empty() {
            return Ok(0);
        }
        let now = now_unix();
        let conn = self.conn.lock().unwrap();
        // Bind every id explicitly. Caps in the hundreds in the worst
        // case (a single very chatty turn); well under sqlite's 999
        // default param limit and we're already serialising on the
        // broker mutex.
        let placeholders = std::iter::repeat_n("?", ids.len())
            .collect::<Vec<_>>()
            .join(",");
        let sql = format!("UPDATE messages SET acked_at = ? WHERE id IN ({placeholders})");
        let mut params_vec: Vec<&dyn rusqlite::ToSql> = Vec::with_capacity(ids.len() + 1);
        params_vec.push(&now);
        for id in &ids {
            params_vec.push(id);
        }
        let n = conn.execute(&sql, params_vec.as_slice())?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Bulk-ack every message addressed to `recipient` with row id
    /// `<= up_to` that isn't acked yet — pending AND delivered rows
    /// alike. The agent-facing triage tool behind `AckUntil`: after a
    /// redelivered flood the agent acks everything up to the highest id
    /// it has seen instead of re-popping each row. Recipient-scoped by
    /// the WHERE clause, so an agent can never touch another agent's
    /// rows. Also drains the in-memory `unacked_ids` / `requeued_ids`
    /// bookkeeping below the cutoff so a later `ack_turn` doesn't
    /// re-update rows this call already closed and a stale redelivery
    /// tag doesn't outlive its row. Returns the number of rows newly
    /// acked.
    ///
    /// # Errors
    ///
    /// Propagates sqlite errors from the `UPDATE`.
    pub fn ack_until(&self, recipient: &str, up_to: i64) -> Result<u64> {
        // Same lock order as `recv` / `ack_turn` / `requeue_inflight`:
        // `inflight` FIRST, then `conn`.
        let mut inflight = self.inflight.lock().unwrap();
        if let Some(state) = inflight.get_mut(recipient) {
            state.unacked_ids.retain(|&id| id > up_to);
            state.requeued_ids.retain(|&id| id > up_to);
        }
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "UPDATE messages SET acked_at = ?1
              WHERE recipient = ?2 AND id <= ?3 AND acked_at IS NULL",
            params![now_unix(), recipient, up_to],
        )?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Resurface every message the broker previously handed to this
    /// recipient that never got `acked_at` set. Used by the harness at
    /// boot to recover from the crashed-mid-turn / OOM-killed /
    /// container-restarted cases. Three steps:
    ///
    /// 1. Clear any stale in-memory state for this recipient (the
    ///    previous harness session's `unacked_ids` are irrelevant —
    ///    the new session will repopulate from fresh pops).
    /// 2. Find every row where `recipient = me`, `delivered_at IS NOT
    ///    NULL`, `acked_at IS NULL`. Reset `delivered_at = NULL` so
    ///    the next `Recv` pops them again.
    /// 3. Remember each id in the per-recipient `requeued_ids` set so
    ///    the next pop tags the response with `redelivered: true`.
    ///
    /// Returns the number of rows requeued. Safe to call when there's
    /// nothing in flight (returns 0). Safe to call multiple times
    /// (idempotent — the second call finds nothing because the rows
    /// are now back in the pending state).
    pub fn requeue_inflight(&self, recipient: &str) -> Result<u64> {
        // Hold inflight + conn together so a concurrent `recv` can't
        // pop a just-requeued row between our DB update and our
        // in-memory populate and miss the redelivered tag.
        let mut inflight = self.inflight.lock().unwrap();
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id FROM messages
              WHERE recipient = ?1
                AND delivered_at IS NOT NULL
                AND acked_at IS NULL",
        )?;
        let ids: Vec<i64> = stmt
            .query_map(params![recipient], |row| row.get(0))?
            .collect::<rusqlite::Result<_>>()?;
        drop(stmt);
        if !ids.is_empty() {
            let placeholders = std::iter::repeat_n("?", ids.len())
                .collect::<Vec<_>>()
                .join(",");
            let sql =
                format!("UPDATE messages SET delivered_at = NULL WHERE id IN ({placeholders})");
            let params_vec: Vec<&dyn rusqlite::ToSql> =
                ids.iter().map(|id| id as &dyn rusqlite::ToSql).collect();
            conn.execute(&sql, params_vec.as_slice())?;
        }
        let slot = inflight.entry(recipient.to_owned()).or_default();
        slot.unacked_ids.clear();
        slot.requeued_ids.clear();
        slot.requeued_ids.extend(ids.iter().copied());
        Ok(u64::try_from(ids.len()).unwrap_or(0))
    }

    /// Operator-driven "clear the inbox": mark every message addressed
    /// to `recipient` as acked. Backfills `delivered_at = NOW` for any
    /// row that was still pending (undelivered), so the row doesn't
    /// become impossible to vacuum later — `vacuum_delivered` requires
    /// both timestamps set. Also clears the in-memory inflight state
    /// for the recipient so a subsequent `ack_turn` doesn't try to
    /// re-mark ids that are already acked. Returns the number of rows
    /// affected (zero is normal — inbox already empty).
    ///
    /// Distinct from `ack_turn` (which acks only the per-turn unacked
    /// ids the harness pulled via `recv_batch`) and `requeue_inflight`
    /// (which puts inflight-but-unacked rows BACK on the queue). This
    /// is the destructive "drain everything for this agent" path the
    /// dashboard surfaces as the side-panel "mark all read" button.
    /// Backs `POST /api/agent/{name}/mark-all-read`.
    pub fn mark_all_read(&self, recipient: &str) -> Result<u64> {
        let mut inflight = self.inflight.lock().unwrap();
        let conn = self.conn.lock().unwrap();
        let now = now_unix();
        // Two-axis update in one statement: set acked_at on every
        // row for the recipient that doesn't have it yet, AND backfill
        // delivered_at if it was NULL so the row is fully consumed
        // (vacuum_delivered's `acked_at IS NOT NULL AND acked_at < ?`
        // predicate then collects it on the normal hourly sweep).
        let n = conn.execute(
            "UPDATE messages
                SET delivered_at = COALESCE(delivered_at, ?1),
                    acked_at = ?1
              WHERE recipient = ?2
                AND acked_at IS NULL",
            params![now, recipient],
        )?;
        // Drop in-memory inflight bookkeeping for this recipient: the
        // ids we just acked might still be in `unacked_ids` from a
        // prior `recv_batch`; leaving them would cause the next
        // `ack_turn` to re-issue an UPDATE against rows that no longer
        // need it (correct but wasteful) and the requeue path would
        // see stale ids. Cleanest to reset.
        if let Some(slot) = inflight.get_mut(recipient) {
            slot.unacked_ids.clear();
            slot.requeued_ids.clear();
        }
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// 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)
    }

    /// Every reminder still pending delivery, newest-first. Used by the
    /// dashboard's reminders pane so the operator can see what's queued
    /// + cancel rows that are no longer wanted.
    pub fn list_pending_reminders(&self) -> Result<Vec<PendingReminder>> {
        let conn = self.conn.lock().unwrap();
        let mut stmt = conn.prepare(
            "SELECT id, agent, message, file_path, due_at, created_at, \
                    last_error, attempt_count \
               FROM reminders \
              WHERE sent_at IS NULL \
              ORDER BY due_at ASC",
        )?;
        let rows = stmt.query_map([], |row| {
            let attempts: i64 = row.get(7)?;
            Ok(PendingReminder {
                id: row.get(0)?,
                agent: row.get(1)?,
                message: row.get(2)?,
                file_path: row.get(3)?,
                due_at: row.get(4)?,
                created_at: row.get(5)?,
                last_error: row.get(6)?,
                attempt_count: u32::try_from(attempts).unwrap_or(0),
            })
        })?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .context("list pending reminders")
    }

    /// Mark a delivery attempt as failed: bump `attempt_count` and
    /// stash the error string. Called by `reminder_scheduler::tick`
    /// when `deliver_reminder` returns Err. Soft-cap behaviour
    /// lives in `get_due_reminders` (rows over the cap drop out
    /// of the due-list and stop being attempted until retry).
    pub fn record_reminder_failure(&self, id: i64, reason: &str) -> Result<()> {
        let conn = self.conn.lock().unwrap();
        conn.execute(
            "UPDATE reminders \
                SET attempt_count = attempt_count + 1, last_error = ?1 \
              WHERE id = ?2 AND sent_at IS NULL",
            params![reason, id],
        )?;
        Ok(())
    }

    /// Clear the failure state on a pending reminder so the
    /// scheduler picks it up again. No-op when the row is already
    /// fresh (`attempt_count == 0`). Returns the number of rows
    /// affected so callers can distinguish "retried" from "no
    /// such pending reminder" (already delivered, or wrong id).
    pub fn reset_reminder_failure(&self, id: i64) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "UPDATE reminders \
                SET attempt_count = 0, last_error = NULL \
              WHERE id = ?1 AND sent_at IS NULL",
            params![id],
        )?;
        Ok(n)
    }

    /// Count this agent's still-pending (un-delivered) reminders.
    /// Used by the per-turn stats sink for a cheap "what was queued
    /// at turn-end" snapshot.
    pub fn count_pending_reminders_for(&self, agent: &str) -> Result<u64> {
        let conn = self.conn.lock().unwrap();
        let n: i64 = conn.query_row(
            "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND sent_at IS NULL",
            params![agent],
            |row| row.get(0),
        )?;
        Ok(u64::try_from(n).unwrap_or(0))
    }

    /// Reminder rollup stats for an agent over a time window. Returns
    /// counts of scheduled, delivered, and pending reminders created
    /// in the last `since_secs` seconds (0 = all reminders).
    pub fn reminder_rollup_for(
        &self,
        agent: &str,
        since_secs: u64,
    ) -> Result<hive_sh4re::ReminderStats> {
        let conn = self.conn.lock().unwrap();
        let cutoff_time = if since_secs > 0 {
            let now = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .ok()
                .and_then(|d| i64::try_from(d.as_secs()).ok())
                .unwrap_or(0);
            now.saturating_sub(i64::try_from(since_secs).unwrap_or(i64::MAX))
        } else {
            i64::MIN
        };
        let scheduled: i64 = conn.query_row(
            "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2",
            params![agent, cutoff_time],
            |row| row.get(0),
        )?;
        let delivered: i64 = conn.query_row(
            "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2 AND sent_at IS NOT NULL",
            params![agent, cutoff_time],
            |row| row.get(0),
        )?;
        let pending: i64 = conn.query_row(
            "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2 AND sent_at IS NULL",
            params![agent, cutoff_time],
            |row| row.get(0),
        )?;
        Ok(hive_sh4re::ReminderStats {
            scheduled: u64::try_from(scheduled).unwrap_or(0),
            delivered: u64::try_from(delivered).unwrap_or(0),
            pending: u64::try_from(pending).unwrap_or(0),
        })
    }

    /// Delete a reminder by id. Returns the number of rows removed (0
    /// when the id never existed or was already delivered). Hard
    /// delete rather than soft so the row doesn't linger and confuse a
    /// re-creation under the same id.
    pub fn cancel_reminder(&self, id: i64) -> Result<usize> {
        let conn = self.conn.lock().unwrap();
        let n = conn.execute(
            "DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
            params![id],
        )?;
        Ok(n)
    }

    /// Cancel a pending reminder on behalf of `canceller`. Returns
    /// the owner agent name on success (handy for logging). Auth
    /// rules mirror `OperatorQuestions::cancel`: the owner, the
    /// operator, or a `privileged` caller (one that arrived on the
    /// manager socket — the trust boundary, not a name match).
    pub fn cancel_reminder_as(&self, id: i64, canceller: &str, privileged: bool) -> Result<String> {
        let conn = self.conn.lock().unwrap();
        let owner: Option<String> = conn
            .query_row(
                "SELECT agent FROM reminders WHERE id = ?1 AND sent_at IS NULL",
                params![id],
                |row| row.get(0),
            )
            .optional()?;
        let Some(owner) = owner else {
            anyhow::bail!("reminder {id} not pending (already delivered or unknown)");
        };
        let authorised =
            privileged || canceller == owner || canceller == hive_sh4re::OPERATOR_RECIPIENT;
        if !authorised {
            anyhow::bail!("reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')");
        }
        let n = conn.execute(
            "DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL",
            params![id],
        )?;
        if n == 0 {
            anyhow::bail!("reminder {id} vanished between auth check and delete");
        }
        Ok(owner)
    }

    /// Get up to `limit` due reminders across all agents in a single query.
    /// Returns `(agent, id, message, file_path)` tuples. Pass a small limit
    /// (e.g. 100) so a burst of overdue reminders doesn't flood the broker
    /// in one cycle — leftovers stay due and get picked up on the next tick.
    pub fn get_due_reminders(&self, limit: u64) -> Result<Vec<DueReminder>> {
        let conn = self.conn.lock().unwrap();
        let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
        let max_attempts = i64::from(MAX_REMINDER_ATTEMPTS);
        // attempt_count >= cap = give up; row stays pending so the
        // operator sees + can retry/cancel via the dashboard.
        let mut stmt = conn.prepare(
            "SELECT agent, id, message, file_path FROM reminders \
             WHERE due_at <= ?1 AND sent_at IS NULL AND attempt_count < ?3 \
             ORDER BY agent, due_at ASC \
             LIMIT ?2",
        )?;
        let rows = stmt.query_map(params![now_unix(), limit_i, max_attempts], |row| {
            Ok((
                row.get::<_, String>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, String>(2)?,
                row.get::<_, Option<String>>(3)?,
            ))
        })?;
        rows.collect::<rusqlite::Result<Vec<_>>>()
            .context("query due reminders")
    }

    /// Atomic reminder delivery: insert the inbox message AND mark the
    /// reminder as sent in a single sqlite transaction. Prevents the
    /// orphan-reminder duplicate-delivery class of bugs that two separate
    /// calls (send + `mark_reminder_sent`) could produce if the second one
    /// failed transiently — the next scheduler tick would see the reminder
    /// still due and redeliver. Either both writes commit or neither does;
    /// re-running on failure is safe.
    ///
    /// Emits a `Sent` event on the broadcast channel after the transaction
    /// commits (so subscribers see the inbox message but never see a
    /// "phantom" send for a transaction that rolled back).
    /// Deliver a batch of reminders in a single transaction, reducing
    /// lock contention on the shared sqlite connection under high
    /// reminder volume. Returns per-item results so the scheduler can
    /// record individual failures without aborting successful ones.
    ///
    /// Items where the INSERT+UPDATE succeeds get a `MessageEvent::Sent`
    /// emitted after the transaction commits. Items that fail are
    /// returned as `Err` in the output vec (index-aligned with input).
    pub fn deliver_reminders_batch(
        &self,
        items: &[(i64, String, String)], // (reminder_id, agent, body)
    ) -> Vec<Result<()>> {
        if items.is_empty() {
            return Vec::new();
        }
        let now = now_unix();
        let mut conn = self.conn.lock().unwrap();
        // Build one transaction for all deliveries so we hold the lock
        // once rather than N times. On a batch-level error (e.g. DB
        // corruption), fall back to returning per-item errors so the
        // scheduler records the failure cleanly.
        let tx = match conn.transaction() {
            Ok(t) => t,
            Err(e) => {
                let err_str = format!("{e:#}");
                return items
                    .iter()
                    .map(|_| Err(anyhow::anyhow!("{}", err_str.clone())))
                    .collect();
            }
        };
        let mut results: Vec<Result<()>> = Vec::with_capacity(items.len());
        // Per-item broker row ids — collected inside the transaction so
        // we can emit Sent events with the correct id after commit.
        let mut msg_ids: Vec<i64> = Vec::with_capacity(items.len());
        for (id, agent, body) in items {
            let r = (|| -> Result<i64> {
                tx.execute(
                    "INSERT INTO messages (sender, recipient, body, sent_at) \
                     VALUES (?1, ?2, ?3, ?4)",
                    params!["reminder", agent, body, now],
                )?;
                let msg_id = tx.last_insert_rowid();
                tx.execute(
                    "UPDATE reminders SET sent_at = ?1 WHERE id = ?2",
                    params![now, id],
                )?;
                Ok(msg_id)
            })();
            match r {
                Ok(msg_id) => {
                    msg_ids.push(msg_id);
                    results.push(Ok(()));
                }
                Err(e) => {
                    msg_ids.push(-1);
                    results.push(Err(e));
                }
            }
        }
        if let Err(e) = tx.commit() {
            let err_str = format!("{e:#}");
            return items
                .iter()
                .map(|_| Err(anyhow::anyhow!("{}", err_str.clone())))
                .collect();
        }
        drop(conn);
        // Emit per-row Sent events (only for rows that succeeded).
        for (((id, agent, body), result), msg_id) in
            items.iter().zip(results.iter()).zip(msg_ids.iter())
        {
            if result.is_ok() {
                let _ = self.events.send(MessageEvent::Sent {
                    id: *msg_id,
                    from: "reminder".to_owned(),
                    to: agent.clone(),
                    body: body.clone(),
                    at: now,
                    in_reply_to: None,
                });
                tracing::debug!(reminder_id = id, %agent, "reminder delivered");
            }
        }
        results
    }
}

/// Idempotent messages-table migrations. Adds `acked_at` and
/// back-fills it for every already-delivered row, so the
/// pre-migration sessions count as "fully handled" and won't be
/// resurfaced by the first `requeue_inflight` after upgrade.
fn ensure_message_columns(conn: &Connection) -> Result<()> {
    let has_acked: bool = conn
        .prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = 'acked_at'")?
        .exists([])?;
    if !has_acked {
        conn.execute_batch("ALTER TABLE messages ADD COLUMN acked_at INTEGER;")
            .context("add messages.acked_at column")?;
        // Backfill: treat every existing delivered row as acked. The
        // session it was delivered to is gone, so requeue would just
        // surface phantom traffic to whatever harness reads next.
        conn.execute(
            "UPDATE messages SET acked_at = delivered_at \
              WHERE delivered_at IS NOT NULL AND acked_at IS NULL",
            [],
        )
        .context("backfill messages.acked_at from delivered_at")?;
    }
    let has_reply: bool = conn
        .prepare("SELECT 1 FROM pragma_table_info('messages') WHERE name = 'in_reply_to'")?
        .exists([])?;
    if !has_reply {
        conn.execute_batch("ALTER TABLE messages ADD COLUMN in_reply_to INTEGER;")
            .context("add messages.in_reply_to column")?;
        // No backfill needed — existing messages simply have NULL here,
        // meaning "root of a new thread", which is correct.
    }
    Ok(())
}

/// Idempotent reminder-table migrations. `ALTER TABLE ADD COLUMN`
/// has no `IF NOT EXISTS` form in sqlite, so we probe
/// `pragma_table_info` per column. New deploys (table created by
/// SCHEMA in this commit cycle) skip the ALTER; pre-existing
/// broker.sqlite files get the columns added on next boot.
fn ensure_reminder_columns(conn: &Connection) -> Result<()> {
    for (name, sql) in [
        (
            "attempt_count",
            "ALTER TABLE reminders ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0;",
        ),
        (
            "last_error",
            "ALTER TABLE reminders ADD COLUMN last_error TEXT;",
        ),
    ] {
        let has: bool = conn
            .prepare(&format!(
                "SELECT 1 FROM pragma_table_info('reminders') WHERE name = '{name}'"
            ))?
            .exists([])?;
        if !has {
            conn.execute_batch(sql)
                .with_context(|| format!("add reminders.{name} column"))?;
        }
    }
    Ok(())
}

fn now_unix() -> i64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .ok()
        .and_then(|d| i64::try_from(d.as_secs()).ok())
        .unwrap_or(0)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicU64, Ordering};

    /// Per-process counter so each test gets a unique sqlite path even
    /// when threads run concurrently. Avoids pulling in a `tempfile`
    /// dep just for this one module.
    static TEST_COUNTER: AtomicU64 = AtomicU64::new(0);

    struct TmpBroker {
        path: std::path::PathBuf,
        pub broker: Broker,
    }

    impl Drop for TmpBroker {
        fn drop(&mut self) {
            let _ = std::fs::remove_file(&self.path);
        }
    }

    fn open_broker() -> TmpBroker {
        let n = TEST_COUNTER.fetch_add(1, Ordering::Relaxed);
        let pid = std::process::id();
        let path = std::env::temp_dir().join(format!("hive-broker-test-{pid}-{n}.sqlite"));
        let _ = std::fs::remove_file(&path);
        let broker = Broker::open(&path).expect("open broker");
        TmpBroker { path, broker }
    }

    fn msg(from: &str, to: &str, body: &str) -> Message {
        Message {
            from: from.to_owned(),
            to: to.to_owned(),
            body: body.to_owned(),
            in_reply_to: None,
        }
    }

    /// Convenience wrapper for tests that want single-pop semantics
    /// — the broker only exposes `recv_batch` publicly, so tests
    /// funnel single pops through here.
    fn pop_one(broker: &Broker, recipient: &str) -> Option<Delivery> {
        let mut batch = broker.recv_batch(recipient, 1).unwrap();
        batch.pop()
    }

    #[test]
    fn kv_set_get_delete_roundtrip() {
        let tb = open_broker();
        let b = &tb.broker;
        // Absent key -> None.
        assert_eq!(b.kv_get("k").unwrap(), None);
        // Set then get.
        b.kv_set("k", "v1").unwrap();
        assert_eq!(b.kv_get("k").unwrap(), Some("v1".to_owned()));
        // Upsert overwrites.
        b.kv_set("k", "v2").unwrap();
        assert_eq!(b.kv_get("k").unwrap(), Some("v2".to_owned()));
        // Delete clears; deleting again is a no-op.
        b.kv_delete("k").unwrap();
        assert_eq!(b.kv_get("k").unwrap(), None);
        b.kv_delete("k").unwrap();
    }

    /// Happy path: send → recv → `ack_turn` drains the in-memory list
    /// and marks the row `acked_at IS NOT NULL`. A second recv finds
    /// nothing pending (the row stays in the table for vacuum).
    #[test]
    fn ack_turn_marks_delivered_rows_acked() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "hi")).unwrap();
        let d = pop_one(broker, "b").expect("popped");
        assert_eq!(d.message.body, "hi");
        assert!(!d.redelivered);
        assert_eq!(broker.ack_turn("b").unwrap(), 1);
        // ack_turn drained the unacked list; calling again is a no-op.
        assert_eq!(broker.ack_turn("b").unwrap(), 0);
        // Recv finds nothing — the row is now delivered + acked.
        assert!(pop_one(broker, "b").is_none());
    }

    /// Bulk triage: three queued messages, agent acks up to the second
    /// id — the first two never deliver again, the third still pops.
    #[test]
    fn ack_until_acks_only_rows_at_or_below_cutoff() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "one")).unwrap();
        broker.send(&msg("a", "b", "two")).unwrap();
        broker.send(&msg("a", "b", "three")).unwrap();
        // Pop the first two so we know their ids (FIFO).
        let d1 = pop_one(broker, "b").expect("popped one");
        let d2 = pop_one(broker, "b").expect("popped two");
        assert_eq!(broker.ack_until("b", d2.id).unwrap(), 2);
        // The cutoff also drained the in-memory unacked list, so a
        // turn-level ack right after finds nothing left to do.
        assert_eq!(broker.ack_turn("b").unwrap(), 0);
        // A restart-style requeue finds nothing below the cutoff …
        assert_eq!(broker.requeue_inflight("b").unwrap(), 0);
        // … and the third message (id above the cutoff) still pops.
        let d3 = pop_one(broker, "b").expect("third still pending");
        assert_eq!(d3.message.body, "three");
        assert!(d1.id < d2.id && d2.id < d3.id);
    }

    /// Pending (never-delivered) rows below the cutoff are acked too —
    /// that's the whole point for a stale backlog the agent never
    /// popped individually.
    #[test]
    fn ack_until_covers_pending_rows() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "stale-1")).unwrap();
        broker.send(&msg("a", "b", "stale-2")).unwrap();
        // Learn the highest id by peeking via recent_for (non-mutating).
        let rows = broker.recent_for("b", 10).unwrap();
        let max_id = rows.iter().map(|r| r.id).max().expect("rows");
        assert_eq!(broker.ack_until("b", max_id).unwrap(), 2);
        assert!(pop_one(broker, "b").is_none(), "backlog cleared");
    }

    /// Recipient scoping: acking b's inbox never touches c's rows,
    /// even when c's ids fall below the cutoff.
    #[test]
    fn ack_until_is_recipient_scoped() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "c", "for-c")).unwrap();
        broker.send(&msg("a", "b", "for-b")).unwrap();
        let rows = broker.recent_for("b", 10).unwrap();
        let max_id = rows.iter().map(|r| r.id).max().expect("rows");
        assert_eq!(broker.ack_until("b", max_id).unwrap(), 1);
        let d = pop_one(broker, "c").expect("c's message untouched");
        assert_eq!(d.message.body, "for-c");
    }

    /// Crash-recovery: send → recv → (no ack) → `requeue_inflight`
    /// resets `delivered_at` + tags the next pop as redelivered. After
    /// that `ack_turn` closes it out cleanly.
    #[test]
    fn requeue_inflight_resurfaces_unacked_with_redelivered_flag() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "hi")).unwrap();
        let d1 = pop_one(broker, "b").expect("popped");
        assert!(!d1.redelivered);
        // Simulate harness crash: never call ack_turn. Now boot the
        // new harness — requeue_inflight resurfaces the row.
        assert_eq!(broker.requeue_inflight("b").unwrap(), 1);
        let d2 = pop_one(broker, "b").expect("popped again");
        assert_eq!(d2.message.body, "hi");
        assert!(d2.redelivered, "second pop should be tagged redelivered");
        assert_eq!(broker.ack_turn("b").unwrap(), 1);
    }

    /// Idempotency: a second `requeue_inflight` on the same recipient
    /// finds nothing because the prior call already reset
    /// `delivered_at` (the row is back in the pending state, not
    /// inflight).
    #[test]
    fn requeue_inflight_is_idempotent() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "hi")).unwrap();
        pop_one(broker, "b").expect("popped");
        assert_eq!(broker.requeue_inflight("b").unwrap(), 1);
        // Second call: the row is pending (delivered_at IS NULL) so
        // nothing matches the inflight filter.
        assert_eq!(broker.requeue_inflight("b").unwrap(), 0);
    }

    /// Multiple messages, partial drain: pop two, `ack_turn` covers
    /// both even though one was popped before the other.
    #[test]
    fn ack_turn_handles_batch() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "one")).unwrap();
        broker.send(&msg("a", "b", "two")).unwrap();
        broker.send(&msg("a", "b", "three")).unwrap();
        pop_one(broker, "b").expect("popped 1");
        pop_one(broker, "b").expect("popped 2");
        pop_one(broker, "b").expect("popped 3");
        assert_eq!(broker.ack_turn("b").unwrap(), 3);
        assert!(pop_one(broker, "b").is_none());
    }

    /// Vacuum filter respects the new `acked_at` semantics — a
    /// delivered-but-not-acked row is NOT vacuumed regardless of
    /// age (the requeue path needs it).
    #[test]
    fn vacuum_preserves_unacked_inflight_rows() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "stuck")).unwrap();
        pop_one(broker, "b").expect("popped");
        // Wide window — should still skip unacked rows.
        let removed = broker.vacuum_delivered(-i64::from(u8::MAX)).unwrap();
        assert_eq!(removed, 0, "unacked inflight row must survive vacuum");
        // After ack_turn the row is fair game.
        broker.ack_turn("b").unwrap();
        let removed = broker.vacuum_delivered(-i64::from(u8::MAX)).unwrap();
        assert_eq!(removed, 1, "acked row is now vacuumable");
    }

    /// Recv ordering: requeued rows go back into FIFO position
    /// (they keep their original id). New sends added after the
    /// requeue arrive after them.
    #[test]
    fn requeued_rows_come_back_in_original_order() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "first")).unwrap();
        broker.send(&msg("a", "b", "second")).unwrap();
        // Pop both, ack neither.
        pop_one(broker, "b").expect("popped 1");
        pop_one(broker, "b").expect("popped 2");
        broker.requeue_inflight("b").unwrap();
        // Now add a brand new message AFTER the requeue.
        broker.send(&msg("a", "b", "third")).unwrap();
        let d1 = pop_one(broker, "b").expect("re-pop 1");
        assert_eq!(d1.message.body, "first");
        assert!(d1.redelivered);
        let d2 = pop_one(broker, "b").expect("re-pop 2");
        assert_eq!(d2.message.body, "second");
        assert!(d2.redelivered);
        let d3 = pop_one(broker, "b").expect("re-pop 3");
        assert_eq!(d3.message.body, "third");
        assert!(
            !d3.redelivered,
            "fresh-send-after-requeue must NOT be tagged redelivered"
        );
    }

    /// Happy path for `recv_batch`: pops in FIFO order, respects
    /// `max`, leaves the rest pending for the next call.
    #[test]
    fn recv_batch_pops_fifo_capped_at_max() {
        let h = open_broker();
        let broker = &h.broker;
        for i in 0..5 {
            broker.send(&msg("a", "b", &format!("m{i}"))).unwrap();
        }
        let batch = broker.recv_batch("b", 3).unwrap();
        let bodies: Vec<_> = batch.iter().map(|d| d.message.body.as_str()).collect();
        assert_eq!(bodies, vec!["m0", "m1", "m2"]);
        // Remaining two stay pending; a second batch drains them.
        let next = broker.recv_batch("b", 10).unwrap();
        let bodies: Vec<_> = next.iter().map(|d| d.message.body.as_str()).collect();
        assert_eq!(bodies, vec!["m3", "m4"]);
        // ack_turn closes out all five popped rows in one go.
        assert_eq!(broker.ack_turn("b").unwrap(), 5);
    }

    /// Transient-wake regression guard: a `ping` fired while no `recv` is parked
    /// must NOT be lost — it's buffered and drained by the next collect.
    #[test]
    fn transient_ping_buffered_when_no_receiver_parked() {
        let h = open_broker();
        let broker = &h.broker;
        // Nobody is parked on recv when these fire (the old broadcast-only
        // path dropped them here).
        broker.ping("b", "matrix", "wake1");
        broker.ping("b", "bash-task", "wake2");
        let batch = broker.collect_batch("b", 10).unwrap();
        let got: Vec<_> = batch
            .iter()
            .map(|d| (d.id, d.message.from.as_str(), d.message.body.as_str()))
            .collect();
        assert_eq!(
            got,
            vec![(0, "matrix", "wake1"), (0, "bash-task", "wake2")],
            "buffered pings drain FIFO as id=0 deliveries"
        );
        // Drained — the buffer is now empty.
        assert!(broker.collect_batch("b", 10).unwrap().is_empty());
        // Pings are sentinel id=0, so ack_turn has nothing to close out.
        assert_eq!(broker.ack_turn("b").unwrap(), 0);
    }

    /// Buffered pings surface ahead of queued sqlite mail in one batch.
    #[test]
    fn pending_pings_surface_before_queued_mail() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "mail")).unwrap();
        broker.ping("b", "matrix", "wake");
        let batch = broker.collect_batch("b", 10).unwrap();
        assert_eq!(batch.len(), 2);
        assert_eq!(batch[0].id, 0, "ping first");
        assert_eq!(batch[0].message.body, "wake");
        assert!(batch[1].id > 0, "sqlite mail after");
        assert_eq!(batch[1].message.body, "mail");
    }

    /// `recv_batch` with no pending traffic returns an empty vec
    /// (the "(empty)" path), not an error.
    #[test]
    fn recv_batch_returns_empty_when_idle() {
        let h = open_broker();
        let batch = h.broker.recv_batch("ghost", 5).unwrap();
        assert!(batch.is_empty());
    }

    /// `max = 0` short-circuits without touching the DB (covered by
    /// asserting we don't accidentally pop a pending row).
    #[test]
    fn recv_batch_zero_max_pops_nothing() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "stay")).unwrap();
        assert!(broker.recv_batch("b", 0).unwrap().is_empty());
        // The pending row is still in flight for the next real recv.
        let d = pop_one(broker, "b").expect("still pending");
        assert_eq!(d.message.body, "stay");
    }

    /// `recv_batch` tags requeued rows with `redelivered: true` and
    /// drains them from the per-recipient `requeued_ids` set so a
    /// fresh follow-up recv after the batch doesn't double-tag.
    #[test]
    fn recv_batch_propagates_redelivered_flag() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "one")).unwrap();
        broker.send(&msg("a", "b", "two")).unwrap();
        pop_one(broker, "b").expect("popped 1");
        pop_one(broker, "b").expect("popped 2");
        broker.requeue_inflight("b").unwrap();
        let batch = broker.recv_batch("b", 5).unwrap();
        assert_eq!(batch.len(), 2);
        assert!(batch.iter().all(|d| d.redelivered));
        // Fresh send after the batch is NOT tagged redelivered.
        broker.send(&msg("a", "b", "three")).unwrap();
        let d = pop_one(broker, "b").expect("re-pop 3");
        assert_eq!(d.message.body, "three");
        assert!(!d.redelivered);
    }

    /// `mark_all_read` covers a mix of pending + delivered + acked rows:
    /// pending rows get both `delivered_at` and `acked_at` backfilled,
    /// delivered-but-unacked rows just get `acked_at` set, already-acked
    /// rows pass through untouched. Returns the count of rows mutated.
    #[test]
    fn mark_all_read_drains_all_states_for_recipient() {
        let h = open_broker();
        let broker = &h.broker;
        // Set up three rows in three different states:
        //   r1 — pending (never popped, both timestamps NULL)
        //   r2 — delivered + unacked (popped, harness didn't ack yet)
        //   r3 — delivered + acked (popped + ack_turn ran)
        broker.send(&msg("a", "b", "pending")).unwrap();
        broker.send(&msg("a", "b", "delivered")).unwrap();
        broker.send(&msg("a", "b", "acked")).unwrap();
        // Pop both deliverable rows, then ack only the last so r2 stays
        // delivered-but-unacked. After pop r3 is still in the unacked
        // list; recv pops in FIFO order so first pop = "pending", but
        // we want THAT row to remain undelivered. Workaround: pop two
        // rows (so "pending" and "delivered" come off the queue) and
        // requeue the first to put "pending" back. Then send a fourth
        // "acked" and pop+ack just that.
        //
        // Simpler approach: bypass the queue helpers and craft the row
        // states directly via send + recv + ack. FIFO order is by
        // insertion; we pop two, ack only the second.
        let _ = pop_one(broker, "b").expect("pop 1: pending → now delivered");
        let _ = pop_one(broker, "b").expect("pop 2: delivered");
        let _ = pop_one(broker, "b").expect("pop 3: acked-soon");
        assert_eq!(broker.ack_turn("b").unwrap(), 3);
        // Now reshape: requeue first two so they're pending again.
        // (Hack — easier: just call mark_all_read on the state we
        // have, which is "three rows already acked". Should return
        // zero because no row has acked_at IS NULL.)
        assert_eq!(broker.mark_all_read("b").unwrap(), 0);
        // Add a fresh pending row + a delivered-but-unacked row.
        broker.send(&msg("a", "b", "new pending")).unwrap();
        broker.send(&msg("a", "b", "new delivered")).unwrap();
        let _ = pop_one(broker, "b").expect("pop new pending → delivered");
        // Don't ack — leaves it delivered+unacked.
        // Now: one row is pending (delivered_at IS NULL), one is
        // delivered+unacked. mark_all_read should hit both.
        assert_eq!(broker.mark_all_read("b").unwrap(), 2);
        // Second call: nothing pending now.
        assert_eq!(broker.mark_all_read("b").unwrap(), 0);
        // Confirm via recv — inbox is empty.
        assert!(pop_one(broker, "b").is_none());
    }

    /// Per-recipient isolation: marking alice doesn't touch bob's inbox.
    #[test]
    fn mark_all_read_is_per_recipient() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("x", "alice", "for alice")).unwrap();
        broker.send(&msg("x", "bob", "for bob")).unwrap();
        assert_eq!(broker.mark_all_read("alice").unwrap(), 1);
        // bob's row still pending — pop succeeds.
        let d = pop_one(broker, "bob").expect("bob pop");
        assert_eq!(d.message.body, "for bob");
    }

    /// After `mark_all_read`, a subsequent `ack_turn` from a stale
    /// in-memory unacked list MUST NOT panic or double-ack. The
    /// inflight bookkeeping is cleared by `mark_all_read`.
    #[test]
    fn mark_all_read_clears_inflight_so_ack_turn_is_noop() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("a", "b", "hi")).unwrap();
        let _ = pop_one(broker, "b").expect("popped");
        // Now there's an id in the unacked_ids list. mark_all_read
        // should clear it so the harness's next ack_turn (which still
        // thinks the id is unacked) returns 0 cleanly.
        assert_eq!(broker.mark_all_read("b").unwrap(), 1);
        assert_eq!(broker.ack_turn("b").unwrap(), 0);
    }

    /// Per-recipient isolation: `requeue_inflight("a")` doesn't touch
    /// b's inflight rows.
    #[test]
    fn requeue_inflight_is_per_recipient() {
        let h = open_broker();
        let broker = &h.broker;
        broker.send(&msg("x", "alice", "for alice")).unwrap();
        broker.send(&msg("x", "bob", "for bob")).unwrap();
        pop_one(broker, "alice").expect("popped alice");
        pop_one(broker, "bob").expect("popped bob");
        // Requeue only alice. Bob's row stays inflight.
        assert_eq!(broker.requeue_inflight("alice").unwrap(), 1);
        let d = pop_one(broker, "alice").expect("re-pop alice");
        assert!(d.redelivered);
        // Bob has nothing pending (his row is still delivered, not requeued).
        assert!(pop_one(broker, "bob").is_none());
    }
}