agent_server: reminder body size cap + extract Remind/AskOperator handlers

TODO.md

@@ -11,13 +11,14 @@
 
 ## Reminder Tool
 
-- Handle text overflow → suggest file_path option for long messages
+- ~~Handle text overflow → suggest file_path option for long messages~~ ✓ fixed — Remind dispatch rejects `message.len() > 4096` (when no `file_path` was supplied) with an error pointing at the `file_path` escape hatch.
 - Per-agent reminder limits (burst capacity, rate limiting)
-- **File path delivery**: currently unused in scheduler delivery loop — implement file write/delivery to /state/<agent>/reminders/ or similar
-- **Orphan reminders**: handle partial failures (e.g. delivery succeeds but mark_reminder_sent fails) to avoid resending
-- **Unbounded batches**: implement per-cycle delivery limit so burst of 10k reminders doesn't flood the broker in one cycle
+- **Expose `remind` MCP tool**: wire protocol exists (`AgentRequest::Remind`) and the broker handles it, but no `#[tool]` method on `AgentServer` actually surfaces it to claude. Until that lands, the Remind path is unreachable from agent turns.
+- **File path delivery**: currently unused in scheduler delivery loop — implement file write/delivery to /state/<agent>/reminders/ or similar (also needed for the overflow-check escape hatch above to actually do anything useful).
+- ~~**Orphan reminders**~~ ✓ fixed — `Broker::deliver_reminder` wraps the inbox INSERT + reminders UPDATE in one sqlite transaction; partial failure can no longer cause duplicate delivery on the next tick.
+- ~~**Unbounded batches**~~ ✓ fixed — scheduler now calls `get_due_reminders(REMINDER_BATCH_LIMIT)` (cap = 100/tick); overflow stays due and gets picked up next cycle.
 - **Scheduler shutdown**: add graceful shutdown signal when coordinator is destroyed (currently runs forever)
-- **DB lock contention**: under high reminder volume, many concurrent mark_reminder_sent calls may serialize behind the Mutex lock — consider batch updates
+- **DB lock contention**: under high reminder volume, the broker's `Mutex<Connection>` serializes every delivery transaction. Consider batching multiple deliveries into one tx, or moving reminders onto a separate sqlite connection.
 
 ## Dashboard
 
@@ -28,4 +29,4 @@
 
 ## Bugs
 
-- **Pending message wake-up**: when a message is pending and an agent turn ends without recv(), session doesn't immediately wake up again. Requires another message to trigger wake-up. (inbox/recv logic issue)
+- ~~**Pending message wake-up**~~ ✓ fixed (e423d57) — subscribe-before-check race in `broker.recv_blocking` meant a send landing between the initial `recv()` and `subscribe()` was missed; agent then sat on the 180s long-poll until another, unrelated message woke it. Now subscribe first.

hive-ag3nt/assets/agent.css

@@ -180,6 +180,12 @@ pre.diff {
   font-size: 0.78em;
   letter-spacing: 0.04em;
 }
+.token-usage {
+  color: var(--muted);
+  font-size: 0.8em;
+  letter-spacing: 0.04em;
+  cursor: default;
+}
 .btn-dashlink {
   color: var(--cyan);
   border: 1px solid var(--cyan);

hive-ag3nt/assets/app.js

@@ -412,6 +412,21 @@
     el_.hidden = false;
     el_.textContent = 'model · ' + model;
   }
+  function renderTokenUsage(u) {
+    const el_ = $('token-usage');
+    if (!el_) return;
+    if (!u) { el_.hidden = true; return; }
+    const ctx = u.input_tokens + u.cache_read_input_tokens + u.cache_creation_input_tokens;
+    const fmt = (n) => n >= 1000 ? (n / 1000).toFixed(1) + 'k' : String(n);
+    el_.hidden = false;
+    el_.title = [
+      'input: ' + u.input_tokens,
+      'output: ' + u.output_tokens,
+      'cache_read: ' + u.cache_read_input_tokens,
+      'cache_write: ' + u.cache_creation_input_tokens,
+    ].join(' · ');
+    el_.textContent = '· ctx ' + fmt(ctx) + ' in · ' + fmt(u.output_tokens) + ' out';
+  }
   function renderLastTurn(ms) {
     const el_ = $('last-turn');
     if (!el_) return;
@@ -485,6 +500,7 @@
         setStateAbs(s.turn_state, s.turn_state_since);
       }
       renderModelChip(s.model);
+      renderTokenUsage(s.token_usage);
       // Skip the re-render if nothing structurally changed. The most
       // common case is `online` polling itself — without this guard, the
       // operator's <input value> gets clobbered every cycle.

hive-ag3nt/assets/index.html

@@ -17,6 +17,7 @@
     <span id="state-badge" class="state-badge state-loading">… booting</span>
     <span id="model-chip" class="model-chip" hidden></span>
     <span id="last-turn" class="last-turn" hidden></span>
+    <span id="token-usage" class="token-usage" hidden></span>
     <button type="button" id="cancel-btn" class="btn-cancel-turn" hidden>■ cancel turn</button>
     <button type="button" id="new-session-btn" class="btn-new-session"
             title="next turn runs without --continue, starting a fresh claude session">↻ new session</button>

hive-ag3nt/src/events.rs

@@ -156,6 +156,41 @@ impl EventStore {
     }
 }
 
+/// Token usage emitted by claude in the final `result` stream-json event.
+/// All counts are in tokens. `None` fields mean the server didn't report them.
+#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
+pub struct TokenUsage {
+    pub input_tokens: u64,
+    pub output_tokens: u64,
+    pub cache_read_input_tokens: u64,
+    pub cache_creation_input_tokens: u64,
+}
+
+impl TokenUsage {
+    /// Total context consumed this turn (input + cache reads + cache writes).
+    pub fn context_tokens(&self) -> u64 {
+        self.input_tokens + self.cache_read_input_tokens + self.cache_creation_input_tokens
+    }
+
+    /// Parse usage from a stream-json event. Returns `Some` only for the
+    /// terminal `result` event (which is the only one that carries `usage`);
+    /// every other event maps to `None`. Missing numeric fields default to 0
+    /// so partial server payloads don't drop the whole snapshot.
+    pub fn from_stream_event(v: &serde_json::Value) -> Option<Self> {
+        if v.get("type").and_then(|t| t.as_str()) != Some("result") {
+            return None;
+        }
+        let u = v.get("usage")?;
+        let field = |k: &str| u.get(k).and_then(serde_json::Value::as_u64).unwrap_or(0);
+        Some(Self {
+            input_tokens: field("input_tokens"),
+            output_tokens: field("output_tokens"),
+            cache_read_input_tokens: field("cache_read_input_tokens"),
+            cache_creation_input_tokens: field("cache_creation_input_tokens"),
+        })
+    }
+}
+
 /// Authoritative turn-loop state. The harness owns it; the web UI
 /// reads via `/api/state` and renders. Lives alongside the bus
 /// because everyone who has a `Bus` already has the right handle to
@@ -191,6 +226,12 @@ pub struct Bus {
     /// Model name passed to `claude --model`. Default `haiku`; the
     /// operator can override at runtime via `POST /api/model`.
     model: Arc<Mutex<String>>,
+    /// Last token usage reported by claude (from the `result` stream-json
+    /// event). `None` until the first turn with usage data completes.
+    /// Updated on every turn; survives across turns within one harness
+    /// process lifetime (resets on container restart, which is fine —
+    /// it's a live indicator, not a cumulative counter).
+    last_usage: Arc<Mutex<Option<TokenUsage>>>,
     /// One-shot: next `run_claude` call drops `--continue`, starting
     /// a fresh claude session. Set by `POST /api/new-session` from
     /// the per-agent web UI; consumed (cleared back to false) by the
@@ -220,6 +261,7 @@ impl Bus {
             store,
             state: Arc::new(Mutex::new((TurnState::Idle, now_unix()))),
             model: Arc::new(Mutex::new(initial_model)),
+            last_usage: Arc::new(Mutex::new(None)),
             skip_continue_once: Arc::new(AtomicBool::new(false)),
         }
     }
@@ -258,6 +300,17 @@ impl Bus {
         }
     }
 
+    /// Record the latest token usage from a completed turn.
+    pub fn record_usage(&self, usage: TokenUsage) {
+        *self.last_usage.lock().unwrap() = Some(usage);
+    }
+
+    /// Last known token usage, or `None` if no turn has completed yet.
+    #[must_use]
+    pub fn last_usage(&self) -> Option<TokenUsage> {
+        *self.last_usage.lock().unwrap()
+    }
+
     /// Update the harness's authoritative turn-loop state. Records
     /// the transition time so `state_snapshot` can return a since-age.
     pub fn set_state(&self, next: TurnState) {

hive-ag3nt/src/plugins.rs

@@ -19,6 +19,7 @@ use crate::client;
 
 const PLUGINS_PATH: &str = "/etc/hyperhive/claude-plugins.json";
 const MARKETPLACES_PATH: &str = "/etc/hyperhive/claude-marketplaces.json";
+const AUTO_UPDATE_PATH: &str = "/etc/hyperhive/claude-plugins-auto-update.json";
 
 /// Add every marketplace from `/etc/hyperhive/claude-marketplaces.json`
 /// via `claude plugin marketplace add <source>`. Idempotent: re-add of
@@ -66,6 +67,15 @@ async fn add_marketplaces() {
     }
 }
 
+/// Read the `hyperhive.claudePluginsAutoUpdate` flag written by the NixOS
+/// module. Defaults to `false` when the file is absent or unparseable.
+async fn auto_update_enabled() -> bool {
+    match tokio::fs::read_to_string(AUTO_UPDATE_PATH).await {
+        Ok(s) => serde_json::from_str::<bool>(s.trim()).unwrap_or(false),
+        Err(_) => false,
+    }
+}
+
 /// Update all configured plugin marketplaces. Non-fatal — logs a warning
 /// on failure but does not abort the install sequence.
 async fn update_marketplaces() {
@@ -112,7 +122,11 @@ pub async fn install_configured(socket: &Path, notify_recipient: Option<&str>) {
         return;
     }
     add_marketplaces().await;
-    update_marketplaces().await;
+    if auto_update_enabled().await {
+        update_marketplaces().await;
+    } else {
+        tracing::debug!("claudePluginsAutoUpdate=false, skipping marketplace update");
+    }
     for spec in specs {
         match Command::new("claude")
             .args(["plugin", "install", &spec])

hive-ag3nt/src/turn.rs

@@ -276,7 +276,12 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<bool>
                 flag_out.store(true, Ordering::Relaxed);
             }
             match serde_json::from_str::<serde_json::Value>(&line) {
-                Ok(v) => bus_out.emit(LiveEvent::Stream(v)),
+                Ok(v) => {
+                    if let Some(usage) = crate::events::TokenUsage::from_stream_event(&v) {
+                        bus_out.record_usage(usage);
+                    }
+                    bus_out.emit(LiveEvent::Stream(v));
+                }
                 Err(_) => bus_out.emit(LiveEvent::Note(format!("(non-json) {line}"))),
             }
         }

hive-ag3nt/src/web_ui.rs

@@ -196,6 +196,9 @@ struct StateSnapshot {
     /// the operator can see what they just switched to (and what's
     /// in flight). Mutable at runtime via `POST /api/model`.
     model: String,
+    /// Token usage from the last completed turn. `null` until the
+    /// first turn with usage data finishes.
+    token_usage: Option<crate::events::TokenUsage>,
 }
 
 #[derive(Serialize)]
@@ -232,6 +235,7 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
     let inbox = recent_inbox(&state.socket, state.flavor()).await;
     let (turn_state, turn_state_since) = state.bus.state_snapshot();
     let model = state.bus.model();
+    let token_usage = state.bus.last_usage();
     axum::Json(StateSnapshot {
         label: state.label.clone(),
         dashboard_port,
@@ -241,6 +245,7 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
         turn_state,
         turn_state_since,
         model,
+        token_usage,
     })
 }
 

hive-c0re/src/agent_server.rs

@@ -172,89 +172,112 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
             options,
             multi,
             ttl_seconds,
-        } => {
-            let deadline_at = ttl_seconds.and_then(|s| {
-                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);
-                i64::try_from(s).ok().map(|s| now + s)
-            });
-            match coord
-                .questions
-                .submit(agent, question, options, *multi, deadline_at)
-            {
-                Ok(id) => {
-                    tracing::info!(%id, %agent, ?deadline_at, "agent question queued");
-                    if let Some(ttl) = *ttl_seconds {
-                        crate::manager_server::spawn_question_watchdog(coord, id, ttl);
-                    }
-                    AgentResponse::QuestionQueued { id }
-                }
-                Err(e) => AgentResponse::Err {
-                    message: format!("{e:#}"),
-                },
-            }
-        }
+        } => handle_ask_operator(coord, agent, question, options, *multi, *ttl_seconds),
         AgentRequest::Remind {
             message,
             timing,
             file_path,
-        } => {
-            use hive_sh4re::ReminderTiming;
+        } => handle_remind(broker, agent, message, timing, file_path.as_deref()),
+    }
+}
 
-            // Calculate the due_at timestamp, propagating errors instead of silently
-            // defaulting to epoch 1970 on overflow/conversion failure.
-            let due_at_result: Result<i64> = match timing {
-                ReminderTiming::InSeconds { seconds } => {
-                    let now = std::time::SystemTime::now();
-                    let future = match now.checked_add(std::time::Duration::from_secs(*seconds)) {
-                        Some(t) => t,
-                        None => {
-                            return AgentResponse::Err {
-                                message: format!(
-                                    "InSeconds overflow: {seconds}s exceeds system time range"
-                                ),
-                            };
-                        }
-                    };
-                    let duration = match future.duration_since(std::time::UNIX_EPOCH) {
-                        Ok(d) => d,
-                        Err(e) => {
-                            return AgentResponse::Err {
-                                message: format!("system time before UNIX_EPOCH: {e}"),
-                            };
-                        }
-                    };
-                    match i64::try_from(duration.as_secs()) {
-                        Ok(ts) => Ok(ts),
-                        Err(e) => {
-                            return AgentResponse::Err {
-                                message: format!("unix timestamp exceeds i64 range: {e}"),
-                            };
-                        }
-                    }
-                }
-                ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp),
+fn handle_ask_operator(
+    coord: &Arc<Coordinator>,
+    agent: &str,
+    question: &str,
+    options: &[String],
+    multi: bool,
+    ttl_seconds: Option<u64>,
+) -> AgentResponse {
+    let deadline_at = ttl_seconds.and_then(|s| {
+        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);
+        i64::try_from(s).ok().map(|s| now + s)
+    });
+    match coord
+        .questions
+        .submit(agent, question, options, multi, deadline_at)
+    {
+        Ok(id) => {
+            tracing::info!(%id, %agent, ?deadline_at, "agent question queued");
+            if let Some(ttl) = ttl_seconds {
+                crate::manager_server::spawn_question_watchdog(coord, id, ttl);
+            }
+            AgentResponse::QuestionQueued { id }
+        }
+        Err(e) => AgentResponse::Err {
+            message: format!("{e:#}"),
+        },
+    }
+}
+
+/// Cap on the inline `message` byte length the Remind request accepts.
+/// Reminders land in the agent's inbox and feed the next wake prompt — a
+/// multi-kilobyte body bloats every subsequent turn's context. Anything
+/// bigger should be persisted to disk by the caller and pointed at via
+/// `file_path` (which the scheduler will deliver as a path reference rather
+/// than the full body).
+const REMIND_MESSAGE_MAX: usize = 4096;
+
+fn handle_remind(
+    broker: &crate::broker::Broker,
+    agent: &str,
+    message: &str,
+    timing: &hive_sh4re::ReminderTiming,
+    file_path: Option<&str>,
+) -> AgentResponse {
+    if file_path.is_none() && message.len() > REMIND_MESSAGE_MAX {
+        return AgentResponse::Err {
+            message: format!(
+                "reminder body too long ({} bytes, max {REMIND_MESSAGE_MAX}); write the \
+                 payload to a file under your /state/ dir and pass its path as \
+                 `file_path` so the reminder delivers a pointer instead of the full body",
+                message.len()
+            ),
+        };
+    }
+    let due_at = match resolve_due_at(timing) {
+        Ok(t) => t,
+        Err(e) => {
+            return AgentResponse::Err {
+                message: format!("invalid reminder timing: {e:#}"),
             };
+        }
+    };
+    match broker.store_reminder(agent, message, file_path, 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:#}"),
+        },
+    }
+}
 
-            match due_at_result {
-                Ok(due_at) => {
-                    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:#}"),
-                        },
-                    }
-                }
-                Err(e) => AgentResponse::Err {
-                    message: format!("invalid reminder timing: {e:#}"),
-                },
-            }
+/// Resolve the `due_at` unix timestamp for a Remind request. Returns
+/// distinct error messages for each failure mode (overflow on
+/// `InSeconds`, pre-epoch clock, `i64` cast wrap) so the caller can tell
+/// what went wrong without inspecting the chain.
+fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result<i64> {
+    use hive_sh4re::ReminderTiming;
+    match timing {
+        ReminderTiming::InSeconds { seconds } => {
+            let now = std::time::SystemTime::now();
+            let future = now
+                .checked_add(std::time::Duration::from_secs(*seconds))
+                .ok_or_else(|| {
+                    anyhow::anyhow!("InSeconds overflow: {seconds}s exceeds system time range")
+                })?;
+            let duration = future
+                .duration_since(std::time::UNIX_EPOCH)
+                .map_err(|e| anyhow::anyhow!("system time before UNIX_EPOCH: {e}"))?;
+            i64::try_from(duration.as_secs())
+                .map_err(|e| anyhow::anyhow!("unix timestamp exceeds i64 range: {e}"))
         }
+        ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp),
     }
 }

hive-c0re/src/broker.rs

@@ -40,6 +40,12 @@ CREATE INDEX IF NOT EXISTS idx_reminders_due
 /// 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>);
+
 #[derive(Debug, Clone, Serialize)]
 #[serde(rename_all = "snake_case", tag = "kind")]
 pub enum MessageEvent {
@@ -142,15 +148,25 @@ impl Broker {
     /// emit a `Sent { to: recipient }` event, then retries the pop. Lets
     /// agents react to new mail without polling their socket on a fixed
     /// interval.
+    ///
+    /// **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()`
+    /// catches the message either way.
     pub async fn recv_blocking(
         &self,
         recipient: &str,
         timeout: std::time::Duration,
     ) -> Result<Option<Message>> {
+        let mut rx = self.subscribe();
         if let Some(m) = self.recv(recipient)? {
             return Ok(Some(m));
         }
-        let mut rx = self.subscribe();
         let deadline = tokio::time::Instant::now() + timeout;
         loop {
             let Some(remaining) = deadline.checked_duration_since(tokio::time::Instant::now())
@@ -235,16 +251,20 @@ impl Broker {
         Ok(id)
     }
 
-    /// Get all reminders for an agent that are due now or in the past.
-    /// Returns (id, message, file_path) tuples.
-    /// Get all due reminders across all agents in a single query.
-    /// Returns a vec of (agent, id, message, file_path) tuples.
-    pub fn get_all_due_reminders(&self) -> Result<Vec<(String, i64, String, Option<String>)>> {
+    /// 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 mut stmt = conn.prepare(
-            "SELECT agent, id, message, file_path FROM reminders WHERE due_at <= ?1 AND sent_at IS NULL ORDER BY agent, due_at ASC"
+            "SELECT agent, id, message, file_path FROM reminders \
+             WHERE due_at <= ?1 AND sent_at IS NULL \
+             ORDER BY agent, due_at ASC \
+             LIMIT ?2",
         )?;
-        let rows = stmt.query_map(params![now_unix()], |row| {
+        let rows = stmt.query_map(params![now_unix(), limit_i], |row| {
             Ok((
                 row.get::<_, String>(0)?,
                 row.get::<_, i64>(1)?,
@@ -253,16 +273,40 @@ impl Broker {
             ))
         })?;
         rows.collect::<rusqlite::Result<Vec<_>>>()
-            .context("query all due reminders")
+            .context("query due 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],
+    /// 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).
+    pub fn deliver_reminder(&self, id: i64, agent: &str, message: &str) -> Result<()> {
+        let now = now_unix();
+        let mut conn = self.conn.lock().unwrap();
+        let tx = conn.transaction()?;
+        tx.execute(
+            "INSERT INTO messages (sender, recipient, body, sent_at) VALUES (?1, ?2, ?3, ?4)",
+            params!["reminder", agent, message, now],
         )?;
+        tx.execute(
+            "UPDATE reminders SET sent_at = ?1 WHERE id = ?2",
+            params![now, id],
+        )?;
+        tx.commit()?;
+        drop(conn);
+        let _ = self.events.send(MessageEvent::Sent {
+            from: "reminder".to_owned(),
+            to: agent.to_owned(),
+            body: message.to_owned(),
+            at: now,
+        });
         Ok(())
     }
 }

hive-c0re/src/main.rs

@@ -86,6 +86,12 @@ enum Cmd {
     Deny { id: i64 },
 }
 
+/// Per-tick cap on reminders the scheduler delivers. Anything over this
+/// stays due in the table and gets picked up on the next 5s tick — keeps
+/// a 10k-deep backlog from flooding the broker (or hogging its mutex) in
+/// one shot.
+const REMINDER_BATCH_LIMIT: u64 = 100;
+
 #[tokio::main]
 async fn main() -> Result<()> {
     tracing_subscriber::fmt()
@@ -166,36 +172,30 @@ async fn main() -> Result<()> {
             // 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".
+            // delivers them atomically (insert inbox + mark sent in one
+            // sqlite transaction so a transient failure on the second step
+            // can never produce a duplicate next tick). Per-cycle batch
+            // limit caps the burst — leftover reminders stay due and get
+            // picked up on the next tick instead of monopolising the broker
+            // mutex.
             let reminder_coord = coord.clone();
             tokio::spawn(async move {
-                use hive_sh4re::Message;
                 loop {
-                    // Query all due reminders in a single DB call
-                    match reminder_coord.broker.get_all_due_reminders() {
+                    match reminder_coord
+                        .broker
+                        .get_due_reminders(REMINDER_BATCH_LIMIT)
+                    {
                         Ok(reminders) => {
                             for (agent, 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(),
-                                }) {
+                                if let Err(e) =
+                                    reminder_coord.broker.deliver_reminder(id, &agent, &message)
+                                {
                                     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"
-                                    );
                                 }
                             }
                         }

nix/templates/harness-base.nix

@@ -149,6 +149,19 @@
     '';
   };
 
+  options.hyperhive.claudePluginsAutoUpdate = lib.mkOption {
+    type = lib.types.bool;
+    default = false;
+    description = ''
+      When true, the harness runs `claude plugin marketplace update`
+      before installing plugins at boot, pulling the latest index from
+      all configured marketplaces. Disabled by default — most agents
+      want pinned plugin versions and the network round-trip adds to
+      boot time. Enable for agents that should always install the latest
+      available version of their plugins.
+    '';
+  };
+
   config = {
     environment.etc."hyperhive/extra-mcp.json".text = builtins.toJSON config.hyperhive.extraMcpServers;
 
@@ -161,6 +174,9 @@
     environment.etc."hyperhive/claude-marketplaces.json".text =
       builtins.toJSON config.hyperhive.claudeMarketplaces;
 
+    environment.etc."hyperhive/claude-plugins-auto-update.json".text =
+      builtins.toJSON config.hyperhive.claudePluginsAutoUpdate;
+
     boot.isNspawnContainer = true;
 
     # `claude-code` is unfree. Each per-agent container's nixosConfiguration