delete c0re-side reminder plumbing (#2635 inc 1 commit 6)

hive-agent-mcp/src/mcp/mod.rs

@@ -35,8 +35,9 @@ pub use args::{
 pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};
 
 use render::{
-    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,
+    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,
 };
 
 /// Write (or remove) the status file in the agent's own `state/` directory.
@@ -343,6 +344,12 @@ 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
@@ -446,6 +453,26 @@ 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;
@@ -516,17 +543,22 @@ impl AgentServer {
                 (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_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,
-            )
+            // 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(),
+            }
         })
         .await
     }

hive-agent-mcp/src/mcp/render.rs

@@ -304,8 +304,10 @@ pub(super) async fn matrix_unread_summary() -> Option<Vec<MatrixRoomUnread>> {
 /// 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`] and
-/// [`mark_local_todo_done`].
+/// 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<LooseEnd>` to merge.
 pub(super) async fn dial_agent_socket(
     req: &hive_agent_sock::Request,
 ) -> Option<hive_agent_sock::Response> {
@@ -335,6 +337,16 @@ pub(super) async fn local_todos() -> Option<Vec<hive_sh4re::LooseEnd>> {
     }
 }
 
+/// 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<Vec<hive_sh4re::LooseEnd>> {
+    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

hive-agent-sock/src/lib.rs

@@ -1,14 +1,15 @@
 //! 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`. 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).
+//! `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).
 //!
 //! 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 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.
+//! 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.
 
 use serde::{Deserialize, Serialize};
 
@@ -58,6 +59,30 @@ 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<String>,
+    },
+    /// 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,
@@ -69,8 +94,13 @@ pub enum Response {
     Ok,
     /// Op succeeded and touched `count` rows (clear / mark-done).
     Acked { count: u64 },
-    /// `ListTodos` result.
+    /// `ListTodos` / `ListReminders` result (the latter wraps each row as
+    /// [`LooseEnd::Reminder`]).
     LooseEnds { loose_ends: Vec<LooseEnd> },
+    /// `CountPendingReminders` result.
+    PendingRemindersCount { count: u64 },
+    /// `ReminderRollup` result.
+    ReminderRollup { stats: hive_sh4re::ReminderStats },
     /// Op failed; `message` is operator-facing.
     Err { message: String },
 }

hive-agent/src/main.rs

@@ -20,6 +20,8 @@ mod mcp_config;
 mod paths;
 mod plugins;
 mod prompt;
+mod reminder_timer;
+mod reminders;
 mod serve_common;
 mod stats;
 mod stream_enrich;
@@ -214,6 +216,11 @@ 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.
@@ -325,15 +332,13 @@ impl Surface for AgentSurface {
                 Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(),
                 _ => None,
             };
-        let reminders = match client::request::<_, Response>(
-            socket,
-            &Request::CountPendingReminders { agent: None },
-        )
-        .await
-        {
-            Ok(Response::PendingRemindersCount { count }) => Some(count),
-            _ => 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,
+            };
         (threads, reminders)
     }
 
@@ -453,18 +458,41 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
             tracing::error!(error = %e, "web_ui::serve exited with error");
         }
     });
-    // 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.
+    // 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).
     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).await {
+                if let Err(e) = todo_server::run(store, wake, reminders).await {
                     tracing::error!(error = %e, "in-agent todo socket exited with error");
                 }
             });
@@ -490,6 +518,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         stats,
         &files,
         todo_wake,
+        reminder_rx,
     )
     .await
 }
@@ -512,6 +541,7 @@ async fn serve_loop<S: Surface>(
     stats: Option<TurnStats>,
     files: &turn::TurnFiles,
     todo_wake: Arc<tokio::sync::Notify>,
+    mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::DeliveredMessage>,
 ) -> Result<()> {
     tracing::info!(socket = %socket.display(), "harness serve");
     S::requeue_inflight(socket).await;
@@ -536,6 +566,7 @@ async fn serve_loop<S: Surface>(
                     biased;
                     o = S::recv_next(socket) => o,
                     () = todo_wake.notified() => RecvOutcome::LocalTodo,
+                    Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
                 }
             } {
                 RecvOutcome::Message(first) => first,

hive-agent/src/paths.rs

@@ -48,6 +48,14 @@ 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

hive-agent/src/reminder_timer.rs

@@ -0,0 +1,303 @@
+//! 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::<u64>().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<Arc<Reminders>>,
+    tx: mpsc::UnboundedSender<hive_sh4re::DeliveredMessage>,
+) {
+    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<hive_sh4re::DeliveredMessage>) {
+    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<i64, String> {
+    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>), 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<PathBuf, String> {
+    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<i64> {
+    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"));
+    }
+}

hive-agent/src/reminders.rs

@@ -0,0 +1,366 @@
+//! 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<String>,
+    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<Connection>,
+}
+
+impl Reminders {
+    /// Open (creating if needed) the reminder store at `path`.
+    ///
+    /// # Errors
+    ///
+    /// Propagates sqlite open / schema-apply failures.
+    pub fn open(path: &Path) -> Result<Self> {
+        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<i64> {
+        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<u64> {
+        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<Vec<Reminder>> {
+        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::<rusqlite::Result<Vec<_>>>()?;
+        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<Vec<Reminder>> {
+        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::<rusqlite::Result<Vec<_>>>()?;
+        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<usize> {
+        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<ReminderStats> {
+        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<usize> {
+        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<Reminder> {
+    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<i64> = {
+            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::<rusqlite::Result<Vec<_>>>()
+                .unwrap()
+        };
+        assert_eq!(remaining_ids, vec![recent, still_pending]);
+    }
+}

hive-agent/src/todo_server.rs

@@ -1,9 +1,14 @@
-//! 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.
+//! 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.
 //!
 //! One request/response line per connection, matching the producers'
 //! existing best-effort JSON-line clients (they just change which socket
@@ -19,6 +24,7 @@ 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
@@ -29,6 +35,32 @@ fn socket_path() -> Option<PathBuf> {
         .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<Response> {
+    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.
@@ -36,7 +68,11 @@ fn socket_path() -> Option<PathBuf> {
 /// # Errors
 ///
 /// Returns an error if the socket path is set but can't be bound.
-pub async fn run(store: Arc<Todos>, wake: Arc<Notify>) -> Result<()> {
+pub async fn run(
+    store: Arc<Todos>,
+    wake: Arc<Notify>,
+    reminders: Option<Arc<Reminders>>,
+) -> Result<()> {
     let Some(path) = socket_path() else {
         tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled");
         return Ok(());
@@ -48,8 +84,9 @@ pub async fn run(store: Arc<Todos>, wake: Arc<Notify>) -> Result<()> {
             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).await {
+                    if let Err(e) = handle_conn(stream, &store, &wake, reminders.as_deref()).await {
                         tracing::warn!(error = ?e, "in-agent todo connection failed");
                     }
                 });
@@ -75,7 +112,12 @@ fn bind(path: &Path) -> Result<UnixListener> {
 
 /// 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) -> Result<()> {
+async fn handle_conn(
+    stream: UnixStream,
+    store: &Todos,
+    wake: &Notify,
+    reminders: Option<&Reminders>,
+) -> Result<()> {
     let (read, mut write) = stream.into_split();
     let mut reader = BufReader::new(read);
     let mut line = String::new();
@@ -83,7 +125,7 @@ async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result
         return Ok(());
     }
     let resp = match serde_json::from_str::<Request>(line.trim()) {
-        Ok(req) => dispatch(req, store, wake),
+        Ok(req) => dispatch(req, store, wake, reminders),
         Err(e) => Response::Err {
             message: format!("bad request: {e}"),
         },
@@ -96,8 +138,9 @@ async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result
 }
 
 /// Apply one request to the store, firing `wake` on a new/changed upsert so
-/// the serve loop runs a turn.
-fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response {
+/// 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 {
     match req {
         Request::UpsertTodo {
             subsystem,
@@ -142,6 +185,78 @@ fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response {
             },
             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,
     }
 }
 

hive-agent/src/vacuum.rs

@@ -28,6 +28,10 @@ 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"];
 
@@ -60,6 +64,24 @@ 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<usize> {
+    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`

hive-agent/src/web_ui/stats.rs

@@ -12,36 +12,30 @@ pub(super) struct StatsQuery {
 }
 
 pub(super) async fn api_stats(
-    State(state): State<AppState>,
+    State(_state): State<AppState>,
     axum::extract::Query(q): axum::extract::Query<StatsQuery>,
 ) -> axum::Json<crate::stats::Snapshot> {
     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 to the reminder-stats RPC so the broker
-    // filters its counts to the same time range as the chart data.
+    // Pass the window span so the local reminder rollup 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(&state.socket, window_secs_u).await;
+    snapshot.reminder_stats = fetch_reminder_stats(window_secs_u).await;
     axum::Json(snapshot)
 }
 
-/// 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<hive_sh4re::ReminderStats> {
-    match super::broker_request(
-        socket,
-        &hive_core_agent_sock::Request::ReminderRollup {
-            since_secs: window_secs,
-            agent: None,
-        },
-    )
-    .await
+/// 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<hive_sh4re::ReminderStats> {
+    match crate::todo_server::dial(&hive_agent_sock::Request::ReminderRollup {
+        since_secs: window_secs,
+    })
+    .await?
     {
-        Ok(hive_core_agent_sock::Response::ReminderRollup(stats)) => Some(stats),
+        hive_agent_sock::Response::ReminderRollup { stats } => Some(stats),
         _ => None,
     }
 }
@@ -53,36 +47,12 @@ async fn fetch_reminder_stats(
 /// `LooseEnd::Todo` (subsystem, key, summary, source, `age_seconds`). Returns
 /// an empty array when the socket is unavailable — best-effort, silent failure.
 pub(super) async fn api_todos() -> Response {
-    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,
+    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,
             _ => Vec::new(),
-        })
-    })
-    .await
-    .unwrap_or_else(|_| Err(anyhow::anyhow!("timeout")))
-    .unwrap_or_default();
+        };
     axum::Json(serde_json::json!({ "todos": todos })).into_response()
 }

hive-c0re/src/container_view.rs

@@ -29,13 +29,6 @@ pub struct ContainerView {
     /// for this agent's input.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub deployed_sha: Option<String>,
-    /// Count of this agent's pending reminders. Computed during
-    /// `build_all` via `Broker::count_pending_reminders_for`; the
-    /// dashboard renders a small chip when > 0. Updates with the
-    /// 10s `crash_watch` rescan + every container mutation site;
-    /// not real-time on remind/cancel-reminder but close enough.
-    #[serde(default)]
-    pub pending_reminders: u64,
     /// Name of this agent's parent in the agent hierarchy. `None`
     /// marks the agent as root-level; the dashboard renders it without
     /// indentation. Sourced from `meta/topology.json` (single source of
@@ -55,7 +48,7 @@ pub struct ContainerView {
 
 /// Build the full container list. Wraps `lifecycle::list()` and
 /// resolves every per-agent attribute the dashboard surfaces.
-pub async fn build_all(coord: &Coordinator) -> Vec<ContainerView> {
+pub async fn build_all() -> Vec<ContainerView> {
     let raw = lifecycle::list().await.unwrap_or_default();
     let locked = read_meta_locked_revs();
     // Pull the topology map once and look up each agent's parent below.
@@ -79,10 +72,6 @@ pub async fn build_all(coord: &Coordinator) -> Vec<ContainerView> {
         let needs_update =
             crate::auto_update::agent_config_pending(logical.as_str(), deployed_full).await;
         let deployed_sha = deployed_full.map(|s| s[..s.len().min(12)].to_owned());
-        let pending_reminders = coord
-            .broker
-            .count_pending_reminders_for(logical.as_str())
-            .unwrap_or(0);
         let parent = topology.get(logical.as_str()).cloned().flatten();
         let running = lifecycle::is_running(logical.as_str()).await;
         // needs_login fires when EITHER the claude session dir is missing
@@ -108,7 +97,6 @@ pub async fn build_all(coord: &Coordinator) -> Vec<ContainerView> {
             needs_update,
             needs_login,
             deployed_sha,
-            pending_reminders,
             parent,
             active_model,
         });

hive-c0re/src/coordinator.rs

@@ -612,24 +612,6 @@ impl Coordinator {
             .map(|m| m.keys().cloned().collect())
     }
 
-    /// Emit a `RemindersChanged` snapshot event. Called from every
-    /// reminder mutation site (agent `remind` calls, operator cancel /
-    /// retry, and the scheduler after each delivery batch) so the
-    /// dashboard's pending-reminders list stays live without polling.
-    pub fn emit_reminders_snapshot(self: &Arc<Self>) {
-        let reminders = match self.broker.list_pending_reminders() {
-            Ok(rows) => rows,
-            Err(e) => {
-                tracing::warn!(error = ?e, "emit_reminders_snapshot: list failed");
-                return;
-            }
-        };
-        self.emit_dashboard_event(DashboardEvent::RemindersChanged {
-            seq: self.next_seq(),
-            reminders,
-        });
-    }
-
     /// Emit a `CapabilitiesChanged` snapshot event. Called from the
     /// rebuild-queue worker after a `PermChange` / Capabilities entry
     /// commits the JSON file, so the P3RM1SS10NS tab updates live.
@@ -930,7 +912,7 @@ impl Coordinator {
     /// Cheap when nothing changed (one `nixos-container list` + a
     /// `HashMap` diff + zero emits).
     pub async fn rescan_containers_and_emit(self: &Arc<Self>) {
-        let fresh = container_view::build_all(self).await;
+        let fresh = container_view::build_all().await;
         let mut last = self.last_containers.lock().await;
         let mut changed_or_new = Vec::new();
         let mut removed = Vec::new();

hive-c0re/src/dashboard/mod.rs

@@ -32,7 +32,6 @@ mod meta_inputs;
 mod misc_api;
 pub(crate) mod permissions;
 mod questions;
-mod reminders;
 mod schedules;
 mod state_files;
 mod state_snapshot;
@@ -93,7 +92,6 @@ pub async fn serve(
             "/api/extra-forge-account",
             post(extra_forges::post_extra_forge_account),
         )
-        .route("/api/reminders", get(reminders::api_reminders))
         .route("/api/operator-inbox", get(misc_api::api_operator_inbox))
         .route("/api/stats-hive", get(misc_api::api_stats_hive))
         .route(
@@ -212,14 +210,6 @@ pub async fn serve(
             "/api/github-account",
             post(matrix_accounts::post_github_account).get(matrix_accounts::get_github_account),
         )
-        .route(
-            "/api/cancel-reminder/{id}",
-            post(reminders::post_cancel_reminder),
-        )
-        .route(
-            "/api/retry-reminder/{id}",
-            post(reminders::post_retry_reminder),
-        )
         .route("/api/request-spawn", post(misc_api::post_request_spawn))
         .route("/api/op-send", post(misc_api::post_op_send))
         .route("/api/meta-update", post(meta_inputs::post_meta_update))

hive-c0re/src/dashboard/reminders.rs

@@ -1,61 +0,0 @@
-//! Reminder endpoints for the dashboard.
-//!
-//! Lists pending reminders for the reminders tab, and lets the operator
-//! cancel a pending reminder or reset its failure state so the scheduler
-//! retries it on the next tick.
-
-use axum::{
-    extract::{Path as AxumPath, State},
-    http::StatusCode,
-    response::{IntoResponse, Response},
-};
-
-use problem_details::ProblemDetails;
-
-use super::{AppState, error_problem, error_response};
-
-pub(super) async fn api_reminders(State(state): State<AppState>) -> 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<AppState>,
-    AxumPath(id): AxumPath<i64>,
-) -> Result<Response, ProblemDetails> {
-    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<AppState>,
-    AxumPath(id): AxumPath<i64>,
-) -> Result<Response, ProblemDetails> {
-    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:#}"))),
-    }
-}

hive-c0re/src/dashboard_events.rs

@@ -219,14 +219,6 @@ pub enum DashboardEvent {
         seq: u64,
         schedules: Vec<hive_sh4re::WireSchedule>,
     },
-    /// 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<crate::broker::PendingReminder>,
-    },
     /// Full snapshot of capability grants (per-agent `Vec<cap_name>`).
     /// Emitted from the rebuild-queue worker after a `PermChange`
     /// `Capabilities` entry commits the JSON file. Lets the P3RM1SS10NS
@@ -294,7 +286,6 @@ impl DashboardEvent {
             DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running",
             DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed",
             DashboardEvent::SchedulesChanged { .. } => "schedules_changed",
-            DashboardEvent::RemindersChanged { .. } => "reminders_changed",
             DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed",
             DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed",
             DashboardEvent::AuditEntryAdded { .. } => "audit_entry_added",
@@ -420,10 +411,6 @@ mod tests {
                 seq: 1,
                 schedules: Vec::new(),
             },
-            DashboardEvent::RemindersChanged {
-                seq: 1,
-                reminders: Vec::new(),
-            },
             DashboardEvent::CapabilitiesChanged {
                 seq: 1,
                 caps: Vec::new(),

hive-c0re/src/loose_ends.rs

@@ -3,7 +3,8 @@
 //! a single agent (`for_agent`) or the whole hive (`hive_wide`).
 //! `Request::GetLooseEnds` from either the agent or manager socket
 //! lands here so the routing logic + age-seconds derivation stay in
-//! one place.
+//! one place. Reminders are agent-local (in-container store) and no
+//! longer sourced from here.
 //!
 //! Call frequency is low (an agent doing self-introspection between
 //! turns), so the sweep happens fresh every time — no caching, no
@@ -27,18 +28,17 @@ use hive_sh4re::wire_time::now_unix;
 ///   root submits for top-level agents). Legacy rows with no recorded
 ///   submitter count as the root's;
 /// - unanswered questions where `agent` is the asker (waiting on
-///   someone) OR the target (owes a reply);
-/// - pending reminders this agent scheduled (`owner == self`).
+///   someone) OR the target (owes a reply).
 ///
-/// Ordered `pending_messages` (when non-zero) → approvals → questions →
-/// reminders within the returned vector. Within each kind,
-/// source-of-truth ordering (sqlite's `pending()` queries return
-/// newest-first within their indexes).
+/// Ordered `pending_messages` (when non-zero) → approvals → questions
+/// within the returned vector. Within each kind, source-of-truth
+/// ordering (sqlite's `pending()` queries return newest-first within
+/// their indexes).
 ///
 /// # Errors
 ///
 /// Propagates errors from `count_pending` and the pending-approval /
-/// question / reminder sqlite queries.
+/// question sqlite queries.
 pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
     let now = now_unix();
     let mut out = Vec::new();
@@ -85,25 +85,13 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result<Vec<LooseEnd>> {
             age_seconds: saturating_age(now, q.asked_at.timestamp()),
         });
     }
-    for r in coord.broker.list_pending_reminders()? {
-        if r.agent != agent {
-            continue;
-        }
-        out.push(LooseEnd::Reminder {
-            id: r.id,
-            owner: r.agent,
-            message: r.message,
-            due_at: r.due_at,
-            age_seconds: saturating_age(now, r.created_at.timestamp()),
-        });
-    }
     Ok(out)
 }
 
 /// Hive-wide loose-ends view: EVERY pending approval + EVERY
-/// unanswered question + EVERY pending reminder. Manager surface
-/// only; sub-agents can't see each other's threads via the agent
-/// surface (`for_agent` filters by name).
+/// unanswered question. Manager surface only; sub-agents can't see
+/// each other's threads via the agent surface (`for_agent` filters by
+/// name).
 pub fn hive_wide(coord: &Coordinator) -> Result<Vec<LooseEnd>> {
     let now = now_unix();
     let mut out = Vec::new();
@@ -125,15 +113,6 @@ pub fn hive_wide(coord: &Coordinator) -> Result<Vec<LooseEnd>> {
             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)
 }
 

hive-c0re/src/main.rs

@@ -45,8 +45,7 @@ pub(crate) use stores::{
     approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
 };
 pub(crate) use workers::{
-    agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler,
-    scheduled_prompts_worker,
+    agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker,
 };
 
 use coordinator::{Coordinator, HiveEnv, ServeConfig};
@@ -214,7 +213,7 @@ fn matrix_sweep_banner(ctx: sweep_health::SweepFailure) -> String {
 }
 
 /// Start the coordinator daemon: open the broker, run migrations, spawn
-/// background tasks (auto-update, vacuums, crash-watcher, reminder-scheduler,
+/// background tasks (auto-update, vacuums, crash-watcher, scheduled-prompts,
 /// dashboard), then serve the admin socket until a signal arrives.
 #[allow(
     clippy::too_many_lines,
@@ -522,9 +521,6 @@ async fn cmd_serve(
     // run_reconcile call register_agent on start; kill/destroy call
     // unregister_agent. No recurring poll needed — c0re owns the listeners.
     mcp_sockets::sync_on_start(coord.clone()).await;
-    // Reminder scheduler: drains due reminders + handles
-    // file_path payload persistence. See reminder_scheduler.rs.
-    reminder_scheduler::spawn(coord.clone());
     // Scheduled-prompts worker: drains due scheduled_prompts rows
     // and fans the body out to each active target's inbox. See
     // scheduled_prompts_worker.rs.

hive-c0re/src/questions.rs

@@ -186,14 +186,15 @@ pub fn handle_cancel_loose_end(
             Ok(())
         }
         hive_sh4re::CancelLooseEndKind::Reminder => {
-            // Agent-socket path: ownership-only (cancel your own reminder).
-            let owner = coord
-                .broker
-                .cancel_reminder_as(id, canceller, false)
-                .map_err(|e| format!("{e:#}"))?;
-            tracing::info!(%id, %canceller, %owner, "reminder cancelled");
-            coord.emit_reminders_snapshot();
-            Ok(())
+            // Reminders are now agent-local (in-container store) — the
+            // agent-mcp `cancel_loose_end` tool branches on this kind and
+            // dials the agent's own socket directly, never forwarding to
+            // hive-c0re. This arm should be unreachable in practice; kept
+            // only so the match stays exhaustive.
+            Err(format!(
+                "reminder {id}: reminders are handled locally by the agent, \
+                 not by hive-c0re"
+            ))
         }
         hive_sh4re::CancelLooseEndKind::Approval => {
             // Withdrawing an approval needs the grantable `approvals`

hive-c0re/src/server.rs

@@ -159,7 +159,7 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
                 HostResponse::dags(dags)
             }
             HostRequest::List => HostResponse::list(lifecycle::list().await?),
-            HostRequest::AgentStatus => handle_agent_status(&coord).await,
+            HostRequest::AgentStatus => handle_agent_status().await,
             // The hive domain + per-surface public URLs are injected into
             // c0re's service env by hive-c0re.nix; surface them so the
             // operator CLI can fill in this hive's own identity (the
@@ -298,8 +298,8 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
 }
 
 /// Collect per-agent status rows for `hivectl status` and the dashboard.
-async fn handle_agent_status(coord: &Arc<Coordinator>) -> HostResponse {
-    let rows = crate::container_view::build_all(coord)
+async fn handle_agent_status() -> HostResponse {
+    let rows = crate::container_view::build_all()
         .await
         .into_iter()
         .map(|v| hive_sh4re::AgentStatusRow {
@@ -308,7 +308,13 @@ async fn handle_agent_status(coord: &Arc<Coordinator>) -> HostResponse {
             needs_update: v.needs_update,
             needs_login: v.needs_login,
             deployed_sha: v.deployed_sha,
-            pending_reminders: v.pending_reminders,
+            // Reminders are agent-local now; c0re has no cross-agent
+            // visibility into pending counts anymore. Stubbed
+            // to 0 rather than deleting the wire field outright — leaves
+            // `hivectl status`/the dashboard column intact syntactically,
+            // just always empty, until iris's frontend follow-up decides
+            // whether to drop the column entirely.
+            pending_reminders: 0,
             parent: v.parent,
         })
         .collect();

hive-c0re/src/socket_server/mod.rs

@@ -23,7 +23,6 @@ use crate::coordinator::Coordinator;
 
 mod config_approvals;
 mod lifecycle_handlers;
-mod reminders;
 mod schedules;
 
 pub(crate) use config_approvals::submit_merge_config_pr;
@@ -34,7 +33,6 @@ use config_approvals::{handle_request_init_config, handle_request_update_meta_in
 use lifecycle_handlers::{
     handle_kill, handle_list_descendants, handle_restart, handle_start, handle_update,
 };
-use reminders::{handle_remind, resolve_agent_state_target};
 use schedules::{
     EditSchedulePatch, handle_cancel_schedule, handle_edit_schedule, handle_fire_schedule_now,
     handle_list_schedules, handle_request_schedule_prompt,
@@ -184,8 +182,8 @@ pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
 /// Handle the subset of `Request` variants that are identical on both
 /// the agent socket and the manager socket. Returns `Some(response)` for
 /// every variant it handles; returns `None` for variants with socket-specific
-/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup`
-/// where the manager can target other agents) or for manager-only variants.
+/// semantics (e.g. `GetLooseEnds`, where the manager can target other
+/// agents) or for manager-only variants.
 ///
 /// The unified `dispatch` calls this first; the remaining arms (which gate
 /// on topology / capabilities / tool-groups) are handled there.
@@ -234,11 +232,6 @@ pub(crate) async fn dispatch_shared(
                 |()| hive_core_agent_sock::Response::Ok,
             )
         }
-        hive_core_agent_sock::Request::Remind {
-            message,
-            timing,
-            file_path,
-        } => handle_remind(coord, agent, message, timing, file_path.as_deref()),
         hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text),
         hive_core_agent_sock::Request::GetAgentMeta { name } => {
             handle_get_agent_meta(coord, agent, name.as_ref()).await
@@ -590,13 +583,6 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Respo
         Request::GetLooseEnds { agent: target } => {
             handle_get_loose_ends(coord, agent, target.as_deref())
         }
-        Request::CountPendingReminders { agent: target } => {
-            handle_count_pending_reminders(coord, agent, target.as_deref())
-        }
-        Request::ReminderRollup {
-            since_secs,
-            agent: target,
-        } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
         // Orchestration / diagnostics verbs — gated per-verb on tool-group
         // membership or topology (see `dispatch_orchestration`).
         _ => dispatch_orchestration(req, agent, coord).await,
@@ -791,41 +777,38 @@ fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str, target: Option<&
     }
 }
 
-/// `CountPendingReminders` — resolve the target (own / subtree free, else
-/// `QueryAgentState`) then count its pending reminders.
-fn handle_count_pending_reminders(
-    coord: &Arc<Coordinator>,
-    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<Coordinator>,
-    agent: &str,
-    target: Option<&str>,
-    since_secs: u64,
-) -> Response {
-    match resolve_agent_state_target(agent, target) {
-        Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) {
-            Ok(stats) => Response::ReminderRollup(stats),
-            Err(e) => Response::Err {
-                message: format!("{e:#}"),
-            },
-        },
-        Err(message) => Response::Err { message },
+/// Resolve the target agent name for a *named* `GetLooseEnds` query. Rules:
+///
+/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed).
+/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability.
+/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise.
+/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate.
+fn resolve_agent_state_target<'a>(
+    caller: &'a str,
+    target: Option<&'a str>,
+) -> Result<&'a str, String> {
+    match target {
+        None => Ok(caller),
+        Some("*") => Err(
+            "hive-wide query (agent=\"*\") is only valid for loose-ends; \
+             not available for this query"
+                .to_owned(),
+        ),
+        Some(name) => {
+            // Own subtree (the root covers all) is visible without extra
+            // capability; `is_descendant_of` returns true for `name == caller`.
+            if crate::topology::is_descendant_of(name, caller) {
+                return Ok(name);
+            }
+            if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) {
+                Ok(name)
+            } else {
+                Err(format!(
+                    "agent `{caller}` cannot query `{name}`: not in its subtree and \
+                     `query_agent_state` capability is not granted"
+                ))
+            }
+        }
     }
 }
 

hive-c0re/src/socket_server/reminders.rs

@@ -1,229 +0,0 @@
-//! Reminder request handling: the `Remind` handler, the shared
-//! `store_remind` storage path with its pending-cap and large-body
-//! auto-save dance, timing resolution, and the agent-state target
-//! resolution shared by the loose-ends / reminder query handlers.
-
-use std::sync::Arc;
-
-use hive_core_agent_sock::Response;
-
-use crate::coordinator::Coordinator;
-
-pub(super) fn handle_remind(
-    coord: &Arc<Coordinator>,
-    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::<u64>().ok())
-        .unwrap_or(DEFAULT_REMIND_MAX_PENDING)
-}
-
-pub(crate) fn store_remind(
-    coord: &Arc<Coordinator>,
-    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/<agent>/state/reminders/auto-<ts>.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>), 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<i64> {
-    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());
-    }
-}

hive-c0re/src/stats/host_stats.rs

@@ -250,7 +250,6 @@ mod tests {
             needs_update: false,
             needs_login,
             deployed_sha: None,
-            pending_reminders: 0,
             parent: None,
             active_model: None,
         }

hive-c0re/src/stores/broker.rs

@@ -7,7 +7,6 @@ use std::sync::Mutex;
 
 use anyhow::{Context, Result};
 
-use chrono::{DateTime, Utc};
 use hive_sh4re::wire_time::now_unix;
 use hive_sh4re::{InboxRow, Message};
 
@@ -30,18 +29,6 @@ CREATE TABLE IF NOT EXISTS messages (
 CREATE INDEX IF NOT EXISTS idx_messages_undelivered
     ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL;
 
-CREATE TABLE IF NOT EXISTS reminders (
-    id        INTEGER PRIMARY KEY AUTOINCREMENT,
-    agent     TEXT    NOT NULL,
-    message   TEXT    NOT NULL,
-    file_path TEXT,
-    due_at    INTEGER NOT NULL,
-    created_at INTEGER NOT NULL,
-    sent_at   INTEGER
-);
-CREATE INDEX IF NOT EXISTS idx_reminders_due
-    ON reminders (agent, due_at) WHERE sent_at IS NULL;
-
 CREATE TABLE IF NOT EXISTS kv (
     key   TEXT PRIMARY KEY,
     value TEXT NOT NULL
@@ -52,12 +39,6 @@ CREATE TABLE IF NOT EXISTS kv (
 /// may drop events past this; we send a `lagged` notice in their stream.
 const EVENT_CHANNEL: usize = 256;
 
-/// Row shape returned by [`Broker::get_due_reminders`]:
-/// `(agent, reminder_id, message, file_path)`. Type alias keeps
-/// `clippy::type_complexity` quiet and makes the scheduler call site
-/// self-documenting.
-pub type DueReminder = (String, i64, String, Option<String>);
-
 /// A single message hand-off from broker to recipient. Carries the
 /// broker's row id (so the harness can drive `ack_turn` later) and
 /// the redelivery flag (so the harness can prepend the
@@ -70,37 +51,6 @@ pub struct Delivery {
     pub message: Message,
 }
 
-/// Row shape for [`Broker::list_pending_reminders`], shipped on the
-/// dashboard `/api/reminders` response.
-#[derive(Debug, Clone, Serialize)]
-pub struct PendingReminder {
-    pub id: i64,
-    pub agent: String,
-    pub message: String,
-    #[serde(skip_serializing_if = "Option::is_none")]
-    pub file_path: Option<String>,
-    pub due_at: DateTime<Utc>,
-    pub created_at: DateTime<Utc>,
-    /// 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<String>,
-    /// Number of failed delivery attempts since the row was
-    /// created or last retried. After `MAX_REMINDER_ATTEMPTS` the
-    /// scheduler stops trying (the row stays in `pending` with the
-    /// error so the operator can decide between retry + cancel).
-    #[serde(default)]
-    pub attempt_count: u32,
-}
-
-/// Stop retrying a row after this many consecutive failures. The
-/// scheduler quits scheduling it until an operator explicitly
-/// retries (which resets the counter) or cancels (which deletes
-/// the row). Below the cap the existing 5s tick re-attempts each
-/// time the row is due.
-pub const MAX_REMINDER_ATTEMPTS: u32 = 5;
-
 /// Intra-process broker event. `recv_blocking_batch` listens on the
 /// same channel as the dashboard forwarder; the forwarder re-emits
 /// each event as a `DashboardEvent` with a freshly-stamped seq from
@@ -193,18 +143,6 @@ const BROKER_MIGRATIONS: &[Migration] = &[
      COMMIT;",
         adds_column: Some(("messages", "priority")),
     },
-    // v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap.
-    Migration {
-        sql: "ALTER TABLE reminders ADD COLUMN \
-         attempt_count INTEGER NOT NULL DEFAULT 0",
-        adds_column: Some(("reminders", "attempt_count")),
-    },
-    // v5: last_error on reminders — last delivery failure surfaced on the
-    // dashboard so a stuck reminder is visible without digging in logs.
-    Migration {
-        sql: "ALTER TABLE reminders ADD COLUMN last_error TEXT",
-        adds_column: Some(("reminders", "last_error")),
-    },
 ];
 
 impl Broker {
@@ -850,309 +788,6 @@ impl Broker {
         }
         Ok(u64::try_from(n).unwrap_or(0))
     }
-
-    /// Store a new reminder. Returns the reminder id.
-    pub fn store_reminder(
-        &self,
-        agent: &str,
-        message: &str,
-        file_path: Option<&str>,
-        due_at: i64,
-    ) -> Result<i64> {
-        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<Vec<PendingReminder>> {
-        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::<rusqlite::Result<Vec<_>>>()
-            .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<usize> {
-        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<u64> {
-        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<hive_sh4re::ReminderStats> {
-        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<usize> {
-        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<String> {
-        let conn = self.conn.lock().unwrap();
-        let owner: Option<String> = 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<Vec<DueReminder>> {
-        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<String>>(3)?,
-            ))
-        })?;
-        rows.collect::<rusqlite::Result<Vec<_>>>()
-            .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<Result<()>> {
-        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<Result<()>> = 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<i64> = Vec::with_capacity(items.len());
-        for (id, agent, body) in items {
-            let r = (|| -> Result<i64> {
-                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)]

hive-c0re/src/workers/mod.rs

@@ -1,5 +1,5 @@
 //! Background tasks and periodic sweeps: crash/login watcher, the
-//! reminder and scheduled-prompt delivery loops, boot-time auto-update
+//! scheduled-prompt delivery loop, boot-time auto-update
 //! reconcile, the agent-sockets.json writer loop, the MCP socket listener
 //! reconcile loop, and knowledge-repo sync. Each submodule is re-exported
 //! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged.
@@ -9,5 +9,4 @@ pub mod auto_update;
 pub mod crash_watch;
 pub mod knowledge;
 pub mod mcp_sockets;
-pub mod reminder_scheduler;
 pub mod scheduled_prompts_worker;

hive-c0re/src/workers/reminder_scheduler.rs

@@ -1,278 +0,0 @@
-//! Background loop that drains due reminders from the broker and
-//! delivers them as inbox messages. 5s poll cadence, shutdown-aware.
-//! File-path semantics (path translation, traversal + symlink defense,
-//! pointer delivery): `docs/approvals.md::Reminder delivery`.
-
-use std::io::Write;
-use std::os::unix::fs::OpenOptionsExt;
-use std::path::{Path, PathBuf};
-use std::sync::Arc;
-use std::time::Duration;
-
-use crate::coordinator::Coordinator;
-
-/// Per-tick cap on reminders delivered. Anything over this stays due
-/// in the table and gets picked up on the next tick — keeps a
-/// 10k-deep backlog from flooding the broker (or hogging the broker
-/// mutex) in one shot. 100/tick × 5s tick = sustained throughput cap
-/// of ~20 reminders/sec; bump together if the loose-ends tracker
-/// starts firing higher rates.
-const REMINDER_BATCH_LIMIT: u64 = 100;
-
-/// Poll interval. Trade-off between latency on a freshly due reminder
-/// and CPU spent on empty sweeps; 5s matches the original inline
-/// scheduler.
-const POLL_INTERVAL: Duration = Duration::from_secs(5);
-
-pub fn spawn(coord: Arc<Coordinator>) {
-    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<Coordinator>) {
-    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/<name>/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<PathBuf, String> {
-    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"));
-    }
-}

hive-core-agent-sock/src/lib.rs

@@ -9,7 +9,7 @@
 
 use hive_sh4re::{
     CancelLooseEndKind, ContainerInfo, DeliveredMessage, InboxRow, JournalPriority, LooseEnd,
-    MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule,
+    MatrixIdentity, SchedulePromptPayload, WireSchedule,
 };
 use hive_types::Ident;
 use serde::{Deserialize, Serialize};
@@ -81,18 +81,6 @@ pub enum Request {
     /// back via `HelperEvent::QuestionAnswered`: see
     /// `docs/conventions.md::Question routing (Ask / Answer)`.
     Answer { id: i64, answer: String },
-    /// Schedule a reminder message to be delivered to this agent at a
-    /// future time. The reminder lands in the agent's inbox as an auto-sent
-    /// message from `"reminder"`. Use for agent follow-ups (e.g. check task
-    /// status, retry failed operation). Message length is limited; pass
-    /// `file_path` to store in a file and get a path-reference message
-    /// instead.
-    Remind {
-        message: String,
-        timing: ReminderTiming,
-        #[serde(default)]
-        file_path: Option<String>,
-    },
     /// Loose-ends view. On the agent socket: `None` = self; direct
     /// children are always accessible; non-children require the
     /// `query_agent_state` capability — rejected with an error otherwise;
@@ -103,33 +91,6 @@ pub enum Request {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         agent: Option<String>,
     },
-    /// 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<String>,
-    },
-    /// 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("<name>")` = 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<String>,
-    },
     /// Set a free-text status string visible on the dashboard. The harness
     /// writes `{state_dir}/hyperhive-status` locally before sending this
     /// request; hive-c0re just triggers a dashboard rescan on receipt.
@@ -308,10 +269,6 @@ pub enum Response {
     /// `GetLooseEnds` result: list of loose ends pending against
     /// this agent. Ordered newest-first within each kind.
     LooseEnds { loose_ends: Vec<LooseEnd> },
-    /// `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 {