diff --git a/CLAUDE.md b/CLAUDE.md index 22b8c5f7..8a523180 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -23,9 +23,6 @@ hive-c0re/ host daemon + CLI (one binary, subcommand-dispatched) hourly vacuum of delivered>30d src/approvals.rs sqlite Approval queue + kinds src/operator_questions.rs sqlite question queue backing `ask_operator` - src/reminder_scheduler.rs 5s poll loop: drains due reminders, - resolves file_path container→host, persists - payload + delivers pointer string src/events_vacuum.rs host-side hourly sweep of every agent's /state/hyperhive-events.sqlite src/crash_watch.rs poll every 10s; fire HelperEvent::ContainerCrash diff --git a/TODO.md b/TODO.md index a83bd801..f7a3c6c9 100644 --- a/TODO.md +++ b/TODO.md @@ -13,9 +13,8 @@ - ~~Handle text overflow → suggest file_path option for long messages~~ ✓ fixed — Remind dispatch rejects `message.len() > 4096` (when no `file_path` was supplied) with an error pointing at the `file_path` escape hatch. - Per-agent reminder limits (burst capacity, rate limiting) -- ~~**Expose `remind` MCP tool**~~ ✓ fixed — `mcp__hyperhive__remind` now on `AgentServer`; takes `message`, exactly one of `delay_seconds` / `at_unix_timestamp`, optional `file_path`. Manager surface still missing (no `ManagerRequest::Remind` variant) — separate item below. -- **Manager-side `remind`**: mirror of the agent tool but on `ManagerServer`. Needs `ManagerRequest::Remind` variant in hive-sh4re, dispatch in manager_server.rs, MCP tool wiring. -- ~~**File path delivery**~~ ✓ fixed — scheduler now writes the reminder body to the requested `file_path` (mapped from container `/agents//state/...` to host `/var/lib/hyperhive/agents//state/...`) and delivers a short pointer message in its place. Path-traversal + foreign-agent-state writes are rejected; on rejection or write failure the body falls back to inline delivery with a noted warning. New module `hive-c0re/src/reminder_scheduler.rs` (extracted from main.rs). +- **Expose `remind` MCP tool**: wire protocol exists (`AgentRequest::Remind`) and the broker handles it, but no `#[tool]` method on `AgentServer` actually surfaces it to claude. Until that lands, the Remind path is unreachable from agent turns. +- **File path delivery**: currently unused in scheduler delivery loop — implement file write/delivery to /state//reminders/ or similar (also needed for the overflow-check escape hatch above to actually do anything useful). - ~~**Orphan reminders**~~ ✓ fixed — `Broker::deliver_reminder` wraps the inbox INSERT + reminders UPDATE in one sqlite transaction; partial failure can no longer cause duplicate delivery on the next tick. - ~~**Unbounded batches**~~ ✓ fixed — scheduler now calls `get_due_reminders(REMINDER_BATCH_LIMIT)` (cap = 100/tick); overflow stays due and gets picked up next cycle. - **Scheduler shutdown**: add graceful shutdown signal when coordinator is destroyed (currently runs forever) diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 384a126c..542abfba 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -144,34 +144,6 @@ pub struct RecvArgs { pub wait_seconds: Option, } -/// MCP tool args for `remind`. Exactly one of `delay_seconds` or -/// `at_unix_timestamp` must be set; both / neither is a tool-side error. -/// Hides the tagged `ReminderTiming` enum behind a flatter schema so the -/// model picks one field instead of building `{"timing_type": "in_seconds", -/// "seconds": 60}` shaped objects. -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct RemindArgs { - /// Body that lands in your inbox when the reminder fires (sender - /// will appear as `reminder`). Capped at 4096 bytes when - /// `file_path` is unset — anything bigger should be persisted to - /// disk and pointed at via `file_path`. - pub message: String, - /// Fire `delay_seconds` from now (relative). Set this OR - /// `at_unix_timestamp`, not both. - #[serde(default)] - pub delay_seconds: Option, - /// Fire at this absolute unix timestamp (seconds since epoch). Set - /// this OR `delay_seconds`, not both. - #[serde(default)] - pub at_unix_timestamp: Option, - /// Optional path to a file the scheduler should reference instead of - /// inlining a long `message`. Use this for large payloads (research - /// notes, file lists, intermediate state). Path must be reachable from - /// the agent's container — typically under `/agents//state/`. - #[serde(default)] - pub file_path: Option, -} - /// Per-agent tool surface. Holds the socket path so each tool call doesn't /// re-derive it; the socket itself is the per-container `/run/hive/mcp.sock`. #[derive(Debug, Clone)] @@ -282,51 +254,12 @@ impl AgentServer { }) .await } - - #[tool( - description = "Schedule a reminder that lands in this agent's own inbox at a future \ - time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \ - status in 60s', 'retry failed deploy at 14:00 UTC', 'nudge me when the operator's \ - deploy window opens'. Set EXACTLY ONE of `delay_seconds` (fire N seconds from now) \ - or `at_unix_timestamp` (fire at absolute epoch second). Body is capped at 4096 bytes \ - when `file_path` is unset; for larger payloads write them to a file under your \ - `/agents//state/` dir and pass the path in `file_path`. Returns immediately — \ - the reminder lives in the broker until due." - )] - async fn remind(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("remind", log, async move { - let timing = match (args.delay_seconds, args.at_unix_timestamp) { - (Some(_), Some(_)) => { - return "remind failed: pass exactly one of `delay_seconds` or \ - `at_unix_timestamp`, not both" - .to_string(); - } - (None, None) => { - return "remind failed: pass exactly one of `delay_seconds` or \ - `at_unix_timestamp`" - .to_string(); - } - (Some(s), None) => hive_sh4re::ReminderTiming::InSeconds { seconds: s }, - (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, - }; - let (resp, retries) = self - .dispatch(hive_sh4re::AgentRequest::Remind { - message: args.message, - timing, - file_path: args.file_path, - }) - .await; - annotate_retries(format_ack(resp, "remind", "reminder scheduled".to_string()), retries) - }) - .await - } } #[tool_handler( instructions = "You are a hyperhive agent. Use `send` to talk to peers (by their logical \ name) or to the operator (recipient `operator`). Use `recv` to drain your inbox one \ - message at a time. Use `remind` to schedule a future wake-up message for yourself." + message at a time." )] impl ServerHandler for AgentServer {} diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 8918e68b..ea84f8c2 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -21,7 +21,6 @@ mod manager_server; mod meta; mod migrate; mod operator_questions; -mod reminder_scheduler; mod server; use coordinator::Coordinator; @@ -87,6 +86,12 @@ enum Cmd { Deny { id: i64 }, } +/// Per-tick cap on reminders the scheduler delivers. Anything over this +/// stays due in the table and gets picked up on the next 5s tick — keeps +/// a 10k-deep backlog from flooding the broker (or hogging its mutex) in +/// one shot. +const REMINDER_BATCH_LIMIT: u64 = 100; + #[tokio::main] async fn main() -> Result<()> { tracing_subscriber::fmt() @@ -166,9 +171,41 @@ async fn main() -> Result<()> { // when a previously-running container goes away without an // operator-initiated transient state. crash_watch::spawn(coord.clone()); - // Reminder scheduler: drains due reminders + handles - // file_path payload persistence. See reminder_scheduler.rs. - reminder_scheduler::spawn(coord.clone()); + // Reminder scheduler: checks for due reminders every 5 seconds, + // delivers them atomically (insert inbox + mark sent in one + // sqlite transaction so a transient failure on the second step + // can never produce a duplicate next tick). Per-cycle batch + // limit caps the burst — leftover reminders stay due and get + // picked up on the next tick instead of monopolising the broker + // mutex. + let reminder_coord = coord.clone(); + tokio::spawn(async move { + loop { + match reminder_coord + .broker + .get_due_reminders(REMINDER_BATCH_LIMIT) + { + Ok(reminders) => { + for (agent, id, message, _file_path) in reminders { + if let Err(e) = + reminder_coord.broker.deliver_reminder(id, &agent, &message) + { + tracing::warn!( + reminder_id = id, + %agent, + error = ?e, + "failed to deliver reminder" + ); + } + } + } + Err(e) => { + tracing::warn!(error = ?e, "failed to query due reminders"); + } + } + tokio::time::sleep(std::time::Duration::from_secs(5)).await; + } + }); let dash_coord = coord.clone(); tokio::spawn(async move { if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await { diff --git a/hive-c0re/src/reminder_scheduler.rs b/hive-c0re/src/reminder_scheduler.rs deleted file mode 100644 index c21b752d..00000000 --- a/hive-c0re/src/reminder_scheduler.rs +++ /dev/null @@ -1,173 +0,0 @@ -//! Background loop that drains due reminders out of the broker and -//! delivers them as inbox messages. Mirrors the `events_vacuum` / -//! `crash_watch` shape — a single `spawn(coord)` entry point started -//! from `main.rs`. -//! -//! File-path semantics: a reminder may carry a `file_path` (the -//! agent-visible path inside its container). On delivery we: -//! -//! - Translate the container path (`/agents//state/foo.md`) to -//! the host path (`/var/lib/hyperhive/agents//state/foo.md`) -//! so hive-c0re can write to it from outside the container. -//! - Reject anything that isn't under the agent's own state subtree, -//! or that contains `..` (path traversal). Falling outside the -//! allowed prefix means the file write is skipped and the original -//! message is delivered inline (with a noted warning) — the -//! reminder still fires, just without the payload split. -//! - Write the reminder body to disk and deliver a short pointer -//! message in its place, so the agent's inbox/wake-prompt stays -//! small and the bulky payload can be read out of band. -//! -//! Atomicity of the inbox INSERT + `reminders.sent_at` UPDATE is handled -//! inside `Broker::deliver_reminder`; this module only computes the -//! body string before calling it. - -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. -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) { - tokio::spawn(async move { - loop { - tick(&coord); - tokio::time::sleep(POLL_INTERVAL).await; - } - }); -} - -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; - } - }; - for (agent, id, message, file_path) in due { - let body = prepare_body(&agent, &message, file_path.as_deref()); - if let Err(e) = coord.broker.deliver_reminder(id, &agent, &body) { - tracing::warn!( - reminder_id = id, - %agent, - error = ?e, - "failed to deliver reminder" - ); - } - } -} - -/// 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, 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 format!( - "[reminder file_path '{req_path}' rejected: {reason}; delivering body inline]\n\n{message}" - ); - } - }; - if let Some(parent) = host_path.parent() - && let Err(e) = std::fs::create_dir_all(parent) - { - tracing::warn!(%agent, path = %host_path.display(), error = ?e, "reminder file_path parent mkdir failed; delivering inline"); - return format!( - "[reminder file_path '{req_path}' parent dir create failed: {e}; delivering body inline]\n\n{message}" - ); - } - if let Err(e) = std::fs::write(&host_path, message) { - tracing::warn!(%agent, path = %host_path.display(), error = ?e, "reminder file_path write failed; delivering inline"); - return format!( - "[reminder file_path '{req_path}' write failed: {e}; delivering body inline]\n\n{message}" - ); - } - let bytes = message.len(); - tracing::info!(%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") -} - -/// Map an agent-visible container path to the matching host path, -/// validating that it lives under the agent's own state subtree and -/// doesn't try to traverse out via `..`. Returns the host `PathBuf` on -/// success, or a human-readable reason string on rejection. -fn resolve_host_path(agent: &str, req_path: &str) -> Result { - let prefix = format!("/agents/{agent}/state/"); - let Some(rel) = req_path.strip_prefix(&prefix) else { - return Err(format!( - "must be absolute and under `{prefix}` (got `{req_path}`)" - )); - }; - 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 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 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")); - } -}