From a80d0b0fed7c89e8667ea7bb759529fcd0851f87 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 23:01:16 +0200 Subject: [PATCH] delete c0re-side reminder plumbing (#2635 inc 1 commit 6) --- hive-c0re/src/container_view.rs | 14 +- hive-c0re/src/coordinator.rs | 20 +- hive-c0re/src/dashboard/mod.rs | 10 - hive-c0re/src/dashboard/reminders.rs | 61 ---- hive-c0re/src/dashboard_events.rs | 13 - hive-c0re/src/loose_ends.rs | 43 +-- hive-c0re/src/main.rs | 8 +- hive-c0re/src/questions.rs | 17 +- hive-c0re/src/server.rs | 14 +- hive-c0re/src/socket_server/mod.rs | 85 ++--- hive-c0re/src/socket_server/reminders.rs | 229 ------------ hive-c0re/src/stats/host_stats.rs | 1 - hive-c0re/src/stores/broker.rs | 365 -------------------- hive-c0re/src/workers/mod.rs | 3 +- hive-c0re/src/workers/reminder_scheduler.rs | 278 --------------- hive-core-agent-sock/src/lib.rs | 45 +-- 16 files changed, 70 insertions(+), 1136 deletions(-) delete mode 100644 hive-c0re/src/dashboard/reminders.rs delete mode 100644 hive-c0re/src/socket_server/reminders.rs delete mode 100644 hive-c0re/src/workers/reminder_scheduler.rs diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs index abefef7b..b76d30de 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -29,13 +29,6 @@ pub struct ContainerView { /// for this agent's input. #[serde(skip_serializing_if = "Option::is_none")] pub deployed_sha: Option, - /// Count of this agent's pending reminders. Computed during - /// `build_all` via `Broker::count_pending_reminders_for`; the - /// dashboard renders a small chip when > 0. Updates with the - /// 10s `crash_watch` rescan + every container mutation site; - /// not real-time on remind/cancel-reminder but close enough. - #[serde(default)] - pub pending_reminders: u64, /// Name of this agent's parent in the agent hierarchy. `None` /// marks the agent as root-level; the dashboard renders it without /// indentation. Sourced from `meta/topology.json` (single source of @@ -55,7 +48,7 @@ pub struct ContainerView { /// Build the full container list. Wraps `lifecycle::list()` and /// resolves every per-agent attribute the dashboard surfaces. -pub async fn build_all(coord: &Coordinator) -> Vec { +pub async fn build_all() -> Vec { let raw = lifecycle::list().await.unwrap_or_default(); let locked = read_meta_locked_revs(); // Pull the topology map once and look up each agent's parent below. @@ -79,10 +72,6 @@ pub async fn build_all(coord: &Coordinator) -> Vec { let needs_update = crate::auto_update::agent_config_pending(logical.as_str(), deployed_full).await; let deployed_sha = deployed_full.map(|s| s[..s.len().min(12)].to_owned()); - let pending_reminders = coord - .broker - .count_pending_reminders_for(logical.as_str()) - .unwrap_or(0); let parent = topology.get(logical.as_str()).cloned().flatten(); let running = lifecycle::is_running(logical.as_str()).await; // needs_login fires when EITHER the claude session dir is missing @@ -108,7 +97,6 @@ pub async fn build_all(coord: &Coordinator) -> Vec { needs_update, needs_login, deployed_sha, - pending_reminders, parent, active_model, }); diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 6c07db17..c24247f9 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -612,24 +612,6 @@ impl Coordinator { .map(|m| m.keys().cloned().collect()) } - /// Emit a `RemindersChanged` snapshot event. Called from every - /// reminder mutation site (agent `remind` calls, operator cancel / - /// retry, and the scheduler after each delivery batch) so the - /// dashboard's pending-reminders list stays live without polling. - pub fn emit_reminders_snapshot(self: &Arc) { - let reminders = match self.broker.list_pending_reminders() { - Ok(rows) => rows, - Err(e) => { - tracing::warn!(error = ?e, "emit_reminders_snapshot: list failed"); - return; - } - }; - self.emit_dashboard_event(DashboardEvent::RemindersChanged { - seq: self.next_seq(), - reminders, - }); - } - /// Emit a `CapabilitiesChanged` snapshot event. Called from the /// rebuild-queue worker after a `PermChange` / Capabilities entry /// commits the JSON file, so the P3RM1SS10NS tab updates live. @@ -930,7 +912,7 @@ impl Coordinator { /// Cheap when nothing changed (one `nixos-container list` + a /// `HashMap` diff + zero emits). pub async fn rescan_containers_and_emit(self: &Arc) { - let fresh = container_view::build_all(self).await; + let fresh = container_view::build_all().await; let mut last = self.last_containers.lock().await; let mut changed_or_new = Vec::new(); let mut removed = Vec::new(); diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index 00a65b4c..580feeba 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -32,7 +32,6 @@ mod meta_inputs; mod misc_api; pub(crate) mod permissions; mod questions; -mod reminders; mod schedules; mod state_files; mod state_snapshot; @@ -93,7 +92,6 @@ pub async fn serve( "/api/extra-forge-account", post(extra_forges::post_extra_forge_account), ) - .route("/api/reminders", get(reminders::api_reminders)) .route("/api/operator-inbox", get(misc_api::api_operator_inbox)) .route("/api/stats-hive", get(misc_api::api_stats_hive)) .route( @@ -212,14 +210,6 @@ pub async fn serve( "/api/github-account", post(matrix_accounts::post_github_account).get(matrix_accounts::get_github_account), ) - .route( - "/api/cancel-reminder/{id}", - post(reminders::post_cancel_reminder), - ) - .route( - "/api/retry-reminder/{id}", - post(reminders::post_retry_reminder), - ) .route("/api/request-spawn", post(misc_api::post_request_spawn)) .route("/api/op-send", post(misc_api::post_op_send)) .route("/api/meta-update", post(meta_inputs::post_meta_update)) diff --git a/hive-c0re/src/dashboard/reminders.rs b/hive-c0re/src/dashboard/reminders.rs deleted file mode 100644 index d61550b8..00000000 --- a/hive-c0re/src/dashboard/reminders.rs +++ /dev/null @@ -1,61 +0,0 @@ -//! Reminder endpoints for the dashboard. -//! -//! Lists pending reminders for the reminders tab, and lets the operator -//! cancel a pending reminder or reset its failure state so the scheduler -//! retries it on the next tick. - -use axum::{ - extract::{Path as AxumPath, State}, - http::StatusCode, - response::{IntoResponse, Response}, -}; - -use problem_details::ProblemDetails; - -use super::{AppState, error_problem, error_response}; - -pub(super) async fn api_reminders(State(state): State) -> Response { - match state.coord.broker.list_pending_reminders() { - Ok(rows) => axum::Json(rows).into_response(), - Err(e) => error_response(&format!("reminders: {e:#}")), - } -} - -pub(super) async fn post_cancel_reminder( - State(state): State, - AxumPath(id): AxumPath, -) -> Result { - match state.coord.broker.cancel_reminder(id) { - Ok(0) => Err(ProblemDetails::from_status_code(StatusCode::NOT_FOUND) - .with_detail(format!("reminder {id} not pending (already delivered?)"))), - Ok(_) => { - tracing::info!(%id, "operator cancelled reminder"); - state.coord.emit_reminders_snapshot(); - Ok((StatusCode::OK, "ok").into_response()) - } - Err(e) => Err(error_problem(&format!( - "cancel reminder {id} failed: {e:#}" - ))), - } -} - -/// Reset a pending reminder's failure state so the scheduler -/// retries it on the next tick. Useful when the failure was -/// transient (sqlite lock contention, disk full → freed up) and -/// the operator wants delivery to resume immediately instead of -/// the row sitting in attempt-count-capped purgatory. -pub(super) async fn post_retry_reminder( - State(state): State, - AxumPath(id): AxumPath, -) -> Result { - match state.coord.broker.reset_reminder_failure(id) { - Ok(0) => Err(ProblemDetails::from_status_code(StatusCode::NOT_FOUND) - .with_detail(format!("reminder {id} not pending (already delivered?)"))), - Ok(_) => { - tracing::info!(%id, "operator reset reminder failure for retry"); - state.coord.emit_reminders_snapshot(); - Ok((StatusCode::OK, "ok").into_response()) - } - Err(e) => Err(error_problem(&format!("retry reminder {id} failed: {e:#}"))), - } -} diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index d3fb9c92..b5451b05 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -219,14 +219,6 @@ pub enum DashboardEvent { seq: u64, schedules: Vec, }, - /// Full snapshot of all pending reminders. Emitted after every - /// reminder mutation: agent `remind` calls, operator cancel / retry, - /// and the scheduler tick after each delivery batch. Lets the - /// dashboard's reminders section stay live without polling. - RemindersChanged { - seq: u64, - reminders: Vec, - }, /// Full snapshot of capability grants (per-agent `Vec`). /// Emitted from the rebuild-queue worker after a `PermChange` /// `Capabilities` entry commits the JSON file. Lets the P3RM1SS10NS @@ -294,7 +286,6 @@ impl DashboardEvent { DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running", DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed", DashboardEvent::SchedulesChanged { .. } => "schedules_changed", - DashboardEvent::RemindersChanged { .. } => "reminders_changed", DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed", DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed", DashboardEvent::AuditEntryAdded { .. } => "audit_entry_added", @@ -420,10 +411,6 @@ mod tests { seq: 1, schedules: Vec::new(), }, - DashboardEvent::RemindersChanged { - seq: 1, - reminders: Vec::new(), - }, DashboardEvent::CapabilitiesChanged { seq: 1, caps: Vec::new(), diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index d510d222..c206b19f 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -3,7 +3,8 @@ //! a single agent (`for_agent`) or the whole hive (`hive_wide`). //! `Request::GetLooseEnds` from either the agent or manager socket //! lands here so the routing logic + age-seconds derivation stay in -//! one place. +//! one place. Reminders are agent-local (in-container store) and no +//! longer sourced from here. //! //! Call frequency is low (an agent doing self-introspection between //! turns), so the sweep happens fresh every time — no caching, no @@ -27,18 +28,17 @@ use hive_sh4re::wire_time::now_unix; /// root submits for top-level agents). Legacy rows with no recorded /// submitter count as the root's; /// - unanswered questions where `agent` is the asker (waiting on -/// someone) OR the target (owes a reply); -/// - pending reminders this agent scheduled (`owner == self`). +/// someone) OR the target (owes a reply). /// -/// Ordered `pending_messages` (when non-zero) → approvals → questions → -/// reminders within the returned vector. Within each kind, -/// source-of-truth ordering (sqlite's `pending()` queries return -/// newest-first within their indexes). +/// Ordered `pending_messages` (when non-zero) → approvals → questions +/// within the returned vector. Within each kind, source-of-truth +/// ordering (sqlite's `pending()` queries return newest-first within +/// their indexes). /// /// # Errors /// /// Propagates errors from `count_pending` and the pending-approval / -/// question / reminder sqlite queries. +/// question sqlite queries. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -85,25 +85,13 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { age_seconds: saturating_age(now, q.asked_at.timestamp()), }); } - for r in coord.broker.list_pending_reminders()? { - if r.agent != agent { - continue; - } - out.push(LooseEnd::Reminder { - id: r.id, - owner: r.agent, - message: r.message, - due_at: r.due_at, - age_seconds: saturating_age(now, r.created_at.timestamp()), - }); - } Ok(out) } /// Hive-wide loose-ends view: EVERY pending approval + EVERY -/// unanswered question + EVERY pending reminder. Manager surface -/// only; sub-agents can't see each other's threads via the agent -/// surface (`for_agent` filters by name). +/// unanswered question. Manager surface only; sub-agents can't see +/// each other's threads via the agent surface (`for_agent` filters by +/// name). pub fn hive_wide(coord: &Coordinator) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -125,15 +113,6 @@ pub fn hive_wide(coord: &Coordinator) -> Result> { age_seconds: saturating_age(now, q.asked_at.timestamp()), }); } - for r in coord.broker.list_pending_reminders()? { - out.push(LooseEnd::Reminder { - id: r.id, - owner: r.agent, - message: r.message, - due_at: r.due_at, - age_seconds: saturating_age(now, r.created_at.timestamp()), - }); - } Ok(out) } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a97aff7b..f8e2870d 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -45,8 +45,7 @@ pub(crate) use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, }; pub(crate) use workers::{ - agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler, - scheduled_prompts_worker, + agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker, }; use coordinator::{Coordinator, HiveEnv, ServeConfig}; @@ -214,7 +213,7 @@ fn matrix_sweep_banner(ctx: sweep_health::SweepFailure) -> String { } /// Start the coordinator daemon: open the broker, run migrations, spawn -/// background tasks (auto-update, vacuums, crash-watcher, reminder-scheduler, +/// background tasks (auto-update, vacuums, crash-watcher, scheduled-prompts, /// dashboard), then serve the admin socket until a signal arrives. #[allow( clippy::too_many_lines, @@ -522,9 +521,6 @@ async fn cmd_serve( // run_reconcile call register_agent on start; kill/destroy call // unregister_agent. No recurring poll needed — c0re owns the listeners. mcp_sockets::sync_on_start(coord.clone()).await; - // Reminder scheduler: drains due reminders + handles - // file_path payload persistence. See reminder_scheduler.rs. - reminder_scheduler::spawn(coord.clone()); // Scheduled-prompts worker: drains due scheduled_prompts rows // and fans the body out to each active target's inbox. See // scheduled_prompts_worker.rs. diff --git a/hive-c0re/src/questions.rs b/hive-c0re/src/questions.rs index d531ee0c..391d6451 100644 --- a/hive-c0re/src/questions.rs +++ b/hive-c0re/src/questions.rs @@ -186,14 +186,15 @@ pub fn handle_cancel_loose_end( Ok(()) } hive_sh4re::CancelLooseEndKind::Reminder => { - // Agent-socket path: ownership-only (cancel your own reminder). - let owner = coord - .broker - .cancel_reminder_as(id, canceller, false) - .map_err(|e| format!("{e:#}"))?; - tracing::info!(%id, %canceller, %owner, "reminder cancelled"); - coord.emit_reminders_snapshot(); - Ok(()) + // Reminders are now agent-local (in-container store) — the + // agent-mcp `cancel_loose_end` tool branches on this kind and + // dials the agent's own socket directly, never forwarding to + // hive-c0re. This arm should be unreachable in practice; kept + // only so the match stays exhaustive. + Err(format!( + "reminder {id}: reminders are handled locally by the agent, \ + not by hive-c0re" + )) } hive_sh4re::CancelLooseEndKind::Approval => { // Withdrawing an approval needs the grantable `approvals` diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 21089867..4bfb3bf4 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -159,7 +159,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { HostResponse::dags(dags) } HostRequest::List => HostResponse::list(lifecycle::list().await?), - HostRequest::AgentStatus => handle_agent_status(&coord).await, + HostRequest::AgentStatus => handle_agent_status().await, // The hive domain + per-surface public URLs are injected into // c0re's service env by hive-c0re.nix; surface them so the // operator CLI can fill in this hive's own identity (the @@ -298,8 +298,8 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result) -> HostResponse { - let rows = crate::container_view::build_all(coord) +async fn handle_agent_status() -> HostResponse { + let rows = crate::container_view::build_all() .await .into_iter() .map(|v| hive_sh4re::AgentStatusRow { @@ -308,7 +308,13 @@ async fn handle_agent_status(coord: &Arc) -> HostResponse { needs_update: v.needs_update, needs_login: v.needs_login, deployed_sha: v.deployed_sha, - pending_reminders: v.pending_reminders, + // Reminders are agent-local now; c0re has no cross-agent + // visibility into pending counts anymore. Stubbed + // to 0 rather than deleting the wire field outright — leaves + // `hivectl status`/the dashboard column intact syntactically, + // just always empty, until iris's frontend follow-up decides + // whether to drop the column entirely. + pending_reminders: 0, parent: v.parent, }) .collect(); diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index ffe3c81f..e9cfc3d0 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -23,7 +23,6 @@ use crate::coordinator::Coordinator; mod config_approvals; mod lifecycle_handlers; -mod reminders; mod schedules; pub(crate) use config_approvals::submit_merge_config_pr; @@ -34,7 +33,6 @@ use config_approvals::{handle_request_init_config, handle_request_update_meta_in use lifecycle_handlers::{ handle_kill, handle_list_descendants, handle_restart, handle_start, handle_update, }; -use reminders::{handle_remind, resolve_agent_state_target}; use schedules::{ EditSchedulePatch, handle_cancel_schedule, handle_edit_schedule, handle_fire_schedule_now, handle_list_schedules, handle_request_schedule_prompt, @@ -184,8 +182,8 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { /// Handle the subset of `Request` variants that are identical on both /// the agent socket and the manager socket. Returns `Some(response)` for /// every variant it handles; returns `None` for variants with socket-specific -/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup` -/// where the manager can target other agents) or for manager-only variants. +/// semantics (e.g. `GetLooseEnds`, where the manager can target other +/// agents) or for manager-only variants. /// /// The unified `dispatch` calls this first; the remaining arms (which gate /// on topology / capabilities / tool-groups) are handled there. @@ -234,11 +232,6 @@ pub(crate) async fn dispatch_shared( |()| hive_core_agent_sock::Response::Ok, ) } - hive_core_agent_sock::Request::Remind { - message, - timing, - file_path, - } => handle_remind(coord, agent, message, timing, file_path.as_deref()), hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text), hive_core_agent_sock::Request::GetAgentMeta { name } => { handle_get_agent_meta(coord, agent, name.as_ref()).await @@ -590,13 +583,6 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> Respo Request::GetLooseEnds { agent: target } => { handle_get_loose_ends(coord, agent, target.as_deref()) } - Request::CountPendingReminders { agent: target } => { - handle_count_pending_reminders(coord, agent, target.as_deref()) - } - Request::ReminderRollup { - since_secs, - agent: target, - } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), // Orchestration / diagnostics verbs — gated per-verb on tool-group // membership or topology (see `dispatch_orchestration`). _ => dispatch_orchestration(req, agent, coord).await, @@ -791,41 +777,38 @@ fn handle_get_loose_ends(coord: &Arc, agent: &str, target: Option<& } } -/// `CountPendingReminders` — resolve the target (own / subtree free, else -/// `QueryAgentState`) then count its pending reminders. -fn handle_count_pending_reminders( - coord: &Arc, - agent: &str, - target: Option<&str>, -) -> Response { - match resolve_agent_state_target(agent, target) { - Ok(name) => match coord.broker.count_pending_reminders_for(name) { - Ok(count) => Response::PendingRemindersCount { count }, - Err(e) => Response::Err { - message: format!("{e:#}"), - }, - }, - Err(message) => Response::Err { message }, - } -} - -/// `ReminderRollup` — resolve the target (own / subtree free, else -/// `QueryAgentState`) then roll up its reminders fired in the last -/// `since_secs`. -fn handle_reminder_rollup( - coord: &Arc, - agent: &str, - target: Option<&str>, - since_secs: u64, -) -> Response { - match resolve_agent_state_target(agent, target) { - Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) { - Ok(stats) => Response::ReminderRollup(stats), - Err(e) => Response::Err { - message: format!("{e:#}"), - }, - }, - Err(message) => Response::Err { message }, +/// Resolve the target agent name for a *named* `GetLooseEnds` query. Rules: +/// +/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed). +/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability. +/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise. +/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate. +fn resolve_agent_state_target<'a>( + caller: &'a str, + target: Option<&'a str>, +) -> Result<&'a str, String> { + match target { + None => Ok(caller), + Some("*") => Err( + "hive-wide query (agent=\"*\") is only valid for loose-ends; \ + not available for this query" + .to_owned(), + ), + Some(name) => { + // Own subtree (the root covers all) is visible without extra + // capability; `is_descendant_of` returns true for `name == caller`. + if crate::topology::is_descendant_of(name, caller) { + return Ok(name); + } + if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { + Ok(name) + } else { + Err(format!( + "agent `{caller}` cannot query `{name}`: not in its subtree and \ + `query_agent_state` capability is not granted" + )) + } + } } } diff --git a/hive-c0re/src/socket_server/reminders.rs b/hive-c0re/src/socket_server/reminders.rs deleted file mode 100644 index ae879da1..00000000 --- a/hive-c0re/src/socket_server/reminders.rs +++ /dev/null @@ -1,229 +0,0 @@ -//! Reminder request handling: the `Remind` handler, the shared -//! `store_remind` storage path with its pending-cap and large-body -//! auto-save dance, timing resolution, and the agent-state target -//! resolution shared by the loose-ends / reminder query handlers. - -use std::sync::Arc; - -use hive_core_agent_sock::Response; - -use crate::coordinator::Coordinator; - -pub(super) fn handle_remind( - coord: &Arc, - agent: &str, - message: &str, - timing: &hive_sh4re::ReminderTiming, - file_path: Option<&str>, -) -> Response { - match store_remind(coord, agent, message, timing, file_path) { - Ok(()) => Response::Ok, - Err(message) => Response::Err { message }, - } -} - -/// Shared remind-storage path used by both the agent and the manager -/// dispatchers. Validates timing, applies the auto-file overflow -/// dance (see [`prepare_remind_storage`]), and writes the reminder -/// row. Returns `Ok(())` on success, or a caller-ready error string -/// the dispatcher wraps in `*Response::Err`. -/// Maximum pending (un-delivered) reminders per agent. Exceeding this -/// causes `store_remind` to return an error so the agent knows to back -/// off instead of silently dropping. Override via -/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; set to `0` to disable the cap -/// (not recommended — a runaway agent can still flood the scheduler). -const DEFAULT_REMIND_MAX_PENDING: u64 = 50; - -fn remind_max_pending() -> u64 { - std::env::var("HIVE_REMIND_MAX_PENDING_PER_AGENT") - .ok() - .and_then(|s| s.trim().parse::().ok()) - .unwrap_or(DEFAULT_REMIND_MAX_PENDING) -} - -pub(crate) fn store_remind( - coord: &Arc, - agent: &str, - message: &str, - timing: &hive_sh4re::ReminderTiming, - file_path: Option<&str>, -) -> Result<(), String> { - let max = remind_max_pending(); - if max > 0 { - let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0); - if pending >= max { - return Err(format!( - "reminder rejected: agent `{agent}` already has {pending} pending \ - reminders (cap {max}). Cancel some via `cancel_loose_end` or wait \ - for them to fire before scheduling more. Override the cap with \ - `HIVE_REMIND_MAX_PENDING_PER_AGENT`." - )); - } - } - let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?; - let (stored_message, stored_path) = prepare_remind_storage(agent, message, file_path)?; - let id = coord - .broker - .store_reminder(agent, &stored_message, stored_path.as_deref(), due_at) - .map_err(|e| format!("failed to store reminder: {e:#}"))?; - tracing::info!(%id, %agent, %due_at, "reminder scheduled"); - coord.emit_reminders_snapshot(); - Ok(()) -} - -/// Decide what we actually store in the reminders row, applying the -/// same byte cap as the rest of the wire protocol -/// ([`crate::limits::MESSAGE_MAX_BYTES`]). Three outcomes: -/// -/// 1. Body within the cap → stored verbatim, with whatever `file_path` -/// the caller passed (None or Some). The scheduler honours -/// `file_path` at delivery time as before. -/// 2. Body over the cap, no caller `file_path` → auto-generate a path -/// under `/agents//state/reminders/auto-.md`, write the -/// body to disk now, store a short pointer hint as the message and -/// clear `file_path` (so the scheduler doesn't re-write at -/// delivery and overwrite the body with the hint). -/// 3. Body over the cap, caller provided `file_path` → honour the -/// caller's path: write the body to it now, store the same hint -/// and clear `file_path` for the same reason as (2). -/// -/// Returns `(stored_message, stored_file_path)` on success, or a -/// caller-ready error string on auto-save failure (which is the only -/// way a Remind request can be refused for size — the agent never has -/// to think about the cap). -fn prepare_remind_storage( - agent: &str, - message: &str, - file_path: Option<&str>, -) -> Result<(String, Option), String> { - if message.len() <= crate::limits::MESSAGE_MAX_BYTES { - return Ok((message.to_owned(), file_path.map(str::to_owned))); - } - let req_path = match file_path { - Some(p) => p.to_owned(), - None => auto_reminder_path(agent), - }; - let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path) - .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; - crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| { - format!("auto-save of large reminder body to `{req_path}` failed: {reason}") - })?; - let hint = format!( - "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]", - message.len() - ); - Ok((hint, None)) -} - -/// Generate a per-agent path for an auto-saved reminder body. Uses -/// `unix_nanos` plus the agent name to keep collisions infinitesimal -/// across the agent's own state subtree (we're not stamping a hostname -/// since hive-c0re is single-host). -fn auto_reminder_path(agent: &str) -> String { - let ts_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| d.as_nanos()); - format!("/agents/{agent}/state/reminders/auto-{ts_ns}.md") -} - -/// Resolve the target agent name for a *named* `GetLooseEnds` / -/// `CountPendingReminders` / `ReminderRollup` query. Rules: -/// -/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed). -/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability. -/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise. -/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate. -pub(super) fn resolve_agent_state_target<'a>( - caller: &'a str, - target: Option<&'a str>, -) -> Result<&'a str, String> { - match target { - None => Ok(caller), - Some("*") => Err( - "hive-wide query (agent=\"*\") is only valid for loose-ends; \ - not available for this query" - .to_owned(), - ), - Some(name) => { - // Own subtree (the root covers all) is visible without extra - // capability; `is_descendant_of` returns true for `name == caller`. - if crate::topology::is_descendant_of(name, caller) { - return Ok(name); - } - if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { - Ok(name) - } else { - Err(format!( - "agent `{caller}` cannot query `{name}`: not in its subtree and \ - `query_agent_state` capability is not granted" - )) - } - } - } -} - -/// 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), - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn auto_reminder_path_format() { - let p = auto_reminder_path("damocles"); - assert!(p.starts_with("/agents/damocles/state/reminders/auto-")); - assert!( - std::path::Path::new(&p) - .extension() - .is_some_and(|ext| ext.eq_ignore_ascii_case("md")) - ); - } - - #[test] - fn prepare_remind_storage_passthrough_under_cap() { - let (msg, fp) = prepare_remind_storage("foo", "small body", None).unwrap(); - assert_eq!(msg, "small body"); - assert_eq!(fp, None); - } - - #[test] - fn prepare_remind_storage_passthrough_with_caller_file_path() { - let (msg, fp) = - prepare_remind_storage("foo", "small", Some("/agents/foo/state/x.md")).unwrap(); - assert_eq!(msg, "small"); - assert_eq!(fp.as_deref(), Some("/agents/foo/state/x.md")); - } - - #[test] - fn resolve_agent_state_target_self_and_default_are_free() { - // No topology/capability state needed for these: `None` and the - // caller's own name resolve to the caller (`is_descendant_of` short- - // circuits to true when candidate == ancestor); `"*"` is rejected - // (the hive-wide sweep is handled by the loose-ends caller instead). - assert_eq!(resolve_agent_state_target("iris", None), Ok("iris")); - assert_eq!(resolve_agent_state_target("iris", Some("iris")), Ok("iris")); - assert!(resolve_agent_state_target("iris", Some("*")).is_err()); - } -} diff --git a/hive-c0re/src/stats/host_stats.rs b/hive-c0re/src/stats/host_stats.rs index 0e8bcaf5..1ccf80bd 100644 --- a/hive-c0re/src/stats/host_stats.rs +++ b/hive-c0re/src/stats/host_stats.rs @@ -250,7 +250,6 @@ mod tests { needs_update: false, needs_login, deployed_sha: None, - pending_reminders: 0, parent: None, active_model: None, } diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 1b8cca58..729bbeed 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -7,7 +7,6 @@ use std::sync::Mutex; use anyhow::{Context, Result}; -use chrono::{DateTime, Utc}; use hive_sh4re::wire_time::now_unix; use hive_sh4re::{InboxRow, Message}; @@ -30,18 +29,6 @@ CREATE TABLE IF NOT EXISTS messages ( CREATE INDEX IF NOT EXISTS idx_messages_undelivered ON messages (recipient, priority DESC, 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 @@ -52,12 +39,6 @@ CREATE TABLE IF NOT EXISTS kv ( /// 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); - /// 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 @@ -70,37 +51,6 @@ pub struct Delivery { 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, - pub due_at: DateTime, - pub created_at: DateTime, - /// 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, - /// 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 @@ -193,18 +143,6 @@ const BROKER_MIGRATIONS: &[Migration] = &[ COMMIT;", adds_column: Some(("messages", "priority")), }, - // v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap. - Migration { - sql: "ALTER TABLE reminders ADD COLUMN \ - attempt_count INTEGER NOT NULL DEFAULT 0", - adds_column: Some(("reminders", "attempt_count")), - }, - // v5: last_error on reminders — last delivery failure surfaced on the - // dashboard so a stuck reminder is visible without digging in logs. - Migration { - sql: "ALTER TABLE reminders ADD COLUMN last_error TEXT", - adds_column: Some(("reminders", "last_error")), - }, ]; impl Broker { @@ -850,309 +788,6 @@ impl Broker { } 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 { - 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> { - 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: hive_sh4re::wire_time::from_secs(row.get(4)?), - created_at: hive_sh4re::wire_time::from_secs(row.get(5)?), - last_error: row.get(6)?, - attempt_count: u32::try_from(attempts).unwrap_or(0), - }) - })?; - rows.collect::>>() - .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 { - 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 { - 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 { - 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 { - 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 { - let conn = self.conn.lock().unwrap(); - let owner: Option = 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> { - 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>(3)?, - )) - })?; - rows.collect::>>() - .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> { - 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> = 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 = Vec::with_capacity(items.len()); - for (id, agent, body) in items { - let r = (|| -> Result { - 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 - } } #[cfg(test)] diff --git a/hive-c0re/src/workers/mod.rs b/hive-c0re/src/workers/mod.rs index 75619f0b..ee5fa0ad 100644 --- a/hive-c0re/src/workers/mod.rs +++ b/hive-c0re/src/workers/mod.rs @@ -1,5 +1,5 @@ //! Background tasks and periodic sweeps: crash/login watcher, the -//! reminder and scheduled-prompt delivery loops, boot-time auto-update +//! scheduled-prompt delivery loop, boot-time auto-update //! reconcile, the agent-sockets.json writer loop, the MCP socket listener //! reconcile loop, and knowledge-repo sync. Each submodule is re-exported //! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged. @@ -9,5 +9,4 @@ pub mod auto_update; pub mod crash_watch; pub mod knowledge; pub mod mcp_sockets; -pub mod reminder_scheduler; pub mod scheduled_prompts_worker; diff --git a/hive-c0re/src/workers/reminder_scheduler.rs b/hive-c0re/src/workers/reminder_scheduler.rs deleted file mode 100644 index f05f8d92..00000000 --- a/hive-c0re/src/workers/reminder_scheduler.rs +++ /dev/null @@ -1,278 +0,0 @@ -//! Background loop that drains due reminders from the broker and -//! delivers them as inbox messages. 5s poll cadence, shutdown-aware. -//! File-path semantics (path translation, traversal + symlink defense, -//! pointer delivery): `docs/approvals.md::Reminder delivery`. - -use std::io::Write; -use std::os::unix::fs::OpenOptionsExt; -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use std::time::Duration; - -use crate::coordinator::Coordinator; - -/// Per-tick cap on reminders delivered. Anything over this stays due -/// in the table and gets picked up on the next tick — keeps a -/// 10k-deep backlog from flooding the broker (or hogging the broker -/// mutex) in one shot. 100/tick × 5s tick = sustained throughput cap -/// of ~20 reminders/sec; bump together if the loose-ends tracker -/// starts firing higher rates. -const REMINDER_BATCH_LIMIT: u64 = 100; - -/// Poll interval. Trade-off between latency on a freshly due reminder -/// and CPU spent on empty sweeps; 5s matches the original inline -/// scheduler. -const POLL_INTERVAL: Duration = Duration::from_secs(5); - -pub fn spawn(coord: Arc) { - let mut shutdown = coord.shutdown_rx(); - tokio::spawn(async move { - loop { - tick(&coord); - tokio::select! { - () = tokio::time::sleep(POLL_INTERVAL) => {} - _ = shutdown.changed() => { - tracing::info!("reminder scheduler: shutdown signal received"); - break; - } - } - } - }); -} - -fn tick(coord: &Arc) { - let due = match coord.broker.get_due_reminders(REMINDER_BATCH_LIMIT) { - Ok(rows) => rows, - Err(e) => { - tracing::warn!(error = ?e, "failed to query due reminders"); - return; - } - }; - if due.is_empty() { - return; - } - // Resolve body strings (file-path writes / inline) before entering - // the batch transaction so the DB lock is held as briefly as possible. - let items: Vec<(i64, String, String)> = due - .iter() - .map(|(agent, id, message, file_path)| { - let body = prepare_body(agent, message, file_path.as_deref()); - (*id, agent.clone(), body) - }) - .collect(); - // Single-transaction batch: one DB lock acquisition for N reminders - // instead of N sequential lock/unlock cycles. - let results = coord.broker.deliver_reminders_batch(&items); - let any_delivered = results.iter().any(Result::is_ok); - for ((id, agent, _body), result) in items.iter().zip(results.iter()) { - if let Err(e) = result { - let reason = format!("{e:#}"); - tracing::warn!( - reminder_id = id, - %agent, - error = %reason, - "failed to deliver reminder" - ); - // Persist the failure so the dashboard can surface it. - if let Err(persist_err) = coord.broker.record_reminder_failure(*id, &reason) { - tracing::warn!( - reminder_id = id, - error = ?persist_err, - "failed to persist reminder failure" - ); - } - } - } - // Emit after the batch so the dashboard's pending-reminders list - // updates when deliveries land (removes delivered rows). - if any_delivered { - coord.emit_reminders_snapshot(); - } -} - -/// Build the inbox body for a due reminder. When `file_path` is None -/// the body is the original message verbatim. When set, we attempt to -/// persist the message body to the requested file and return a short -/// pointer string instead. Failures (bad prefix, symlink escape, -/// write error, missing parent) fall back to inline delivery with a -/// noted warning so the reminder still fires. -fn prepare_body(agent: &str, message: &str, file_path: Option<&str>) -> String { - let Some(req_path) = file_path else { - return message.to_owned(); - }; - let host_path = match resolve_host_path(agent, req_path) { - Ok(p) => p, - Err(reason) => { - tracing::warn!(%agent, %req_path, %reason, "reminder file_path rejected; delivering inline"); - return inline_fallback(req_path, &format!("rejected: {reason}"), message); - } - }; - match write_payload(agent, &host_path, message) { - Ok(()) => { - let bytes = message.len(); - // debug! not info! — under load this would dominate the log. - tracing::debug!(%agent, path = %host_path.display(), bytes, "reminder body written to file"); - format!( - "reminder body persisted to `{req_path}` ({bytes} bytes); read with your filesystem tools" - ) - } - Err(reason) => { - tracing::warn!(%agent, path = %host_path.display(), %reason, "reminder file_path write failed; delivering inline"); - inline_fallback(req_path, &reason, message) - } - } -} - -fn inline_fallback(req_path: &str, reason: &str, message: &str) -> String { - format!("[reminder file_path '{req_path}' {reason}; delivering body inline]\n\n{message}") -} - -/// Persist `message` to `host_path` with the symlink-escape defenses -/// described in the module docs. Returns `Ok(())` on success, or a -/// human-readable reason string on any failure (caller logs + -/// inline-falls-back). `pub` because `socket_server::handle_remind` -/// reuses it for the at-remind-time auto-file path. -pub fn write_payload(agent: &str, host_path: &Path, message: &str) -> Result<(), String> { - let agent = hive_types::Ident::parse(agent) - .map_err(|e| format!("invalid agent name {agent:?}: {e}"))?; - let Some(parent) = host_path.parent() else { - return Err("internal: host path has no parent".to_owned()); - }; - std::fs::create_dir_all(parent).map_err(|e| format!("parent dir create failed: {e}"))?; - // Resolve symlinks in the parent chain, then re-verify the - // canonical form still lives under the agent's host state root — - // catches `ln -s /etc state/escape` style attacks. - let parent_canonical = parent - .canonicalize() - .map_err(|e| format!("parent canonicalize failed: {e}"))?; - let agent_root = Coordinator::agent_notes_dir(&agent) - .canonicalize() - .map_err(|e| format!("agent state root canonicalize failed: {e}"))?; - if !parent_canonical.starts_with(&agent_root) { - return Err(format!( - "symlink escape: canonical parent `{}` outside agent root `{}`", - parent_canonical.display(), - agent_root.display() - )); - } - let basename = host_path - .file_name() - .ok_or_else(|| "missing basename".to_owned())?; - let target = parent_canonical.join(basename); - // O_NOFOLLOW on the final component refuses to open if the - // basename is itself an existing symlink. Combined with the - // canonicalize-parent check above, no symlink anywhere in the - // path can redirect the write. - let mut file = std::fs::OpenOptions::new() - .write(true) - .create(true) - .truncate(true) - .custom_flags(libc::O_NOFOLLOW) - .open(&target) - .map_err(|e| format!("open failed: {e}"))?; - file.write_all(message.as_bytes()) - .map_err(|e| format!("write failed: {e}"))?; - Ok(()) -} - -/// Container-visible state prefix the caller's `file_path` must live -/// under. Every agent sees its state at `/agents//state/` -/// (see `lifecycle::set_nspawn_flags`). Auto-file paths use the same -/// prefix so the round-trip is symmetric. -#[must_use] -pub fn container_state_prefix(agent: &str) -> String { - format!("/agents/{agent}/state/") -} - -/// Map an agent-visible container path to the matching host path, -/// validating that it lives under the agent's own state subtree, has -/// a non-empty relative tail, and doesn't try to traverse out via -/// `..`. Returns the host `PathBuf` on success, or a human-readable -/// reason string on rejection. `pub` so `socket_server::handle_remind` -/// can reuse it for the at-remind-time auto-file path. -pub fn resolve_host_path(agent: &str, req_path: &str) -> Result { - let agent = hive_types::Ident::parse(agent) - .map_err(|e| format!("invalid agent name {agent:?}: {e}"))?; - let prefix = container_state_prefix(agent.as_str()); - let Some(rel) = req_path.strip_prefix(&prefix) else { - return Err(format!( - "must be absolute and under `{prefix}` (got `{req_path}`)" - )); - }; - if rel.is_empty() { - return Err("file_path must include a filename, not just the state dir".to_owned()); - } - let rel_path = Path::new(rel); - for comp in rel_path.components() { - match comp { - std::path::Component::Normal(_) => {} - other => { - return Err(format!( - "path component `{other:?}` not allowed (no traversal / absolute / root)" - )); - } - } - } - Ok(Coordinator::agent_notes_dir(&agent).join(rel_path)) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn rejects_paths_outside_agent_state() { - assert!(resolve_host_path("foo", "/etc/passwd").is_err()); - assert!(resolve_host_path("foo", "/agents/bar/state/x.md").is_err()); - assert!(resolve_host_path("foo", "relative.md").is_err()); - } - - #[test] - fn rejects_traversal() { - assert!(resolve_host_path("foo", "/agents/foo/state/../../etc/passwd").is_err()); - assert!(resolve_host_path("foo", "/agents/foo/state/./x.md").is_err()); - } - - #[test] - fn rejects_empty_relative_tail() { - // Trailing slash → empty tail. Used to fall through to - // create_dir_all + write-to-dir → confusing inline fallback; - // explicit reject gives a cleaner log. - let err = resolve_host_path("foo", "/agents/foo/state/").unwrap_err(); - assert!(err.contains("must include a filename"), "got: {err}"); - } - - #[test] - fn accepts_well_formed_path() { - let p = resolve_host_path("foo", "/agents/foo/state/reminders/123.md").unwrap(); - assert_eq!( - p, - PathBuf::from("/var/lib/hyperhive/agents/foo/state/reminders/123.md") - ); - } - - #[test] - fn manager_uses_container_name_prefix() { - // Manager's container view of its state is at `/agents/ruth/state/`. - assert_eq!(container_state_prefix("ruth"), "/agents/ruth/state/"); - let p = resolve_host_path("ruth", "/agents/ruth/state/reminders/x.md").unwrap(); - assert_eq!( - p, - PathBuf::from("/var/lib/hyperhive/agents/ruth/state/reminders/x.md") - ); - assert!(resolve_host_path("ruth", "/state/x.md").is_err()); - } - - #[test] - fn prepare_body_passthrough_when_no_file_path() { - let s = prepare_body("foo", "hello world", None); - assert_eq!(s, "hello world"); - } - - #[test] - fn prepare_body_falls_back_inline_on_bad_path() { - let s = prepare_body("foo", "payload", Some("/etc/passwd")); - assert!(s.starts_with("[reminder file_path '/etc/passwd' rejected:")); - assert!(s.contains("payload")); - } -} diff --git a/hive-core-agent-sock/src/lib.rs b/hive-core-agent-sock/src/lib.rs index 90e18c4d..cb390201 100644 --- a/hive-core-agent-sock/src/lib.rs +++ b/hive-core-agent-sock/src/lib.rs @@ -9,7 +9,7 @@ use hive_sh4re::{ CancelLooseEndKind, ContainerInfo, DeliveredMessage, InboxRow, JournalPriority, LooseEnd, - MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule, + MatrixIdentity, SchedulePromptPayload, WireSchedule, }; use hive_types::Ident; use serde::{Deserialize, Serialize}; @@ -81,18 +81,6 @@ pub enum Request { /// back via `HelperEvent::QuestionAnswered`: see /// `docs/conventions.md::Question routing (Ask / Answer)`. Answer { id: i64, answer: String }, - /// Schedule a reminder message to be delivered to this agent at a - /// future time. The reminder lands in the agent's inbox as an auto-sent - /// message from `"reminder"`. Use for agent follow-ups (e.g. check task - /// status, retry failed operation). Message length is limited; pass - /// `file_path` to store in a file and get a path-reference message - /// instead. - Remind { - message: String, - timing: ReminderTiming, - #[serde(default)] - file_path: Option, - }, /// Loose-ends view. On the agent socket: `None` = self; direct /// children are always accessible; non-children require the /// `query_agent_state` capability — rejected with an error otherwise; @@ -103,33 +91,6 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] agent: Option, }, - /// Count of pending (un-delivered) reminders. On the agent socket: - /// same target rules as `GetLooseEnds` (self/children free; - /// non-children require `query_agent_state`; `"*"` rejected). - /// On the manager socket: `None` = self, any name = that agent. - /// Used by the harness's per-turn stats sink. - CountPendingReminders { - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, - /// Reminder statistics: counts of scheduled, delivered, and pending - /// reminders over a time window. `since_secs` filters to reminders - /// created in the last N seconds (0 = all). On the agent socket: - /// same target rules as `GetLooseEnds` (self/children free; - /// non-children require `query_agent_state`; `"*"` rejected). - /// On the manager socket: `None` = self, any name = that agent. - ReminderRollup { - /// Only count reminders created in the last N seconds from now. - /// Pass 0 to include all reminders. - #[serde(default)] - since_secs: u64, - /// Whose reminders to roll up. `None` = the caller's own. - /// `Some("")` = that agent's (requires `query_agent_state` - /// capability on the agent socket; always available on the manager - /// socket). - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, /// Set a free-text status string visible on the dashboard. The harness /// writes `{state_dir}/hyperhive-status` locally before sending this /// request; hive-c0re just triggers a dashboard rescan on receipt. @@ -308,10 +269,6 @@ pub enum Response { /// `GetLooseEnds` result: list of loose ends pending against /// this agent. Ordered newest-first within each kind. LooseEnds { loose_ends: Vec }, - /// `CountPendingReminders` result. - PendingRemindersCount { count: u64 }, - /// `ReminderRollup` result: reminder activity stats for the agent. - ReminderRollup(ReminderStats), /// `GetAgentMeta` result. Per-field semantics + serde defaults /// live in `docs/conventions.md::Agent metadata`. AgentMeta {