diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 177379c7..0ca18172 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -35,9 +35,8 @@ pub use args::{ pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; use render::{ - dial_agent_socket, format_matrix_summary, local_reminders, local_todos, loose_end_kind_label, - mark_local_todo_done, matrix_unread_summary, parse_loose_end_kind, render_loose_ends, - reply_err, + format_matrix_summary, local_todos, loose_end_kind_label, mark_local_todo_done, + matrix_unread_summary, parse_loose_end_kind, render_loose_ends, reply_err, }; /// Write (or remove) the status file in the agent's own `state/` directory. @@ -344,12 +343,6 @@ impl AgentServer { if is_self_query && let Some(todos) = local_todos().await { loose_ends.extend(todos); } - // Merge local pending reminders — same self-query-only - // restriction: a manager asking for a child's loose-ends no longer - // sees the child's reminders, matching the todos precedent above). - if is_self_query && let Some(reminders) = local_reminders().await { - loose_ends.extend(reminders); - } annotate_retries(render_loose_ends(&loose_ends), retries) }) .await @@ -453,26 +446,6 @@ impl AgentServer { Err(e) => return e, }; let kind_label = loose_end_kind_label(kind); - // Reminders are harness-local — dial the in-agent socket - // directly instead of the broker; every other kind - // (question/approval) still lives in c0re. - if kind == hive_sh4re::CancelLooseEndKind::Reminder { - return match dial_agent_socket(&hive_agent_sock::Request::CancelReminder { id }) - .await - { - Some(hive_agent_sock::Response::Acked { count }) if count > 0 => { - format!("cancelled {kind_label} {id}") - } - Some(hive_agent_sock::Response::Acked { .. }) => { - format!("cancel_loose_end failed: no pending {kind_label} {id}") - } - Some(hive_agent_sock::Response::Err { message }) => { - format!("cancel_loose_end failed: {message}") - } - Some(other) => format!("cancel_loose_end unexpected response: {other:?}"), - None => "cancel_loose_end failed: in-agent socket unavailable".to_owned(), - }; - } let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id }) .await; @@ -543,22 +516,17 @@ impl AgentServer { (Some(s), None) => hive_sh4re::ReminderTiming::InSeconds { seconds: s }, (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; - // Reminders are harness-local — dial the in-agent socket - // directly instead of the broker. - match dial_agent_socket(&hive_agent_sock::Request::StoreReminder { - message: args.message, - timing, - file_path: args.file_path, - }) - .await - { - Some(hive_agent_sock::Response::Ok) => "reminder scheduled".to_owned(), - Some(hive_agent_sock::Response::Err { message }) => { - format!("remind failed: {message}") - } - Some(other) => format!("remind unexpected response: {other:?}"), - None => "remind failed: in-agent socket unavailable".to_owned(), - } + let (resp, retries) = self + .dispatch(hive_core_agent_sock::Request::Remind { + message: args.message, + timing, + file_path: args.file_path, + }) + .await; + annotate_retries( + format_ack(resp, "remind", "reminder scheduled".to_string()), + retries, + ) }) .await } diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index fb9821cc..8402c995 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -304,10 +304,8 @@ pub(super) async fn matrix_unread_summary() -> Option> { /// best-effort, single-shot (no retry, unlike the broker's /// `client::request_retried`): this socket lives in the SAME container, so /// a connect failure means the harness itself isn't up, which a retry -/// won't fix within a tool call's budget. Shared by [`local_todos`], -/// [`mark_local_todo_done`], and [`local_reminders`]; `remind`/ -/// `cancel_loose_end`(reminder) in `mod.rs` dial it directly since their -/// happy path is a plain `Ok`/`Err`, not a `Vec` to merge. +/// won't fix within a tool call's budget. Shared by [`local_todos`] and +/// [`mark_local_todo_done`]. pub(super) async fn dial_agent_socket( req: &hive_agent_sock::Request, ) -> Option { @@ -337,16 +335,6 @@ pub(super) async fn local_todos() -> Option> { } } -/// Query the harness's in-agent socket for this agent's local pending -/// reminders — was a broker query before reminders moved in-container. -/// Same best-effort contract as [`local_todos`]. -pub(super) async fn local_reminders() -> Option> { - match dial_agent_socket(&hive_agent_sock::Request::ListReminders).await? { - hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends), - _ => None, - } -} - /// Mark one of this agent's local todos (loose-ends v2) done by id, via /// the harness's in-agent socket — reachable through `cancel_loose_end` /// kind `"todo"` so clearing a todo never has to shell out through a diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index a94a5908..dc7bd7a8 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -1,15 +1,14 @@ //! Wire types for the *in-agent* socket, served by the hive-agent harness //! to the in-container producers (matrix / bash MCP daemons) and -//! `forge_notify`. Carries the loose-ends-v2 *todo* op family plus the -//! harness-local *reminder* op family; more in-agent request families may -//! be added over time (the socket is deliberately named for the agent, -//! not the todos). +//! `forge_notify`. Currently carries the loose-ends-v2 *todo* op family; +//! more in-agent request families may be added over time (the socket is +//! deliberately named for the agent, not the todos). //! //! Distinct from `hive-core-agent-sock`, the *host*-served core↔agent //! protocol on `/run/hive/mcp.sock`: this socket never leaves the -//! container. The harness owns the todo + reminder stores locally and -//! signals its own turn loop directly, so hive-c0re is not in either path — -//! no broker round-trip, no long-poll, no marker files. +//! container. The harness owns the todo store locally and signals its own +//! turn loop directly, so hive-c0re is not in the todo path — no broker +//! round-trip, no long-poll, no marker files. use serde::{Deserialize, Serialize}; @@ -59,30 +58,6 @@ pub enum Request { }, /// The agent marks one of its own todos done, by id. MarkTodoDone { id: i64 }, - /// Schedule a reminder that fires into this agent's own turn loop at - /// `timing` (harness-local — no broker round-trip). Same semantics as - /// the old broker `Remind` request. `file_path`, when set, is where the - /// harness persists an over-cap body instead of inlining it. - StoreReminder { - message: String, - timing: hive_sh4re::ReminderTiming, - #[serde(default, skip_serializing_if = "Option::is_none")] - file_path: Option, - }, - /// List this agent's pending reminders (single-agent scope — unlike - /// the old broker query, there's no cross-agent `agent` filter here). - ListReminders, - /// Cancel one of this agent's own pending reminders by id, before it - /// fires. - CancelReminder { id: i64 }, - /// This agent's pending-reminder count — used by the pre-`remind` cap - /// check and the harness's own turn-stats sink. - CountPendingReminders, - /// Scheduled/delivered/pending reminder counts over a trailing - /// `since_secs` window (`0` = all time) — the harness's own `/api/stats` - /// reminder-activity chart data (was `hive-core-agent-sock`'s - /// `ReminderRollup` against the broker; now served from the local store). - ReminderRollup { since_secs: u64 }, } /// A response on the in-agent socket. Serialised with a `kind` tag, @@ -94,13 +69,8 @@ pub enum Response { Ok, /// Op succeeded and touched `count` rows (clear / mark-done). Acked { count: u64 }, - /// `ListTodos` / `ListReminders` result (the latter wraps each row as - /// [`LooseEnd::Reminder`]). + /// `ListTodos` result. LooseEnds { loose_ends: Vec }, - /// `CountPendingReminders` result. - PendingRemindersCount { count: u64 }, - /// `ReminderRollup` result. - ReminderRollup { stats: hive_sh4re::ReminderStats }, /// Op failed; `message` is operator-facing. Err { message: String }, } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index d4b546a3..f167da3c 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -20,8 +20,6 @@ mod mcp_config; mod paths; mod plugins; mod prompt; -mod reminder_timer; -mod reminders; mod serve_common; mod stats; mod stream_enrich; @@ -216,11 +214,6 @@ fn graceful_stop_message() -> hive_sh4re::DeliveredMessage { /// enum so `serve_loop` can pattern-match without seeing it directly. enum RecvOutcome { /// Long-poll returned at least one message; first one is detached. - /// Also reused for a fired reminder: `reminder_timer` pushes a *real* - /// `DeliveredMessage` (unlike `LocalTodo`'s synthetic hint) since the - /// body/id is per-row data the producer already resolved, so the - /// select arm wraps it straight into this variant — no dedicated - /// `LocalReminder` variant needed. Message(hive_sh4re::DeliveredMessage), /// Long-poll timed out cleanly (empty `Messages` response). Caller /// sleeps then retries. @@ -332,13 +325,15 @@ impl Surface for AgentSurface { Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), _ => None, }; - // Reminders are harness-local — dial the in-agent socket directly - // instead of the broker. - let reminders = - match todo_server::dial(&hive_agent_sock::Request::CountPendingReminders).await { - Some(hive_agent_sock::Response::PendingRemindersCount { count }) => Some(count), - _ => None, - }; + let reminders = match client::request::<_, Response>( + socket, + &Request::CountPendingReminders { agent: None }, + ) + .await + { + Ok(Response::PendingRemindersCount { count }) => Some(count), + _ => None, + }; (threads, reminders) } @@ -458,41 +453,18 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { tracing::error!(error = %e, "web_ui::serve exited with error"); } }); - // Harness-local reminders: a due reminder fires - // straight into an mpsc channel the serve loop races against the - // broker long-poll (unlike todos, a fire carries real per-row data, - // so a bare `Notify` doesn't fit — see `reminder_timer` docs). - // Best-effort: `reminder_timer::run` parks forever (never sends) - // instead of exiting when the store can't open, so the channel - // never observes a "closed" state. Opened before the todo socket - // below so the same `Arc` can be handed to its request dispatch - // (reminder ops share the todo socket/listener). - let (reminder_tx, reminder_rx) = tokio::sync::mpsc::unbounded_channel(); - let reminder_store = match reminders::Reminders::open(&paths::reminders_db()) { - Ok(store) => Some(Arc::new(store)), - Err(e) => { - tracing::error!(error = ?e, "open reminders db failed — reminder delivery disabled"); - None - } - }; - tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx)); - // In-agent todo socket (loose-ends v2 + harness-local reminders): the - // harness owns the todo + reminder stores locally and serves the - // in-container producers on `HIVE_AGENT_SOCKET`. A new/changed todo - // upsert fires `todo_wake` so the serve loop drives a turn directly — - // no broker round-trip, no marker files. Best-effort: if the todos - // store can't open, the whole socket isn't served (reminder ops ride - // along on the same listener, so they're gated on the same store — - // acceptable since a from-scratch harness boot either has a writable - // harness dir or doesn't). + // In-agent todo socket (loose-ends v2): the harness owns the todo store + // locally and serves the in-container producers on `HIVE_AGENT_SOCKET`. + // A new/changed upsert fires `todo_wake` so the serve loop drives a turn + // directly — no broker round-trip, no marker files. Best-effort: if the + // store can't open, the socket just isn't served. let todo_wake = Arc::new(tokio::sync::Notify::new()); match todos::Todos::open(&paths::todos_db()) { Ok(store) => { let store = Arc::new(store); let wake = todo_wake.clone(); - let reminders = reminder_store.clone(); tokio::spawn(async move { - if let Err(e) = todo_server::run(store, wake, reminders).await { + if let Err(e) = todo_server::run(store, wake).await { tracing::error!(error = %e, "in-agent todo socket exited with error"); } }); @@ -518,7 +490,6 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { stats, &files, todo_wake, - reminder_rx, ) .await } @@ -541,7 +512,6 @@ async fn serve_loop( stats: Option, files: &turn::TurnFiles, todo_wake: Arc, - mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); S::requeue_inflight(socket).await; @@ -566,7 +536,6 @@ async fn serve_loop( biased; o = S::recv_next(socket) => o, () = todo_wake.notified() => RecvOutcome::LocalTodo, - Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm), } } { RecvOutcome::Message(first) => first, diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index 9756eb38..2d7c7b99 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -48,14 +48,6 @@ pub fn todos_db() -> PathBuf { harness_dir().join("hyperhive-todos.sqlite") } -/// Harness-local reminder store. Same rationale as [`todos_db`] — mutable -/// per-agent state the harness owns, kept out of the append-only events -/// sink. -#[must_use] -pub fn reminders_db() -> PathBuf { - harness_dir().join("hyperhive-reminders.sqlite") -} - /// Per-turn config dir for the regenerated claude-{mcp-config,settings, /// system-prompt} files the harness drops before each turn. Set by /// systemd via `RuntimeDirectory = "hive-config"`: a per-service runtime diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs deleted file mode 100644 index 326f8ef5..00000000 --- a/hive-agent/src/reminder_timer.rs +++ /dev/null @@ -1,303 +0,0 @@ -//! Delivery half of the harness-local reminders store. Polls [`Reminders`] -//! for due rows and pushes each as a -//! [`hive_sh4re::DeliveredMessage`] down an mpsc channel the serve loop -//! races against the broker long-poll — mirrors `todo_server`'s `Notify` -//! wake, but a reminder fire carries real per-row data (message/id), so a -//! bare `Notify` doesn't fit; the channel carries the finished message -//! instead. -//! -//! File-path semantics (large-body auto-save, delivery-time persist) -//! port the old `hive-c0re::workers::reminder_scheduler` / -//! `socket_server::reminders` logic, simplified: this runs *inside* the -//! agent's own container now, so the symlink-escape defense that guarded -//! `hive-c0re` (running outside the container, writing into a path an -//! agent claimed was its own) is moot — the harness IS the agent, -//! already sandboxed by the container boundary. Still keeps the cheap -//! belt-and-suspenders checks: `file_path` must resolve under -//! [`crate::paths::state_dir`] with no `..`/absolute components in the -//! relative tail. - -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use std::time::Duration; - -use tokio::sync::mpsc; - -use crate::reminders::Reminders; - -/// Per-tick cap on reminders delivered — mirrors the old c0re -/// `REMINDER_BATCH_LIMIT`, same rationale (bound a deep backlog's -/// per-tick cost). -const REMINDER_BATCH_LIMIT: u64 = 100; - -/// Poll interval — matches the old c0re `POLL_INTERVAL`. -const POLL_INTERVAL: Duration = Duration::from_secs(5); - -/// Same cap the broker used to enforce on `send`/`ask`/`remind` bodies -/// ([`hive-c0re::agent_config::limits::MESSAGE_MAX_BYTES`], not -/// reachable from here — hive-agent doesn't depend on hive-c0re). -/// Duplicated rather than shared: this is the last remaining reminder -/// caller of that constant once the c0re-side store is deleted (commit 6). -const REMINDER_BODY_MAX_BYTES: usize = 4096; - -/// Maximum pending (undelivered) reminders this agent may hold at once. -/// Exceeding this makes `store` return an error so the caller backs off -/// instead of silently flooding the table. Override via -/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; `0` disables the cap. -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) -} - -/// Run the delivery timer. `store: None` (reminders db failed to open at -/// boot) parks forever instead of looping — keeps `tx` alive so the -/// serve loop's receiver never observes a closed channel (which would -/// otherwise resolve immediately every iteration and busy-loop the -/// select), while cleanly disabling delivery. -pub async fn run( - store: Option>, - tx: mpsc::UnboundedSender, -) { - let Some(store) = store else { - tracing::error!("reminders db unavailable — reminder delivery disabled"); - std::future::pending::<()>().await; - return; - }; - loop { - tick(&store, &tx); - tokio::time::sleep(POLL_INTERVAL).await; - } -} - -fn tick(store: &Reminders, tx: &mpsc::UnboundedSender) { - let now = hive_sh4re::wire_time::now_unix(); - let due = match store.due(now, REMINDER_BATCH_LIMIT) { - Ok(rows) => rows, - Err(e) => { - tracing::warn!(error = ?e, "failed to query due reminders"); - return; - } - }; - for r in due { - let body = prepare_body(&r.message, r.file_path.as_deref()); - let dm = hive_sh4re::DeliveredMessage { - from: "reminder".into(), - body, - id: 0, - redelivered: false, - in_reply_to: None, - }; - if tx.send(dm).is_err() { - // Receiver (serve loop) is gone — process is shutting down. - // Leave the row pending; nothing else to do here. - tracing::warn!(reminder_id = r.id, "reminder delivery channel closed"); - return; - } - if let Err(e) = store.mark_delivered(r.id) { - tracing::warn!(reminder_id = r.id, error = ?e, "failed to mark reminder delivered"); - } - } -} - -/// Store a new reminder, applying the pending cap + large-body auto-save -/// dance. Returns the new row id, or a caller-ready error string (used -/// directly as `Response::Err.message` by the in-agent socket dispatch). -pub fn store( - store: &Reminders, - message: &str, - timing: &hive_sh4re::ReminderTiming, - file_path: Option<&str>, -) -> Result { - let max = remind_max_pending(); - if max > 0 { - let pending = store.count_pending().unwrap_or(0); - if pending >= max { - return Err(format!( - "reminder rejected: already {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(message, file_path)?; - store - .store(&stored_message, stored_path.as_deref(), due_at) - .map_err(|e| format!("failed to store reminder: {e:#}")) -} - -/// Decide what to actually persist in the reminder row (see the old -/// `prepare_remind_storage` this ports — same three outcomes: verbatim -/// under the cap, auto-saved-to-a-generated-path over cap with no -/// caller path, or auto-saved-to-the-caller's-path over cap). -fn prepare_remind_storage( - message: &str, - file_path: Option<&str>, -) -> Result<(String, Option), String> { - if message.len() <= REMINDER_BODY_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(), - }; - let path = resolve_state_path(&req_path) - .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; - write_payload(&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 fresh auto-save path under this agent's own -/// `state/reminders/` dir. -fn auto_reminder_path() -> String { - let ts_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| d.as_nanos()); - crate::paths::state_dir() - .join("reminders") - .join(format!("auto-{ts_ns}.md")) - .to_string_lossy() - .into_owned() -} - -/// Build the delivered body for a due reminder: verbatim when -/// `file_path` is unset, otherwise persist to that path and return a -/// short pointer (falling back to inline delivery with a warning note -/// on any rejection/write failure, so the reminder still fires). -fn prepare_body(message: &str, file_path: Option<&str>) -> String { - let Some(req_path) = file_path else { - return message.to_owned(); - }; - let path = match resolve_state_path(req_path) { - Ok(p) => p, - Err(reason) => return inline_fallback(req_path, &format!("rejected: {reason}"), message), - }; - match write_payload(&path, message) { - Ok(()) => format!( - "reminder body persisted to `{req_path}` ({} bytes); read with your filesystem tools", - message.len() - ), - Err(reason) => 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}") -} - -/// Validate `req_path` is absolute, lives under this agent's own -/// `state_dir()`, and its relative tail has no traversal / absolute -/// components. Returns the (already-real, no host/container -/// translation needed in-container) path. -fn resolve_state_path(req_path: &str) -> Result { - let base = crate::paths::state_dir(); - let path = Path::new(req_path); - if !path.is_absolute() { - return Err(format!("must be absolute (got `{req_path}`)")); - } - let Ok(rel) = path.strip_prefix(&base) else { - return Err(format!( - "must live under `{}` (got `{req_path}`)", - base.display() - )); - }; - if rel.as_os_str().is_empty() { - return Err("file_path must include a filename, not just the state dir".to_owned()); - } - for comp in rel.components() { - match comp { - std::path::Component::Normal(_) => {} - other => { - return Err(format!( - "path component `{other:?}` not allowed (no traversal / absolute / root)" - )); - } - } - } - Ok(path.to_path_buf()) -} - -/// Write `message` to `path`, creating parent dirs as needed. -fn write_payload(path: &Path, message: &str) -> Result<(), String> { - let Some(parent) = path.parent() else { - return Err("internal: path has no parent".to_owned()); - }; - std::fs::create_dir_all(parent).map_err(|e| format!("parent dir create failed: {e}"))?; - std::fs::write(path, message).map_err(|e| format!("write failed: {e}")) -} - -/// Resolve the `due_at` unix timestamp for a `StoreReminder` request. -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 resolve_state_path_rejects_non_absolute() { - assert!(resolve_state_path("relative.md").is_err()); - } - - #[test] - fn resolve_due_at_in_seconds_is_close_to_now_plus_n() { - let due = resolve_due_at(&hive_sh4re::ReminderTiming::InSeconds { seconds: 60 }).unwrap(); - let now = hive_sh4re::wire_time::now_unix(); - assert!((due - now - 60).abs() <= 2, "due={due} now={now}"); - } - - #[test] - fn resolve_due_at_at_passes_through() { - let due = resolve_due_at(&hive_sh4re::ReminderTiming::At { - unix_timestamp: 123_456, - }) - .unwrap(); - assert_eq!(due, 123_456); - } - - #[test] - fn prepare_remind_storage_passthrough_under_cap() { - let (msg, fp) = prepare_remind_storage("small body", None).unwrap(); - assert_eq!(msg, "small body"); - assert_eq!(fp, None); - } - - #[test] - fn prepare_body_passthrough_when_no_file_path() { - assert_eq!(prepare_body("hello world", None), "hello world"); - } - - #[test] - fn prepare_body_falls_back_inline_on_bad_path() { - let s = prepare_body("payload", Some("/etc/passwd")); - assert!(s.starts_with("[reminder file_path '/etc/passwd' rejected:")); - assert!(s.contains("payload")); - } -} diff --git a/hive-agent/src/reminders.rs b/hive-agent/src/reminders.rs deleted file mode 100644 index 4f4936f3..00000000 --- a/hive-agent/src/reminders.rs +++ /dev/null @@ -1,366 +0,0 @@ -//! Harness-local reminder store — the persistent, DB-backed half of the -//! in-container reminders migration. Mirrors `todos.rs`'s shape: one -//! sqlite db under the harness dir, single-agent -//! scope (no `agent` column — every row belongs to this agent, unlike the -//! old c0re-side store which served every agent in the hive). -//! -//! Unlike todos (discovered passively via `get_loose_ends`), a reminder is -//! *active push*: it must fire into the turn loop at `due_at` with its own -//! message body (see `reminder_timer.rs`). Delivery is *soft*-deleted -//! (`sent_at` set, not `DELETE`) so the harness's own `/api/stats` reminder -//! rollup can still report scheduled/delivered/pending counts over a -//! trailing window (mirrors the old c0re `ReminderStats` shape exactly). A -//! periodic prune (`vacuum.rs`) reaps old delivered rows so the table -//! doesn't grow unbounded. - -use std::path::Path; -use std::sync::Mutex; - -use anyhow::{Context, Result}; -use hive_sh4re::ReminderStats; -use hive_sh4re::wire_time::now_unix; -use rusqlite::{Connection, params}; - -const SCHEMA: &str = r" -CREATE TABLE IF NOT EXISTS reminders ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - 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 (due_at) WHERE sent_at IS NULL; -"; - -/// One reminder row. `file_path`, when set, is where the harness persists -/// the body instead of inlining it at delivery time (see the old -/// `store_remind`/`prepare_body` split this mirrors, now folded into -/// `reminder_timer.rs`). -#[derive(Debug, Clone)] -pub struct Reminder { - pub id: i64, - pub message: String, - pub file_path: Option, - pub due_at: i64, - pub created_at: i64, -} - -/// The harness-local reminder store. Cheap to share behind an `Arc`; the -/// inner connection is guarded by a `Mutex` (reminder ops are short sqlite -/// writes, same as `Todos`). -pub struct Reminders { - conn: Mutex, -} - -impl Reminders { - /// Open (creating if needed) the reminder store at `path`. - /// - /// # Errors - /// - /// Propagates sqlite open / schema-apply failures. - pub fn open(path: &Path) -> Result { - let conn = Connection::open(path) - .with_context(|| format!("open reminders db {}", path.display()))?; - conn.execute_batch(SCHEMA) - .context("apply reminders schema")?; - Ok(Self { - conn: Mutex::new(conn), - }) - } - - /// Store a new pending reminder, due at `due_at` (unix seconds). - /// Returns the new row id. - /// - /// # Errors - /// - /// Propagates the sqlite insert failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn store(&self, message: &str, file_path: Option<&str>, due_at: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let now = now_unix(); - conn.execute( - "INSERT INTO reminders (message, file_path, due_at, created_at, sent_at) \ - VALUES (?1, ?2, ?3, ?4, NULL)", - params![message, file_path, due_at, now], - )?; - Ok(conn.last_insert_rowid()) - } - - /// Count of currently-pending (undelivered) reminders. - /// - /// # Errors - /// - /// Propagates the sqlite query failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn count_pending(&self) -> Result { - let conn = self.conn.lock().unwrap(); - let count: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE sent_at IS NULL", - [], - |row| row.get(0), - )?; - Ok(u64::try_from(count).unwrap_or(0)) - } - - /// List pending reminders, soonest-due first — for `get_loose_ends` - /// rendering. - /// - /// # Errors - /// - /// Propagates the sqlite prepare / query failures. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn list_pending(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT id, message, file_path, due_at, created_at \ - FROM reminders WHERE sent_at IS NULL ORDER BY due_at ASC", - )?; - let rows = stmt - .query_map([], row_to_reminder)? - .collect::>>()?; - Ok(rows) - } - - /// Pending reminders due at or before `now`, oldest-due first, capped - /// at `limit` rows — the batch the delivery timer drains per tick. - /// - /// # Errors - /// - /// Propagates the sqlite prepare / query failures. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn due(&self, now: i64, limit: u64) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT id, message, file_path, due_at, created_at \ - FROM reminders WHERE sent_at IS NULL AND due_at <= ?1 \ - ORDER BY due_at ASC LIMIT ?2", - )?; - let limit = i64::try_from(limit).unwrap_or(i64::MAX); - let rows = stmt - .query_map(params![now, limit], row_to_reminder)? - .collect::>>()?; - Ok(rows) - } - - /// Soft-delete: stamp `sent_at` so the row drops out of - /// `count_pending`/`list_pending`/`due` but survives for the rollup - /// stats until `prune_delivered_older_than` reaps it. - /// - /// # Errors - /// - /// Propagates the sqlite update failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn mark_delivered(&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], - )?; - Ok(()) - } - - /// Cancel a still-pending reminder by id (hard delete — it never - /// fired, so there's nothing to keep for the rollup). Returns the - /// number of rows deleted (0 = unknown id, or already delivered). - /// - /// # Errors - /// - /// Propagates the sqlite delete failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn cancel(&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) - } - - /// Scheduled/delivered/pending counts within the trailing - /// `since_secs` window (`0` = all time) — the local equivalent of the - /// old c0re `ReminderRollup` query, same `sent_at IS NULL/NOT NULL` - /// shape. - /// - /// # Errors - /// - /// Propagates the sqlite query failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn rollup(&self, since_secs: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let cutoff = if since_secs > 0 { - now_unix().saturating_sub(since_secs) - } else { - i64::MIN - }; - let scheduled: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE created_at >= ?1", - params![cutoff], - |row| row.get(0), - )?; - let delivered: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE created_at >= ?1 AND sent_at IS NOT NULL", - params![cutoff], - |row| row.get(0), - )?; - let pending: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE created_at >= ?1 AND sent_at IS NULL", - params![cutoff], - |row| row.get(0), - )?; - Ok(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), - }) - } - - /// Reap delivered rows older than `cutoff` (unix seconds) — called - /// from `vacuum.rs`'s periodic sweep so the table doesn't grow - /// unbounded. Returns the number of rows deleted. - /// - /// # Errors - /// - /// Propagates the sqlite delete failure. - /// - /// # Panics - /// - /// Panics if the connection mutex is poisoned. - pub fn prune_delivered_older_than(&self, cutoff: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "DELETE FROM reminders WHERE sent_at IS NOT NULL AND sent_at < ?1", - params![cutoff], - )?; - Ok(n) - } -} - -fn row_to_reminder(row: &rusqlite::Row) -> rusqlite::Result { - Ok(Reminder { - id: row.get(0)?, - message: row.get(1)?, - file_path: row.get(2)?, - due_at: row.get(3)?, - created_at: row.get(4)?, - }) -} - -#[cfg(test)] -mod tests { - use super::*; - - // Return the `TempDir` alongside the store so it outlives the test — - // dropping it early deletes the dir and SQLite fails with - // `SQLITE_READONLY_DBMOVED`. - fn store() -> (tempfile::TempDir, Reminders) { - let dir = tempfile::tempdir().unwrap(); - let db = Reminders::open(&dir.path().join("reminders.sqlite")).unwrap(); - (dir, db) - } - - #[test] - fn store_and_list_pending() { - let (_dir, s) = store(); - let id = s.store("check on x", None, 1000).unwrap(); - let pending = s.list_pending().unwrap(); - assert_eq!(pending.len(), 1); - assert_eq!(pending[0].id, id); - assert_eq!(pending[0].due_at, 1000); - assert_eq!(s.count_pending().unwrap(), 1); - } - - #[test] - fn due_filters_by_time_and_limit() { - let (_dir, s) = store(); - s.store("a", None, 100).unwrap(); - s.store("b", None, 200).unwrap(); - s.store("c", None, 300).unwrap(); - let due = s.due(200, 10).unwrap(); - assert_eq!(due.len(), 2, "only a and b are due by t=200"); - let capped = s.due(300, 1).unwrap(); - assert_eq!(capped.len(), 1, "limit caps the batch"); - } - - #[test] - fn mark_delivered_removes_from_pending_but_counts_in_rollup() { - let (_dir, s) = store(); - let id = s.store("a", None, 100).unwrap(); - s.mark_delivered(id).unwrap(); - assert_eq!(s.count_pending().unwrap(), 0); - assert!(s.list_pending().unwrap().is_empty()); - let rollup = s.rollup(0).unwrap(); - assert_eq!(rollup.scheduled, 1); - assert_eq!(rollup.delivered, 1); - assert_eq!(rollup.pending, 0); - } - - #[test] - fn cancel_only_removes_pending() { - let (_dir, s) = store(); - let pending_id = s.store("a", None, 100).unwrap(); - let delivered_id = s.store("b", None, 100).unwrap(); - s.mark_delivered(delivered_id).unwrap(); - assert_eq!(s.cancel(pending_id).unwrap(), 1); - assert_eq!( - s.cancel(delivered_id).unwrap(), - 0, - "already-delivered rows aren't cancellable" - ); - assert_eq!(s.cancel(999).unwrap(), 0, "unknown id is a no-op"); - } - - #[test] - fn prune_reaps_only_old_delivered_rows() { - let (_dir, s) = store(); - let old = s.store("old", None, 100).unwrap(); - let recent = s.store("recent", None, 100).unwrap(); - let still_pending = s.store("pending", None, 100).unwrap(); - s.mark_delivered(old).unwrap(); - s.mark_delivered(recent).unwrap(); - // Backdate `old`'s sent_at directly so the cutoff test is deterministic. - { - let conn = s.conn.lock().unwrap(); - conn.execute( - "UPDATE reminders SET sent_at = 1 WHERE id = ?1", - params![old], - ) - .unwrap(); - } - let cutoff = now_unix() - 10; - let n = s.prune_delivered_older_than(cutoff).unwrap(); - assert_eq!(n, 1, "only the backdated row is older than cutoff"); - let remaining_ids: Vec = { - let conn = s.conn.lock().unwrap(); - let mut stmt = conn - .prepare("SELECT id FROM reminders ORDER BY id") - .unwrap(); - stmt.query_map([], |row| row.get(0)) - .unwrap() - .collect::>>() - .unwrap() - }; - assert_eq!(remaining_ids, vec![recent, still_pending]); - } -} diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index ace90cb5..7ad2382c 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -1,14 +1,9 @@ -//! In-agent socket server (loose-ends v2 + harness-local reminders). Binds -//! the harness-owned `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` -//! protocol to the in-container producers (matrix / bash daemons, -//! forge-notify) and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/ -//! `cancel_loose_end` tool impls. Todo ops hit the harness-local [`Todos`] -//! store; a new-or-changed upsert fires an in-process [`Notify`] so the -//! serve loop drives a turn. Reminder ops hit the harness-local -//! [`Reminders`] store (`None` when the store failed to open — every -//! reminder op then returns `Response::Err`); a reminder *firing* is a -//! separate path (`reminder_timer`), not driven through this socket. No -//! hive-c0re round-trip, no broker long-poll, no marker files. +//! In-agent socket server (loose-ends v2). Binds the harness-owned +//! `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` protocol to the +//! in-container producers (matrix / bash daemons, forge-notify). Todo ops +//! hit the harness-local [`Todos`] store; a new-or-changed upsert fires an +//! in-process [`Notify`] so the serve loop drives a turn — no hive-c0re +//! round-trip, no broker long-poll, no marker files. //! //! One request/response line per connection, matching the producers' //! existing best-effort JSON-line clients (they just change which socket @@ -24,7 +19,6 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::sync::Notify; -use crate::reminders::{Reminder, Reminders}; use crate::todos::{Todo, Todos}; /// Resolve the in-agent socket path from `HIVE_AGENT_SOCKET`. `None` when @@ -35,32 +29,6 @@ fn socket_path() -> Option { .map(PathBuf::from) } -/// Dial this same in-agent socket from elsewhere IN THIS PROCESS — the -/// harness's own `web_ui` endpoints (`api_todos`, `/api/stats` reminder -/// rollup) and `Surface::post_turn_counts` all need read access to the -/// `Todos`/`Reminders` stores this module owns behind an `Arc` on a -/// different spawned task, and a loopback dial is simpler than threading -/// those `Arc`s through every caller. One-shot, best-effort (no retry, -/// unlike the broker client): a connect failure means the socket server -/// task itself isn't up, which a retry within one call wouldn't fix. -pub(crate) async fn dial(req: &Request) -> Option { - let path = socket_path()?; - if !path.exists() { - return None; - } - tokio::time::timeout(std::time::Duration::from_secs(3), async move { - let mut stream = UnixStream::connect(&path).await.ok()?; - let mut line = serde_json::to_string(req).ok()?; - line.push('\n'); - stream.write_all(line.as_bytes()).await.ok()?; - let mut lines = BufReader::new(stream).lines(); - let resp_line = lines.next_line().await.ok()??; - serde_json::from_str(&resp_line).ok() - }) - .await - .ok()? -} - /// Run the in-agent socket server: bind + accept loop, one request/response /// line per connection. A no-op (returns `Ok`) when `HIVE_AGENT_SOCKET` is /// unset, so a standalone harness without producers just skips it. @@ -68,11 +36,7 @@ pub(crate) async fn dial(req: &Request) -> Option { /// # Errors /// /// Returns an error if the socket path is set but can't be bound. -pub async fn run( - store: Arc, - wake: Arc, - reminders: Option>, -) -> Result<()> { +pub async fn run(store: Arc, wake: Arc) -> Result<()> { let Some(path) = socket_path() else { tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled"); return Ok(()); @@ -84,9 +48,8 @@ pub async fn run( Ok((stream, _)) => { let store = store.clone(); let wake = wake.clone(); - let reminders = reminders.clone(); tokio::spawn(async move { - if let Err(e) = handle_conn(stream, &store, &wake, reminders.as_deref()).await { + if let Err(e) = handle_conn(stream, &store, &wake).await { tracing::warn!(error = ?e, "in-agent todo connection failed"); } }); @@ -112,12 +75,7 @@ fn bind(path: &Path) -> Result { /// Handle one connection: read a single JSON request line, apply it to the /// store, write the JSON response line back. -async fn handle_conn( - stream: UnixStream, - store: &Todos, - wake: &Notify, - reminders: Option<&Reminders>, -) -> Result<()> { +async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); let mut line = String::new(); @@ -125,7 +83,7 @@ async fn handle_conn( return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(req, store, wake, reminders), + Ok(req) => dispatch(req, store, wake), Err(e) => Response::Err { message: format!("bad request: {e}"), }, @@ -138,9 +96,8 @@ async fn handle_conn( } /// Apply one request to the store, firing `wake` on a new/changed upsert so -/// the serve loop runs a turn. `reminders` is `None` when that store -/// failed to open at boot — every reminder op then returns an `Err`. -fn dispatch(req: Request, store: &Todos, wake: &Notify, reminders: Option<&Reminders>) -> Response { +/// the serve loop runs a turn. +fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response { match req { Request::UpsertTodo { subsystem, @@ -185,78 +142,6 @@ fn dispatch(req: Request, store: &Todos, wake: &Notify, reminders: Option<&Remin }, Err(e) => err(&e), }, - Request::StoreReminder { - message, - timing, - file_path, - } => match reminders { - Some(r) => { - match crate::reminder_timer::store(r, &message, &timing, file_path.as_deref()) { - Ok(_id) => Response::Ok, - Err(message) => Response::Err { message }, - } - } - None => no_reminders_store(), - }, - Request::ListReminders => match reminders { - Some(r) => match r.list_pending() { - Ok(rows) => Response::LooseEnds { - loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(), - }, - Err(e) => err(&e), - }, - None => no_reminders_store(), - }, - Request::CancelReminder { id } => match reminders { - Some(r) => match r.cancel(id) { - Ok(count) => Response::Acked { - count: u64::try_from(count).unwrap_or(0), - }, - Err(e) => err(&e), - }, - None => no_reminders_store(), - }, - Request::CountPendingReminders => match reminders { - Some(r) => match r.count_pending() { - Ok(count) => Response::PendingRemindersCount { count }, - Err(e) => err(&e), - }, - None => no_reminders_store(), - }, - Request::ReminderRollup { since_secs } => match reminders { - Some(r) => { - let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX); - match r.rollup(since_secs) { - Ok(stats) => Response::ReminderRollup { stats }, - Err(e) => err(&e), - } - } - None => no_reminders_store(), - }, - } -} - -/// Shared "reminders db unavailable" response for every reminder op when -/// the store failed to open at boot (see `main.rs`'s best-effort open). -fn no_reminders_store() -> Response { - Response::Err { - message: "reminders store unavailable on this agent".to_owned(), - } -} - -/// Map a stored [`Reminder`] to a [`LooseEnd::Reminder`], deriving -/// `age_seconds` from `created_at` (mirrors the old c0re rendering — -/// "age" is how long the reminder has been *scheduled*, not how soon -/// it's due). -fn reminder_to_loose_end(r: Reminder) -> LooseEnd { - let now = hive_sh4re::wire_time::now_unix(); - let age = u64::try_from(now.saturating_sub(r.created_at)).unwrap_or(0); - LooseEnd::Reminder { - id: r.id, - owner: crate::identity::label(), - message: r.message, - due_at: hive_sh4re::wire_time::from_secs(r.due_at), - age_seconds: age, } } diff --git a/hive-agent/src/vacuum.rs b/hive-agent/src/vacuum.rs index 06275f9a..c33ec8ad 100644 --- a/hive-agent/src/vacuum.rs +++ b/hive-agent/src/vacuum.rs @@ -28,10 +28,6 @@ const BASH_KEEP_SECS: i64 = 48 * 3600; /// kinds are never deleted by this sweep — they carry the semantic per-turn /// history the operator scrolls back through. const STREAM_KEEP_SECS: i64 = 14 * 24 * 3600; -/// Keep delivered (soft-deleted, `sent_at` set) reminder rows this long -/// before reaping them — same window as `STREAM_KEEP_SECS`, kept around -/// only to serve the trailing-window `ReminderRollup` stats. -const REMINDER_KEEP_SECS: i64 = 14 * 24 * 3600; /// Terminal bash-task statuses whose files are eligible for deletion. const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"]; @@ -64,24 +60,6 @@ fn sweep_once() { Err(e) => tracing::warn!(error = ?e, "events vacuum failed"), } } - - let reminders_db = crate::paths::reminders_db(); - if reminders_db.exists() { - match vacuum_reminders(&reminders_db) { - Ok(0) => {} - Ok(n) => tracing::info!(removed = n, "reminders vacuum"), - Err(e) => tracing::warn!(error = ?e, "reminders vacuum failed"), - } - } -} - -/// Reap delivered reminder rows older than [`REMINDER_KEEP_SECS`] via the -/// typed store API (own short-lived connection — mirrors `vacuum_events`'s -/// own connection to `events.sqlite` rather than sharing the harness's live -/// `Reminders` handle). -fn vacuum_reminders(path: &Path) -> anyhow::Result { - let store = crate::reminders::Reminders::open(path)?; - store.prune_delivered_older_than(now_unix() - REMINDER_KEEP_SECS) } /// Delete eligible bash-task trios in `dir`. Returns the count of `.json` diff --git a/hive-agent/src/web_ui/stats.rs b/hive-agent/src/web_ui/stats.rs index 3cbc3ae3..8bf59747 100644 --- a/hive-agent/src/web_ui/stats.rs +++ b/hive-agent/src/web_ui/stats.rs @@ -12,30 +12,36 @@ pub(super) struct StatsQuery { } pub(super) async fn api_stats( - State(_state): State, + State(state): State, axum::extract::Query(q): axum::extract::Query, ) -> axum::Json { let window = crate::stats::Window::parse(q.window.as_deref().unwrap_or("24h")); let mut snapshot = crate::stats::snapshot_default(window); - // Pass the window span so the local reminder rollup filters its counts - // to the same time range as the chart data. + // Pass the window span to the reminder-stats RPC so the broker + // filters its counts to the same time range as the chart data. let window_secs = window.span_secs(); let window_secs_u = u64::try_from(window_secs).unwrap_or(0); - snapshot.reminder_stats = fetch_reminder_stats(window_secs_u).await; + snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await; axum::Json(snapshot) } -/// Fetch reminder activity stats from the harness-local reminder store over -/// `HIVE_AGENT_SOCKET` — was a broker RPC before reminders moved -/// in-container. Returns `None` on any transport / decode failure or -/// when the socket is unset — the stats are decorative, not authoritative. -async fn fetch_reminder_stats(window_secs: u64) -> Option { - match crate::todo_server::dial(&hive_agent_sock::Request::ReminderRollup { - since_secs: window_secs, - }) - .await? +/// Fetch reminder activity stats from the broker via the per-agent / manager +/// socket. Returns None on any transport / decode failure — the stats are +/// decorative, not authoritative. +async fn fetch_reminder_stats( + socket: &std::path::Path, + window_secs: u64, +) -> Option { + match super::broker_request( + socket, + &hive_core_agent_sock::Request::ReminderRollup { + since_secs: window_secs, + agent: None, + }, + ) + .await { - hive_agent_sock::Response::ReminderRollup { stats } => Some(stats), + Ok(hive_core_agent_sock::Response::ReminderRollup(stats)) => Some(stats), _ => None, } } @@ -47,12 +53,36 @@ async fn fetch_reminder_stats(window_secs: u64) -> Option Response { - let todos = - match crate::todo_server::dial(&hive_agent_sock::Request::ListTodos { subsystem: None }) - .await - { - Some(hive_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends, + use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; + use tokio::net::UnixStream; + + let socket_path = match std::env::var_os("HIVE_AGENT_SOCKET") { + Some(p) => std::path::PathBuf::from(p), + None => return axum::Json(serde_json::json!({ "todos": [] })).into_response(), + }; + if !socket_path.exists() { + return axum::Json(serde_json::json!({ "todos": [] })).into_response(); + } + let todos = tokio::time::timeout(std::time::Duration::from_secs(3), async move { + let mut stream = UnixStream::connect(&socket_path).await?; + let req = hive_agent_sock::Request::ListTodos { subsystem: None }; + let mut line = serde_json::to_string(&req)?; + line.push('\n'); + stream.write_all(line.as_bytes()).await?; + stream.flush().await?; + let mut lines = BufReader::new(stream).lines(); + let resp_line = lines + .next_line() + .await? + .ok_or_else(|| anyhow::anyhow!("agent socket closed without response"))?; + let resp: hive_agent_sock::Response = serde_json::from_str(&resp_line)?; + anyhow::Ok(match resp { + hive_agent_sock::Response::LooseEnds { loose_ends } => loose_ends, _ => Vec::new(), - }; + }) + }) + .await + .unwrap_or_else(|_| Err(anyhow::anyhow!("timeout"))) + .unwrap_or_default(); axum::Json(serde_json::json!({ "todos": todos })).into_response() } diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs index b76d30de..abefef7b 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -29,6 +29,13 @@ 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 @@ -48,7 +55,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() -> Vec { +pub async fn build_all(coord: &Coordinator) -> 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. @@ -72,6 +79,10 @@ pub async fn build_all() -> 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 @@ -97,6 +108,7 @@ pub async fn build_all() -> 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 c24247f9..6c07db17 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -612,6 +612,24 @@ 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. @@ -912,7 +930,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().await; + let fresh = container_view::build_all(self).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 580feeba..00a65b4c 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -32,6 +32,7 @@ mod meta_inputs; mod misc_api; pub(crate) mod permissions; mod questions; +mod reminders; mod schedules; mod state_files; mod state_snapshot; @@ -92,6 +93,7 @@ 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( @@ -210,6 +212,14 @@ 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 new file mode 100644 index 00000000..d61550b8 --- /dev/null +++ b/hive-c0re/src/dashboard/reminders.rs @@ -0,0 +1,61 @@ +//! 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 b5451b05..d3fb9c92 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -219,6 +219,14 @@ 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 @@ -286,6 +294,7 @@ 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", @@ -411,6 +420,10 @@ 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 c206b19f..d510d222 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -3,8 +3,7 @@ //! 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. Reminders are agent-local (in-container store) and no -//! longer sourced from here. +//! one place. //! //! Call frequency is low (an agent doing self-introspection between //! turns), so the sweep happens fresh every time — no caching, no @@ -28,17 +27,18 @@ 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). +/// someone) OR the target (owes a reply); +/// - pending reminders this agent scheduled (`owner == self`). /// -/// 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). +/// 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). /// /// # Errors /// /// Propagates errors from `count_pending` and the pending-approval / -/// question sqlite queries. +/// question / reminder sqlite queries. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -85,13 +85,25 @@ 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. Manager surface only; sub-agents can't see -/// each other's threads via the agent surface (`for_agent` filters by -/// name). +/// 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). pub fn hive_wide(coord: &Coordinator) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -113,6 +125,15 @@ 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 f8e2870d..a97aff7b 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -45,7 +45,8 @@ 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, scheduled_prompts_worker, + agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler, + scheduled_prompts_worker, }; use coordinator::{Coordinator, HiveEnv, ServeConfig}; @@ -213,7 +214,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, scheduled-prompts, +/// background tasks (auto-update, vacuums, crash-watcher, reminder-scheduler, /// dashboard), then serve the admin socket until a signal arrives. #[allow( clippy::too_many_lines, @@ -521,6 +522,9 @@ 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 391d6451..d531ee0c 100644 --- a/hive-c0re/src/questions.rs +++ b/hive-c0re/src/questions.rs @@ -186,15 +186,14 @@ pub fn handle_cancel_loose_end( Ok(()) } hive_sh4re::CancelLooseEndKind::Reminder => { - // 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" - )) + // 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(()) } 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 4bfb3bf4..21089867 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().await, + HostRequest::AgentStatus => handle_agent_status(&coord).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() +async fn handle_agent_status(coord: &Arc) -> HostResponse { + let rows = crate::container_view::build_all(coord) .await .into_iter() .map(|v| hive_sh4re::AgentStatusRow { @@ -308,13 +308,7 @@ async fn handle_agent_status() -> HostResponse { needs_update: v.needs_update, needs_login: v.needs_login, deployed_sha: v.deployed_sha, - // 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, + pending_reminders: v.pending_reminders, parent: v.parent, }) .collect(); diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index e9cfc3d0..ffe3c81f 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -23,6 +23,7 @@ use crate::coordinator::Coordinator; mod config_approvals; mod lifecycle_handlers; +mod reminders; mod schedules; pub(crate) use config_approvals::submit_merge_config_pr; @@ -33,6 +34,7 @@ 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, @@ -182,8 +184,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`, where the manager can target other -/// agents) or for manager-only variants. +/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup` +/// 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. @@ -232,6 +234,11 @@ 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 @@ -583,6 +590,13 @@ 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, @@ -777,38 +791,41 @@ fn handle_get_loose_ends(coord: &Arc, agent: &str, target: Option<& } } -/// 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" - )) - } - } +/// `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 }, } } diff --git a/hive-c0re/src/socket_server/reminders.rs b/hive-c0re/src/socket_server/reminders.rs new file mode 100644 index 00000000..ae879da1 --- /dev/null +++ b/hive-c0re/src/socket_server/reminders.rs @@ -0,0 +1,229 @@ +//! 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 1ccf80bd..0e8bcaf5 100644 --- a/hive-c0re/src/stats/host_stats.rs +++ b/hive-c0re/src/stats/host_stats.rs @@ -250,6 +250,7 @@ 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 729bbeed..1b8cca58 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -7,6 +7,7 @@ use std::sync::Mutex; use anyhow::{Context, Result}; +use chrono::{DateTime, Utc}; use hive_sh4re::wire_time::now_unix; use hive_sh4re::{InboxRow, Message}; @@ -29,6 +30,18 @@ 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 @@ -39,6 +52,12 @@ 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 @@ -51,6 +70,37 @@ 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 @@ -143,6 +193,18 @@ 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 { @@ -788,6 +850,309 @@ 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 ee5fa0ad..75619f0b 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 -//! scheduled-prompt delivery loop, boot-time auto-update +//! reminder and scheduled-prompt delivery loops, 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,4 +9,5 @@ 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 new file mode 100644 index 00000000..f05f8d92 --- /dev/null +++ b/hive-c0re/src/workers/reminder_scheduler.rs @@ -0,0 +1,278 @@ +//! 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 cb390201..90e18c4d 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, SchedulePromptPayload, WireSchedule, + MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule, }; use hive_types::Ident; use serde::{Deserialize, Serialize}; @@ -81,6 +81,18 @@ 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; @@ -91,6 +103,33 @@ 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. @@ -269,6 +308,10 @@ 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 {