diff --git a/TODO.md b/TODO.md index 6e3d90b1..f7a3c6c9 100644 --- a/TODO.md +++ b/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//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//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` 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. diff --git a/hive-ag3nt/assets/agent.css b/hive-ag3nt/assets/agent.css index a15548cb..0cf22279 100644 --- a/hive-ag3nt/assets/agent.css +++ b/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); diff --git a/hive-ag3nt/assets/app.js b/hive-ag3nt/assets/app.js index 8f786c19..3c9461bd 100644 --- a/hive-ag3nt/assets/app.js +++ b/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 gets clobbered every cycle. diff --git a/hive-ag3nt/assets/index.html b/hive-ag3nt/assets/index.html index c1e7ae4a..5ebae41f 100644 --- a/hive-ag3nt/assets/index.html +++ b/hive-ag3nt/assets/index.html @@ -17,6 +17,7 @@ … booting + diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 979b5f4e..0588c820 100644 --- a/hive-ag3nt/src/events.rs +++ b/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 { + 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>, + /// 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>>, /// 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 { + *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) { diff --git a/hive-ag3nt/src/plugins.rs b/hive-ag3nt/src/plugins.rs index 7530656c..baea3630 100644 --- a/hive-ag3nt/src/plugins.rs +++ b/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 `. 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::(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]) diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 1ff442c5..19923179 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -276,7 +276,12 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result flag_out.store(true, Ordering::Relaxed); } match serde_json::from_str::(&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}"))), } } diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs index 02a9df2e..80a142fb 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/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, } #[derive(Serialize)] @@ -232,6 +235,7 @@ async fn api_state(State(state): State) -> axum::Json { 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) -> axum::Json { turn_state, turn_state_since, model, + token_usage, }) } diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 8ba71fc0..e90da403 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -172,89 +172,112 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> 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 = 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, + agent: &str, + question: &str, + options: &[String], + multi: bool, + ttl_seconds: Option, +) -> 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 { + 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), } } diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index c1bae452..b9d6a20b 100644 --- a/hive-c0re/src/broker.rs +++ b/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); + #[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> { + 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)>> { + /// 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> { 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::>>() - .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(()) } } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 3ebbabee..ea84f8c2 100644 --- a/hive-c0re/src/main.rs +++ b/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" - ); } } } diff --git a/nix/templates/harness-base.nix b/nix/templates/harness-base.nix index 8b43e7af..5dea0e2b 100644 --- a/nix/templates/harness-base.nix +++ b/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