diff --git a/TODO.md b/TODO.md index f7a3c6c9..6e3d90b1 100644 --- a/TODO.md +++ b/TODO.md @@ -11,14 +11,13 @@ ## Reminder Tool -- ~~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. +- Handle text overflow → suggest file_path option for long messages - Per-agent reminder limits (burst capacity, rate limiting) -- **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. +- **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 - **Scheduler shutdown**: add graceful shutdown signal when coordinator is destroyed (currently runs forever) -- **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. +- **DB lock contention**: under high reminder volume, many concurrent mark_reminder_sent calls may serialize behind the Mutex lock — consider batch updates ## Dashboard @@ -29,4 +28,4 @@ ## Bugs -- ~~**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. +- **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) diff --git a/hive-ag3nt/assets/agent.css b/hive-ag3nt/assets/agent.css index 0cf22279..a15548cb 100644 --- a/hive-ag3nt/assets/agent.css +++ b/hive-ag3nt/assets/agent.css @@ -180,12 +180,6 @@ 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 3c9461bd..8f786c19 100644 --- a/hive-ag3nt/assets/app.js +++ b/hive-ag3nt/assets/app.js @@ -412,21 +412,6 @@ 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; @@ -500,7 +485,6 @@ 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 5ebae41f..c1e7ae4a 100644 --- a/hive-ag3nt/assets/index.html +++ b/hive-ag3nt/assets/index.html @@ -17,7 +17,6 @@ … booting - diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index 0588c820..979b5f4e 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -156,41 +156,6 @@ 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 @@ -226,12 +191,6 @@ 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 @@ -261,7 +220,6 @@ 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)), } } @@ -300,17 +258,6 @@ 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 baea3630..7530656c 100644 --- a/hive-ag3nt/src/plugins.rs +++ b/hive-ag3nt/src/plugins.rs @@ -19,7 +19,6 @@ 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 @@ -67,15 +66,6 @@ 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() { @@ -122,11 +112,7 @@ pub async fn install_configured(socket: &Path, notify_recipient: Option<&str>) { return; } add_marketplaces().await; - if auto_update_enabled().await { - update_marketplaces().await; - } else { - tracing::debug!("claudePluginsAutoUpdate=false, skipping marketplace update"); - } + update_marketplaces().await; 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 19923179..1ff442c5 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -276,12 +276,7 @@ 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) => { - if let Some(usage) = crate::events::TokenUsage::from_stream_event(&v) { - bus_out.record_usage(usage); - } - bus_out.emit(LiveEvent::Stream(v)); - } + Ok(v) => 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 80a142fb..02a9df2e 100644 --- a/hive-ag3nt/src/web_ui.rs +++ b/hive-ag3nt/src/web_ui.rs @@ -196,9 +196,6 @@ 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)] @@ -235,7 +232,6 @@ 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, @@ -245,7 +241,6 @@ 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 e90da403..8ba71fc0 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -172,112 +172,89 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> options, multi, ttl_seconds, - } => handle_ask_operator(coord, agent, question, 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:#}"), + }, + } + } AgentRequest::Remind { message, timing, file_path, - } => handle_remind(broker, agent, message, timing, file_path.as_deref()), - } -} + } => { + use hive_sh4re::ReminderTiming; -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:#}"), + // 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), }; - } - }; - 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:#}"), - }, - } -} -/// 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}")) + 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:#}"), + }, + } } - ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp), } } diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index b9d6a20b..c1bae452 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -40,12 +40,6 @@ 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 { @@ -148,25 +142,15 @@ 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()) @@ -251,20 +235,16 @@ impl Broker { Ok(id) } - /// 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> { + /// 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)>> { 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 \ - LIMIT ?2", + "SELECT agent, id, message, file_path FROM reminders WHERE due_at <= ?1 AND sent_at IS NULL ORDER BY agent, due_at ASC" )?; - let rows = stmt.query_map(params![now_unix(), limit_i], |row| { + let rows = stmt.query_map(params![now_unix()], |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, i64>(1)?, @@ -273,40 +253,16 @@ impl Broker { )) })?; rows.collect::>>() - .context("query due reminders") + .context("query all 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). - 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( + /// 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, id], + params![now_unix(), 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 ea84f8c2..3ebbabee 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -86,12 +86,6 @@ 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() @@ -172,30 +166,36 @@ async fn main() -> Result<()> { // operator-initiated transient state. crash_watch::spawn(coord.clone()); // Reminder scheduler: checks for due reminders every 5 seconds, - // 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. + // delivers them as inbox messages from "reminder". let reminder_coord = coord.clone(); tokio::spawn(async move { + use hive_sh4re::Message; loop { - match reminder_coord - .broker - .get_due_reminders(REMINDER_BATCH_LIMIT) - { + // Query all due reminders in a single DB call + match reminder_coord.broker.get_all_due_reminders() { Ok(reminders) => { for (agent, id, message, _file_path) in reminders { - if let Err(e) = - reminder_coord.broker.deliver_reminder(id, &agent, &message) - { + // 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(), + }) { 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 5dea0e2b..8b43e7af 100644 --- a/nix/templates/harness-base.nix +++ b/nix/templates/harness-base.nix @@ -149,19 +149,6 @@ ''; }; - 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; @@ -174,9 +161,6 @@ 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