scheduled_prompts: fix worker doc nits (argus review on #454)

hive-c0re/src/actions.rs

@@ -5,7 +5,7 @@
 
 use std::sync::Arc;
 
-use anyhow::{Result, bail};
+use anyhow::{Context as _, Result, bail};
 use hive_sh4re::{ApprovalKind, ApprovalStatus, HelperEvent, MANAGER_AGENT};
 
 use crate::coordinator::{Coordinator, TransientKind};
@@ -90,6 +90,15 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
             coord.emit_rebuild_queue_snapshot();
             Ok(())
         }
+        ApprovalKind::SchedulePrompt => {
+            // No queue card for SchedulePrompt — the work is a single
+            // sqlite insert, the actual "running" lifetime lives on
+            // the scheduled-prompts surface itself (worker fires it
+            // at the scheduled time). Run inline + fire
+            // `ApprovalResolved` so the approval row leaves Pending
+            // immediately.
+            run_approval_schedule_prompt(&coord, approval).await
+        }
     }
 }
 
@@ -127,6 +136,39 @@ pub async fn run_approval_apply_commit(
     finish_approval(coord, &approval, result, terminal_tag, is_first_spawn)
 }
 
+/// Inline (non-queued) handler for `ApprovalKind::SchedulePrompt`.
+/// On approve, decode the `SchedulePromptPayload` JSON from the
+/// approval's `commit_ref`, insert a row into `scheduled_prompts`
+/// (with `source = Approval { id }`), and fire `ApprovalResolved`.
+/// The worker takes over from here — fan-out at fire time.
+async fn run_approval_schedule_prompt(
+    coord: &Coordinator,
+    approval: hive_sh4re::Approval,
+) -> Result<()> {
+    let result: Result<()> = async {
+        let payload: hive_sh4re::SchedulePromptPayload =
+            serde_json::from_str(&approval.commit_ref)
+                .context("decode SchedulePromptPayload from approval.commit_ref")?;
+        coord
+            .scheduled_prompts
+            .submit(crate::scheduled_prompts::NewSchedule {
+                owner: approval.agent.clone(),
+                targets: payload.targets,
+                body: payload.body,
+                first_fire_at_unix: payload.first_fire_at_unix,
+                interval_seconds: payload.interval_seconds,
+                description: payload.description,
+                source: crate::scheduled_prompts::ScheduleSource::Approval {
+                    id: approval.id,
+                },
+            })
+            .map(|_| ())
+            .context("insert scheduled prompt")
+    }
+    .await;
+    finish_approval(coord, &approval, result, None, false)
+}
+
 /// Worker entry point for `ApprovalKind::UpdateMetaInputs` queue
 /// entries. Inputs come from the approval row's `commit_ref` field
 /// (JSON-encoded by the manager submit path), not the queue entry's
@@ -301,6 +343,7 @@ fn finish_approval(
         ApprovalKind::ApplyCommit => "apply_commit",
         ApprovalKind::InitConfig => "init_config",
         ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
+        ApprovalKind::SchedulePrompt => "schedule_prompt",
     };
     let sha_short = approval
         .fetched_sha
@@ -350,9 +393,9 @@ fn finish_approval(
             sha: approval.fetched_sha.clone(),
             tag: terminal_tag,
         }),
-        // UpdateMetaInputs: ApprovalResolved already carries the result.
-        // No separate lifecycle event needed.
-        ApprovalKind::UpdateMetaInputs => {}
+        // UpdateMetaInputs / SchedulePrompt: ApprovalResolved already
+        // carries the result. No separate lifecycle event needed.
+        ApprovalKind::UpdateMetaInputs | ApprovalKind::SchedulePrompt => {}
     }
     result
 }
@@ -684,6 +727,7 @@ pub async fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()
             ApprovalKind::ApplyCommit => "apply_commit",
             ApprovalKind::InitConfig => "init_config",
             ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
+            ApprovalKind::SchedulePrompt => "schedule_prompt",
         };
         let sha_short = sha.as_deref().map(|s| s[..s.len().min(12)].to_owned());
         let description = a.description.clone();

hive-c0re/src/approvals.rs

@@ -286,6 +286,7 @@ fn row_to_approval(row: &rusqlite::Row<'_>) -> rusqlite::Result<Approval> {
         "spawn" => ApprovalKind::Spawn,
         "init_config" => ApprovalKind::InitConfig,
         "update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
+        "schedule_prompt" => ApprovalKind::SchedulePrompt,
         other => {
             return Err(rusqlite::Error::FromSqlConversionFailure(
                 2,
@@ -328,6 +329,7 @@ fn kind_to_str(kind: ApprovalKind) -> &'static str {
         ApprovalKind::Spawn => "spawn",
         ApprovalKind::InitConfig => "init_config",
         ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
+        ApprovalKind::SchedulePrompt => "schedule_prompt",
     }
 }
 
@@ -337,6 +339,7 @@ fn kind_from_str(s: &str) -> Result<ApprovalKind> {
         "spawn" => ApprovalKind::Spawn,
         "init_config" => ApprovalKind::InitConfig,
         "update_meta_inputs" => ApprovalKind::UpdateMetaInputs,
+        "schedule_prompt" => ApprovalKind::SchedulePrompt,
         other => bail!("unknown approval kind '{other}'"),
     })
 }

hive-c0re/src/coordinator.rs

@@ -37,6 +37,10 @@ pub struct Coordinator {
     pub broker: Arc<Broker>,
     pub approvals: Arc<Approvals>,
     pub questions: Arc<OperatorQuestions>,
+    /// Scheduled-prompts queue (#444). One sqlite connection,
+    /// internal mutex; the worker drains due rows and the manager
+    /// handlers insert / cancel through the same handle.
+    pub scheduled_prompts: Arc<crate::scheduled_prompts::ScheduledPrompts>,
     /// URL of the hyperhive flake (no fragment). Inlined into per-agent
     /// `flake.nix` files as `inputs.hyperhive.url`.
     pub hyperhive_flake: String,
@@ -192,12 +196,15 @@ impl Coordinator {
         let broker = Broker::open(db_path).context("open broker")?;
         let approvals = Approvals::open(db_path).context("open approvals")?;
         let questions = OperatorQuestions::open(db_path).context("open operator_questions")?;
+        let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path)
+            .context("open scheduled_prompts")?;
         let (dashboard_events, _) = broadcast::channel(DASHBOARD_CHANNEL);
         let (shutdown_tx, _) = watch::channel(false);
         Ok(Self {
             broker: Arc::new(broker),
             approvals: Arc::new(approvals),
             questions: Arc::new(questions),
+            scheduled_prompts: Arc::new(scheduled_prompts),
             hyperhive_flake,
             dashboard_port,
             operator_pronouns,

hive-c0re/src/dashboard.rs

@@ -74,6 +74,8 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
         .route("/request-spawn", post(post_request_spawn))
         .route("/op-send", post(post_op_send))
         .route("/meta-update", post(post_meta_update))
+        .route("/api/schedules", get(api_schedules).post(post_schedule_new))
+        .route("/api/schedules/{id}/cancel", post(post_schedule_cancel))
         .route("/dashboard/stream", get(dashboard_stream))
         .route("/dashboard/history", get(dashboard_history))
         // Anything not matched by the dynamic routes above falls
@@ -601,6 +603,7 @@ fn history_view(a: Approval) -> ApprovalHistoryView {
         hive_sh4re::ApprovalKind::Spawn => "spawn",
         hive_sh4re::ApprovalKind::InitConfig => "init_config",
         hive_sh4re::ApprovalKind::UpdateMetaInputs => "update_meta_inputs",
+        hive_sh4re::ApprovalKind::SchedulePrompt => "schedule_prompt",
     };
     ApprovalHistoryView {
         id: a.id,
@@ -661,6 +664,15 @@ async fn build_approval_views(approvals: Vec<Approval>) -> Vec<ApprovalView> {
                 description: a.description,
                 requested_at: a.requested_at,
             },
+            hive_sh4re::ApprovalKind::SchedulePrompt => ApprovalView {
+                id: a.id,
+                agent: a.agent,
+                kind: "schedule_prompt",
+                sha_short: None,
+                diff: None,
+                description: a.description,
+                requested_at: a.requested_at,
+            },
         });
     }
     out
@@ -1379,6 +1391,84 @@ async fn api_reminders(State(state): State<AppState>) -> Response {
     }
 }
 
+/// `GET /api/schedules` — snapshot of every schedule for the
+/// scheduled-prompts tab (#444). Returns the wire shape directly
+/// so the frontend can render without an extra translation layer.
+async fn api_schedules(State(state): State<AppState>) -> Response {
+    match state.coord.scheduled_prompts.list() {
+        Ok(rows) => axum::Json(
+            rows.into_iter()
+                .map(crate::manager_server::schedule_to_wire_public)
+                .collect::<Vec<_>>(),
+        )
+        .into_response(),
+        Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
+    }
+}
+
+/// `POST /api/schedules` — operator-direct schedule creation
+/// (mara: "user can add them manually"). Accepts the same
+/// `SchedulePromptPayload` shape as the manager request flow but
+/// skips the approval gate — the operator click *is* the
+/// approval. The schedule lands directly with
+/// `source = Operator` and the worker picks it up at fire time.
+async fn post_schedule_new(
+    State(state): State<AppState>,
+    axum::Json(payload): axum::Json<hive_sh4re::SchedulePromptPayload>,
+) -> Response {
+    if payload.targets.is_empty() {
+        return error_response("schedule must have at least one target");
+    }
+    if payload.body.trim().is_empty() {
+        return error_response("schedule body must be non-empty");
+    }
+    if let Some(0) = payload.interval_seconds {
+        return error_response("interval_seconds must be > 0 (use None for one-shot)");
+    }
+    let new = crate::scheduled_prompts::NewSchedule {
+        owner: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
+        targets: payload.targets,
+        body: payload.body,
+        first_fire_at_unix: payload.first_fire_at_unix,
+        interval_seconds: payload.interval_seconds,
+        description: payload.description,
+        source: crate::scheduled_prompts::ScheduleSource::Operator,
+    };
+    match state.coord.scheduled_prompts.submit(new) {
+        Ok(id) => axum::Json(serde_json::json!({"id": id})).into_response(),
+        Err(e) => error_response(&format!("schedule submit: {e:#}")),
+    }
+}
+
+#[derive(serde::Deserialize, Default)]
+struct CancelScheduleForm {
+    /// `None` / absent / empty array → cancel whole schedule.
+    #[serde(default)]
+    targets: Option<Vec<String>>,
+}
+
+/// `POST /api/schedules/{id}/cancel` — operator-side cancel
+/// (whole schedule when no `targets` field, partial when one is
+/// provided). Operator bypasses the topology check; the manager
+/// surface enforces it for agent callers.
+async fn post_schedule_cancel(
+    State(state): State<AppState>,
+    AxumPath(id): AxumPath<i64>,
+    body: Option<axum::Json<CancelScheduleForm>>,
+) -> Response {
+    let targets = body
+        .and_then(|axum::Json(b)| b.targets)
+        .filter(|t| !t.is_empty());
+    let result = match targets.as_deref() {
+        Some(list) => state.coord.scheduled_prompts.cancel_targets(id, list),
+        None => state.coord.scheduled_prompts.cancel_all(id),
+    };
+    match result {
+        Ok(()) => (StatusCode::OK, "ok").into_response(),
+        Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")),
+    }
+}
+
 /// Same-origin proxy that fetches the named agent's
 /// `GET /api/state` and forwards only the `links` field to the
 /// dashboard JS (issue #262). Lets the agent backend stay the

hive-c0re/src/main.rs

@@ -21,6 +21,8 @@ mod stats_vacuum;
 mod flake_check;
 mod forge;
 mod lifecycle;
+mod scheduled_prompts;
+mod scheduled_prompts_worker;
 mod limits;
 mod loose_ends;
 mod manager_server;
@@ -235,6 +237,10 @@ async fn cmd_serve(
     // 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 (#444).
+    scheduled_prompts_worker::spawn(coord.clone());
     // Rebuild-queue worker: drains the global rebuild/meta-update/
     // spawn queue FIFO so hive-c0re never runs two heavyweight
     // container ops concurrently. Existing rebuild call sites

hive-c0re/src/manager_server.rs

@@ -344,6 +344,20 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
             );
             ManagerResponse::Ok
         }
+        ManagerRequest::RequestSchedulePrompt(payload) => {
+            handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload).await
+        }
+        ManagerRequest::CancelSchedule { id, targets } => {
+            handle_cancel_schedule(coord, hive_sh4re::MANAGER_AGENT, *id, targets.as_deref())
+        }
+        ManagerRequest::ListSchedules => match coord.scheduled_prompts.list() {
+            Ok(schedules) => ManagerResponse::Schedules {
+                schedules: schedules.into_iter().map(schedule_to_wire).collect(),
+            },
+            Err(e) => ManagerResponse::Err {
+                message: format!("list scheduled prompts: {e:#}"),
+            },
+        },
         ManagerRequest::Ask {
             question,
             options,
@@ -692,6 +706,179 @@ async fn submit_apply_commit(
     Ok((id, sha))
 }
 
+/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt`
+/// row. Encodes the payload into the approval's `commit_ref` so the
+/// approve handler can re-parse it without a side table. Validates
+/// inputs (non-empty targets, non-empty body, sane interval) at
+/// submit time — the operator should never see a malformed schedule
+/// pending approval.
+async fn handle_request_schedule_prompt(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    payload: &hive_sh4re::SchedulePromptPayload,
+) -> ManagerResponse {
+    if payload.targets.is_empty() {
+        return ManagerResponse::Err {
+            message: "schedule must have at least one target".into(),
+        };
+    }
+    if payload.body.trim().is_empty() {
+        return ManagerResponse::Err {
+            message: "schedule body must be non-empty".into(),
+        };
+    }
+    if let Some(0) = payload.interval_seconds {
+        return ManagerResponse::Err {
+            message: "interval_seconds must be > 0 (use None for one-shot)".into(),
+        };
+    }
+    let commit_ref = match serde_json::to_string(payload) {
+        Ok(s) => s,
+        Err(e) => {
+            return ManagerResponse::Err {
+                message: format!("encode SchedulePromptPayload: {e:#}"),
+            }
+        }
+    };
+    let id = match coord.approvals.submit_kind(
+        requester,
+        hive_sh4re::ApprovalKind::SchedulePrompt,
+        &commit_ref,
+        payload.description.as_deref(),
+    ) {
+        Ok(id) => id,
+        Err(e) => {
+            return ManagerResponse::Err {
+                message: format!("queue schedule_prompt approval: {e:#}"),
+            }
+        }
+    };
+    tracing::info!(
+        %id,
+        requester,
+        targets = ?payload.targets,
+        first_fire_at = payload.first_fire_at_unix,
+        interval = ?payload.interval_seconds,
+        "schedule_prompt approval queued"
+    );
+    coord.emit_approval_added(
+        id,
+        requester,
+        "schedule_prompt",
+        None,
+        None,
+        payload.description.clone(),
+    );
+    ManagerResponse::Ok
+}
+
+/// Cancel a schedule (whole or per-target). Manager-surface
+/// authorization: a manager can cancel its own schedules + any
+/// schedule whose owner is one of its sub-agents (topology-walked).
+/// The operator surface bypasses this and can cancel anything;
+/// agents reaching this path through the manager get the
+/// topology-scoped check.
+fn handle_cancel_schedule(
+    coord: &Arc<Coordinator>,
+    requester: &str,
+    schedule_id: i64,
+    targets: Option<&[String]>,
+) -> ManagerResponse {
+    let schedule = match coord.scheduled_prompts.get(schedule_id) {
+        Ok(Some(s)) => s,
+        Ok(None) => {
+            return ManagerResponse::Err {
+                message: format!("schedule {schedule_id} not found"),
+            }
+        }
+        Err(e) => {
+            return ManagerResponse::Err {
+                message: format!("read schedule {schedule_id}: {e:#}"),
+            }
+        }
+    };
+    if !cancel_authorized(requester, &schedule.owner) {
+        return ManagerResponse::Err {
+            message: format!(
+                "not authorized: {requester} cannot cancel schedule owned by {owner}",
+                owner = schedule.owner
+            ),
+        };
+    }
+    let result = match targets {
+        Some(list) if !list.is_empty() => coord
+            .scheduled_prompts
+            .cancel_targets(schedule_id, list)
+            .map_err(|e| format!("cancel targets: {e:#}")),
+        _ => coord
+            .scheduled_prompts
+            .cancel_all(schedule_id)
+            .map_err(|e| format!("cancel all: {e:#}")),
+    };
+    match result {
+        Ok(()) => ManagerResponse::Ok,
+        Err(message) => ManagerResponse::Err { message },
+    }
+}
+
+/// Permission check for `CancelSchedule` on the manager surface.
+/// `requester` (always `hm1nd` here) can cancel its own schedules.
+/// Sub-agent ownership is delegated to topology — see
+/// `crate::topology::is_descendant_of`.
+fn cancel_authorized(requester: &str, owner: &str) -> bool {
+    if requester == owner {
+        return true;
+    }
+    if requester == hive_sh4re::OPERATOR_RECIPIENT {
+        return true;
+    }
+    // Manager can cancel anything owned by an agent in its subtree.
+    // For the current single-manager topology that covers everything,
+    // but the check stays correct as the tree grows.
+    crate::topology::is_descendant_of(owner, requester)
+}
+
+/// Map a `scheduled_prompts::Schedule` to its public wire shape.
+/// Field-by-field copy — the two types are intentionally identical;
+/// the separation keeps hive-sh4re free of hive-c0re-internal types.
+/// Public alias `schedule_to_wire_public` re-exports for
+/// `dashboard.rs::api_schedules` without crossing the module
+/// boundary into the manager-server file.
+pub fn schedule_to_wire_public(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
+    schedule_to_wire(s)
+}
+
+fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::WireSchedule {
+    hive_sh4re::WireSchedule {
+        id: s.id,
+        owner: s.owner,
+        body: s.body,
+        interval_seconds: s.interval_seconds,
+        next_fire_at_unix: s.next_fire_at_unix,
+        created_at_unix: s.created_at_unix,
+        source: match s.source {
+            crate::scheduled_prompts::ScheduleSource::Operator => {
+                hive_sh4re::WireScheduleSource::Operator
+            }
+            crate::scheduled_prompts::ScheduleSource::Approval { id } => {
+                hive_sh4re::WireScheduleSource::Approval { id }
+            }
+        },
+        cancelled_at_unix: s.cancelled_at_unix,
+        description: s.description,
+        targets: s
+            .targets
+            .into_iter()
+            .map(|t| hive_sh4re::WireScheduleTarget {
+                target: t.target,
+                cancelled_at_unix: t.cancelled_at_unix,
+                last_fired_at_unix: t.last_fired_at_unix,
+                last_result: t.last_result,
+            })
+            .collect(),
+    }
+}
+
 /// On `Ask { ttl_seconds: Some(n) }`, sleep n seconds and then try to
 /// resolve the question with `[expired]`. If the operator (or any
 /// other path) already answered it, `answer()` returns Err and we

hive-c0re/src/scheduled_prompts.rs

@@ -0,0 +1,680 @@
+//! Scheduled prompts (closes #444). Persistent sqlite queue of
+//! `(fire_at, targets, body)` rows that the worker fans out as
+//! broker `Message`s to each target's inbox at fire time. Recurring
+//! schedules carry `interval_seconds` and re-arm `next_fire_at` on
+//! delivery; one-shots are reaped.
+//!
+//! ## Three submit paths
+//!
+//! - **Operator-direct** (`source = Operator`): the operator adds
+//!   a schedule through the dashboard form. Lands in the table
+//!   immediately, no approval gate.
+//! - **Agent-requested** (`source = Approval { id }`): a sub-agent
+//!   (or the manager) submits a `RequestSchedulePrompt` through the
+//!   manager socket. An `ApprovalKind::SchedulePrompt` row is
+//!   queued; on approve, hive-c0re inserts the schedule row with
+//!   `source = Approval { id: approval_id }` so the audit trail
+//!   points back at the operator decision.
+//! - **No self-target shortcut**: even agent-self schedules need
+//!   approval. The existing `remind` MCP tool stays the quick
+//!   self-wake path; this module is the bigger, multi-recipient,
+//!   operator-visible thing.
+//!
+//! ## Catch-up clamp (missed-while-down)
+//!
+//! When hive-c0re comes back from being down, the worker sees rows
+//! whose `next_fire_at` is well in the past. For recurring rows
+//! that would mean firing N delayed pulses in a row — spammy and
+//! useless. Instead the worker fires ONCE per row and bumps
+//! `next_fire_at` to the next interval slot ≥ `now`, recording how
+//! many cycles were skipped in `last_result`. Operators see "fired
+//! late, caught up from 17 skipped" instead of 17 wake-up storms.
+//!
+//! ## Per-target state
+//!
+//! `targets` is its own table so partial cancellation flips a
+//! single row + so the dashboard can show last-fired / last-result
+//! per recipient. Cancelling every target reaps the parent row on
+//! the next worker pass.
+
+use std::path::Path;
+use std::sync::Mutex;
+
+use anyhow::{Context, Result, bail};
+use rusqlite::{Connection, OptionalExtension, params};
+use serde::{Deserialize, Serialize};
+
+const SCHEMA: &str = r"
+CREATE TABLE IF NOT EXISTS scheduled_prompts (
+    id                INTEGER PRIMARY KEY AUTOINCREMENT,
+    owner             TEXT    NOT NULL,
+    body              TEXT    NOT NULL,
+    interval_seconds  INTEGER,
+    next_fire_at_unix INTEGER NOT NULL,
+    created_at_unix   INTEGER NOT NULL,
+    source            TEXT    NOT NULL,
+    cancelled_at_unix INTEGER,
+    description       TEXT
+);
+CREATE INDEX IF NOT EXISTS idx_scheduled_due
+    ON scheduled_prompts (next_fire_at_unix)
+    WHERE cancelled_at_unix IS NULL;
+
+CREATE TABLE IF NOT EXISTS scheduled_prompt_targets (
+    schedule_id        INTEGER NOT NULL,
+    target             TEXT    NOT NULL,
+    cancelled_at_unix  INTEGER,
+    last_fired_at_unix INTEGER,
+    last_result        TEXT,
+    PRIMARY KEY (schedule_id, target),
+    FOREIGN KEY (schedule_id) REFERENCES scheduled_prompts(id) ON DELETE CASCADE
+);
+";
+
+/// One scheduled-prompt row + its current target set. Returned by
+/// `list` / `get`; the per-target last-result blob is suitable for
+/// rendering on the dashboard without a second query.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Schedule {
+    pub id: i64,
+    /// `"operator"` or an agent name — drives cancel-permission
+    /// checks. For approval-sourced rows this is the requesting
+    /// agent (the `Approval.agent` at submit time).
+    pub owner: String,
+    pub body: String,
+    /// `None` = one-shot, deleted after first fire.
+    /// `Some(n)` = recurring every `n` seconds.
+    pub interval_seconds: Option<u64>,
+    pub next_fire_at_unix: i64,
+    pub created_at_unix: i64,
+    pub source: ScheduleSource,
+    /// Set when the *entire* schedule was cancelled (all targets
+    /// flipped, or operator cancel-all). Worker reaps these on the
+    /// next pass.
+    pub cancelled_at_unix: Option<i64>,
+    pub description: Option<String>,
+    pub targets: Vec<ScheduleTarget>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct ScheduleTarget {
+    pub target: String,
+    pub cancelled_at_unix: Option<i64>,
+    pub last_fired_at_unix: Option<i64>,
+    pub last_result: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum ScheduleSource {
+    Operator,
+    Approval { id: i64 },
+}
+
+impl ScheduleSource {
+    fn to_db_string(&self) -> String {
+        match self {
+            ScheduleSource::Operator => "operator".to_owned(),
+            ScheduleSource::Approval { id } => format!("approval:{id}"),
+        }
+    }
+
+    fn from_db_string(s: &str) -> Self {
+        if let Some(rest) = s.strip_prefix("approval:")
+            && let Ok(id) = rest.parse::<i64>()
+        {
+            return ScheduleSource::Approval { id };
+        }
+        // Unknown / "operator" / corrupt → operator (best the row
+        // can do without an unparseable-source variant).
+        ScheduleSource::Operator
+    }
+}
+
+/// Submission payload — everything the caller knows at insert time.
+/// Used by both the operator-direct path and the approval-flow
+/// path; the latter sets `source = Approval { id }`.
+#[derive(Debug, Clone)]
+pub struct NewSchedule {
+    pub owner: String,
+    pub targets: Vec<String>,
+    pub body: String,
+    pub first_fire_at_unix: i64,
+    pub interval_seconds: Option<u64>,
+    pub description: Option<String>,
+    pub source: ScheduleSource,
+}
+
+pub struct ScheduledPrompts {
+    conn: Mutex<Connection>,
+}
+
+impl ScheduledPrompts {
+    pub fn open(path: &Path) -> Result<Self> {
+        if let Some(parent) = path.parent() {
+            std::fs::create_dir_all(parent).with_context(|| {
+                format!("create scheduled_prompts db parent {}", parent.display())
+            })?;
+        }
+        let conn = Connection::open(path)
+            .with_context(|| format!("open scheduled_prompts db {}", path.display()))?;
+        // Required for ON DELETE CASCADE to actually fire — sqlite
+        // ships with FKs disabled per connection by default.
+        conn.execute_batch("PRAGMA foreign_keys = ON;")
+            .context("enable foreign keys")?;
+        conn.execute_batch(SCHEMA)
+            .context("apply scheduled_prompts schema")?;
+        Ok(Self {
+            conn: Mutex::new(conn),
+        })
+    }
+
+    /// Insert a new schedule. Returns the new id. Empty `targets` is
+    /// rejected — a schedule with no recipients would silently
+    /// never fan out, masking caller bugs.
+    pub fn submit(&self, new: NewSchedule) -> Result<i64> {
+        if new.targets.is_empty() {
+            bail!("schedule must have at least one target");
+        }
+        let mut conn = self.conn.lock().unwrap();
+        let tx = conn.transaction()?;
+        tx.execute(
+            "INSERT INTO scheduled_prompts
+                (owner, body, interval_seconds, next_fire_at_unix,
+                 created_at_unix, source, description)
+             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)",
+            params![
+                new.owner,
+                new.body,
+                new.interval_seconds.map(i64::try_from).and_then(Result::ok),
+                new.first_fire_at_unix,
+                now_unix(),
+                new.source.to_db_string(),
+                new.description,
+            ],
+        )?;
+        let id = tx.last_insert_rowid();
+        for target in &new.targets {
+            tx.execute(
+                "INSERT INTO scheduled_prompt_targets (schedule_id, target)
+                 VALUES (?1, ?2)",
+                params![id, target],
+            )?;
+        }
+        tx.commit()?;
+        Ok(id)
+    }
+
+    /// Fetch a single schedule by id (with its target rows).
+    /// `Ok(None)` for a non-existent / already-reaped id.
+    pub fn get(&self, id: i64) -> Result<Option<Schedule>> {
+        let conn = self.conn.lock().unwrap();
+        let row = conn
+            .query_row(
+                "SELECT id, owner, body, interval_seconds, next_fire_at_unix,
+                        created_at_unix, source, cancelled_at_unix, description
+                   FROM scheduled_prompts WHERE id = ?1",
+                params![id],
+                row_to_schedule_header,
+            )
+            .optional()?;
+        let Some(mut s) = row else {
+            return Ok(None);
+        };
+        s.targets = load_targets(&conn, id)?;
+        Ok(Some(s))
+    }
+
+    /// Every active (non-globally-cancelled) schedule in insert
+    /// order. Used by the dashboard list view + the cancel-auth
+    /// check (the latter only needs the header but list() is the
+    /// shared hot path).
+    pub fn list(&self) -> Result<Vec<Schedule>> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare(
+            "SELECT id, owner, body, interval_seconds, next_fire_at_unix,
+                    created_at_unix, source, cancelled_at_unix, description
+               FROM scheduled_prompts
+              ORDER BY next_fire_at_unix ASC, id ASC",
+        )?;
+        let rows = stmt.query_map([], row_to_schedule_header)?;
+        let mut out = Vec::new();
+        for row in rows {
+            let mut s = row?;
+            s.targets = load_targets(&conn, s.id)?;
+            out.push(s);
+        }
+        Ok(out)
+    }
+
+    /// Pop the set of active rows that are due (`next_fire_at_unix <= now`)
+    /// up to `limit`. Read-only — the worker calls `mark_fired`
+    /// after each successful fan-out so the rows reappear on the
+    /// next tick when re-armed.
+    pub fn due(&self, now: i64, limit: u64) -> Result<Vec<Schedule>> {
+        let conn = self.conn.lock().unwrap();
+        let mut stmt = conn.prepare(
+            "SELECT id, owner, body, interval_seconds, next_fire_at_unix,
+                    created_at_unix, source, cancelled_at_unix, description
+               FROM scheduled_prompts
+              WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1
+              ORDER BY next_fire_at_unix ASC, id ASC
+              LIMIT ?2",
+        )?;
+        let rows = stmt.query_map(params![now, limit], row_to_schedule_header)?;
+        let mut out = Vec::new();
+        for row in rows {
+            let mut s = row?;
+            s.targets = load_targets(&conn, s.id)?;
+            out.push(s);
+        }
+        Ok(out)
+    }
+
+    /// Record a per-target fan-out result. `result` is "ok" or a
+    /// short error string; surfaces on the dashboard last-result
+    /// column. No-op for a target row that's already cancelled.
+    pub fn record_target_result(
+        &self,
+        schedule_id: i64,
+        target: &str,
+        fired_at_unix: i64,
+        result: &str,
+    ) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "UPDATE scheduled_prompt_targets
+                SET last_fired_at_unix = ?1, last_result = ?2
+              WHERE schedule_id = ?3 AND target = ?4 AND cancelled_at_unix IS NULL",
+            params![fired_at_unix, result, schedule_id, target],
+        )?;
+        Ok(())
+    }
+
+    /// Advance a recurring schedule's `next_fire_at` to the smallest
+    /// multiple-of-interval > `from`. Returns the count of skipped
+    /// cycles (≥ 0); the worker stamps that into the per-row
+    /// last_result so operators see "caught up from N missed".
+    ///
+    /// For one-shots (`interval_seconds IS NULL`) this is a no-op
+    /// at the SQL level; callers should `delete` them after fan-out
+    /// instead.
+    pub fn rearm(&self, id: i64, from_unix: i64) -> Result<u64> {
+        let conn = self.conn.lock().unwrap();
+        let row: Option<(Option<i64>, i64)> = conn
+            .query_row(
+                "SELECT interval_seconds, next_fire_at_unix
+                   FROM scheduled_prompts WHERE id = ?1",
+                params![id],
+                |row| Ok((row.get(0)?, row.get(1)?)),
+            )
+            .optional()?;
+        let Some((interval, current_next)) = row else {
+            return Ok(0);
+        };
+        let Some(interval) = interval.filter(|&i| i > 0) else {
+            return Ok(0);
+        };
+        // Smallest multiple-of-interval > `from_unix`. Catch-up
+        // semantics: when from_unix > current_next, every step in
+        // between gets counted as a skipped cycle.
+        let mut next = current_next + interval;
+        let mut skipped: u64 = 0;
+        while next <= from_unix {
+            next += interval;
+            skipped += 1;
+        }
+        conn.execute(
+            "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2",
+            params![next, id],
+        )?;
+        Ok(skipped)
+    }
+
+    /// Delete a one-shot row after its single fire. Cascades the
+    /// target rows via the FK constraint. Idempotent on a missing
+    /// id.
+    pub fn delete(&self, id: i64) -> Result<()> {
+        let conn = self.conn.lock().unwrap();
+        conn.execute(
+            "DELETE FROM scheduled_prompts WHERE id = ?1",
+            params![id],
+        )?;
+        Ok(())
+    }
+
+    /// Cancel the entire schedule (all targets flipped + the parent
+    /// row marked cancelled). Idempotent; safe to call on an
+    /// already-cancelled row.
+    pub fn cancel_all(&self, id: i64) -> Result<()> {
+        let mut conn = self.conn.lock().unwrap();
+        let now = now_unix();
+        let tx = conn.transaction()?;
+        tx.execute(
+            "UPDATE scheduled_prompts
+                SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
+              WHERE id = ?2",
+            params![now, id],
+        )?;
+        tx.execute(
+            "UPDATE scheduled_prompt_targets
+                SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
+              WHERE schedule_id = ?2",
+            params![now, id],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// Cancel a subset of a schedule's targets. When the last
+    /// active target is cancelled, the parent row is auto-cancelled
+    /// too (so the worker reaps it). Unknown targets in the list
+    /// are silently skipped.
+    pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> {
+        let mut conn = self.conn.lock().unwrap();
+        let now = now_unix();
+        let tx = conn.transaction()?;
+        for target in targets {
+            tx.execute(
+                "UPDATE scheduled_prompt_targets
+                    SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
+                  WHERE schedule_id = ?2 AND target = ?3",
+                params![now, id, target],
+            )?;
+        }
+        // If every target on this schedule is now cancelled, flip
+        // the parent so the worker stops scanning it.
+        let active_targets: i64 = tx.query_row(
+            "SELECT COUNT(*) FROM scheduled_prompt_targets
+              WHERE schedule_id = ?1 AND cancelled_at_unix IS NULL",
+            params![id],
+            |row| row.get(0),
+        )?;
+        if active_targets == 0 {
+            tx.execute(
+                "UPDATE scheduled_prompts
+                    SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1)
+                  WHERE id = ?2",
+                params![now, id],
+            )?;
+        }
+        tx.commit()?;
+        Ok(())
+    }
+
+    /// Reap cancelled rows older than `older_than_unix`. Returns
+    /// the number of rows deleted. Called from the worker on each
+    /// tick so cancellations clear out of the dashboard without an
+    /// extra periodic vacuum task.
+    pub fn reap_cancelled(&self, older_than_unix: i64) -> Result<usize> {
+        let conn = self.conn.lock().unwrap();
+        let n = conn.execute(
+            "DELETE FROM scheduled_prompts
+              WHERE cancelled_at_unix IS NOT NULL
+                AND cancelled_at_unix <= ?1",
+            params![older_than_unix],
+        )?;
+        Ok(n)
+    }
+}
+
+fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result<Schedule> {
+    let interval: Option<i64> = row.get(3)?;
+    let source_str: String = row.get(6)?;
+    Ok(Schedule {
+        id: row.get(0)?,
+        owner: row.get(1)?,
+        body: row.get(2)?,
+        interval_seconds: interval.and_then(|i| u64::try_from(i).ok()),
+        next_fire_at_unix: row.get(4)?,
+        created_at_unix: row.get(5)?,
+        source: ScheduleSource::from_db_string(&source_str),
+        cancelled_at_unix: row.get(7)?,
+        description: row.get(8)?,
+        targets: Vec::new(),
+    })
+}
+
+fn load_targets(conn: &Connection, schedule_id: i64) -> Result<Vec<ScheduleTarget>> {
+    let mut stmt = conn.prepare(
+        "SELECT target, cancelled_at_unix, last_fired_at_unix, last_result
+           FROM scheduled_prompt_targets
+          WHERE schedule_id = ?1
+          ORDER BY target ASC",
+    )?;
+    let rows = stmt.query_map(params![schedule_id], |row| {
+        Ok(ScheduleTarget {
+            target: row.get(0)?,
+            cancelled_at_unix: row.get(1)?,
+            last_fired_at_unix: row.get(2)?,
+            last_result: row.get(3)?,
+        })
+    })?;
+    let mut out = Vec::new();
+    for row in rows {
+        out.push(row?);
+    }
+    Ok(out)
+}
+
+fn now_unix() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .ok()
+        .and_then(|d| i64::try_from(d.as_secs()).ok())
+        .unwrap_or(0)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+
+    fn open() -> (TempDir, ScheduledPrompts) {
+        let dir = TempDir::new().expect("tempdir");
+        let path = dir.path().join("schedules.sqlite");
+        let db = ScheduledPrompts::open(&path).expect("open");
+        (dir, db)
+    }
+
+    fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 {
+        db.submit(NewSchedule {
+            owner: "operator".into(),
+            targets: targets.iter().map(|t| (*t).to_owned()).collect(),
+            body: "wake".into(),
+            first_fire_at_unix: fire_at,
+            interval_seconds: None,
+            description: None,
+            source: ScheduleSource::Operator,
+        })
+        .expect("submit")
+    }
+
+    #[test]
+    fn submit_and_get_round_trips_targets() {
+        let (_dir, db) = open();
+        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
+        let s = db.get(id).expect("get").expect("present");
+        assert_eq!(s.id, id);
+        assert_eq!(s.targets.len(), 2);
+        assert_eq!(s.targets[0].target, "alice");
+        assert_eq!(s.targets[1].target, "bob");
+        assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_none()));
+    }
+
+    #[test]
+    fn submit_rejects_empty_targets() {
+        let (_dir, db) = open();
+        let err = db
+            .submit(NewSchedule {
+                owner: "operator".into(),
+                targets: Vec::new(),
+                body: "wake".into(),
+                first_fire_at_unix: 100,
+                interval_seconds: None,
+                description: None,
+                source: ScheduleSource::Operator,
+            })
+            .unwrap_err();
+        assert!(format!("{err:#}").contains("at least one target"));
+    }
+
+    #[test]
+    fn due_returns_only_past_active_rows() {
+        let (_dir, db) = open();
+        let _future = submit_one_shot(&db, 1000, &["alice"]);
+        let past = submit_one_shot(&db, 50, &["alice"]);
+        let cancelled_past = submit_one_shot(&db, 50, &["bob"]);
+        db.cancel_all(cancelled_past).expect("cancel");
+        let due = db.due(100, 10).expect("due");
+        assert_eq!(due.len(), 1);
+        assert_eq!(due[0].id, past);
+    }
+
+    #[test]
+    fn rearm_handles_catch_up() {
+        let (_dir, db) = open();
+        // Recurring every 60s, last fire at t=100.
+        let id = db
+            .submit(NewSchedule {
+                owner: "operator".into(),
+                targets: vec!["alice".into()],
+                body: "wake".into(),
+                first_fire_at_unix: 100,
+                interval_seconds: Some(60),
+                description: None,
+                source: ScheduleSource::Operator,
+            })
+            .expect("submit");
+        // Worker comes back at t=400 — 5 missed cycles (160, 220,
+        // 280, 340, 400) → next should be 460, skipped = 5.
+        let skipped = db.rearm(id, 400).expect("rearm");
+        assert_eq!(skipped, 5);
+        let s = db.get(id).expect("get").expect("present");
+        assert_eq!(s.next_fire_at_unix, 460);
+    }
+
+    #[test]
+    fn rearm_advances_one_step_when_caught_up() {
+        let (_dir, db) = open();
+        let id = db
+            .submit(NewSchedule {
+                owner: "operator".into(),
+                targets: vec!["alice".into()],
+                body: "wake".into(),
+                first_fire_at_unix: 100,
+                interval_seconds: Some(60),
+                description: None,
+                source: ScheduleSource::Operator,
+            })
+            .expect("submit");
+        // Worker fires right at t=100 — single advance to t=160.
+        let skipped = db.rearm(id, 100).expect("rearm");
+        assert_eq!(skipped, 0);
+        let s = db.get(id).expect("get").expect("present");
+        assert_eq!(s.next_fire_at_unix, 160);
+    }
+
+    #[test]
+    fn rearm_is_a_no_op_for_one_shots() {
+        let (_dir, db) = open();
+        let id = submit_one_shot(&db, 100, &["alice"]);
+        let skipped = db.rearm(id, 1000).expect("rearm");
+        assert_eq!(skipped, 0);
+        // next_fire_at unchanged — one-shots are reaped via delete().
+        let s = db.get(id).expect("get").expect("present");
+        assert_eq!(s.next_fire_at_unix, 100);
+    }
+
+    #[test]
+    fn cancel_targets_auto_cancels_parent_when_last_drops() {
+        let (_dir, db) = open();
+        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
+        db.cancel_targets(id, &["alice".to_owned()]).expect("cancel");
+        let s = db.get(id).expect("get").expect("present");
+        // Parent still active (bob remains).
+        assert!(s.cancelled_at_unix.is_none());
+        db.cancel_targets(id, &["bob".to_owned()]).expect("cancel");
+        let s = db.get(id).expect("get").expect("present");
+        // Parent auto-cancels once every target is gone.
+        assert!(s.cancelled_at_unix.is_some());
+        assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some()));
+    }
+
+    #[test]
+    fn cancel_all_flips_parent_and_targets() {
+        let (_dir, db) = open();
+        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
+        db.cancel_all(id).expect("cancel");
+        let s = db.get(id).expect("get").expect("present");
+        assert!(s.cancelled_at_unix.is_some());
+        assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some()));
+    }
+
+    #[test]
+    fn delete_cascades_targets() {
+        let (_dir, db) = open();
+        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
+        db.delete(id).expect("delete");
+        assert!(db.get(id).expect("get").is_none());
+    }
+
+    #[test]
+    fn record_target_result_skips_cancelled_targets() {
+        let (_dir, db) = open();
+        let id = submit_one_shot(&db, 100, &["alice", "bob"]);
+        db.cancel_targets(id, &["alice".to_owned()]).expect("cancel");
+        db.record_target_result(id, "alice", 200, "ok")
+            .expect("record alice");
+        db.record_target_result(id, "bob", 200, "ok")
+            .expect("record bob");
+        let s = db.get(id).expect("get").expect("present");
+        let alice = s.targets.iter().find(|t| t.target == "alice").unwrap();
+        let bob = s.targets.iter().find(|t| t.target == "bob").unwrap();
+        // Cancelled targets do NOT get last-result writes.
+        assert!(alice.last_fired_at_unix.is_none());
+        assert!(alice.last_result.is_none());
+        assert_eq!(bob.last_fired_at_unix, Some(200));
+        assert_eq!(bob.last_result.as_deref(), Some("ok"));
+    }
+
+    #[test]
+    fn approval_source_round_trips() {
+        let (_dir, db) = open();
+        let id = db
+            .submit(NewSchedule {
+                owner: "manager".into(),
+                targets: vec!["alice".into()],
+                body: "wake".into(),
+                first_fire_at_unix: 100,
+                interval_seconds: None,
+                description: None,
+                source: ScheduleSource::Approval { id: 42 },
+            })
+            .expect("submit");
+        let s = db.get(id).expect("get").expect("present");
+        assert_eq!(s.source, ScheduleSource::Approval { id: 42 });
+    }
+
+    #[test]
+    fn reap_cancelled_removes_old_rows_only() {
+        let (_dir, db) = open();
+        let stale = submit_one_shot(&db, 100, &["alice"]);
+        let recent = submit_one_shot(&db, 100, &["bob"]);
+        db.cancel_all(stale).expect("cancel stale");
+        // Manually backdate the stale row.
+        {
+            let conn = db.conn.lock().unwrap();
+            conn.execute(
+                "UPDATE scheduled_prompts SET cancelled_at_unix = ?1 WHERE id = ?2",
+                params![10_i64, stale],
+            )
+            .expect("backdate");
+        }
+        db.cancel_all(recent).expect("cancel recent");
+        let n = db.reap_cancelled(100).expect("reap");
+        assert_eq!(n, 1);
+        assert!(db.get(stale).expect("get stale").is_none());
+        assert!(db.get(recent).expect("get recent").is_some());
+    }
+}

hive-c0re/src/scheduled_prompts_worker.rs

@@ -0,0 +1,247 @@
+//! Background loop that drains due `scheduled_prompts` rows
+//! (#444) and fans the body out as inbox `Message`s to each
+//! active target. Mirrors `reminder_scheduler::spawn` shape:
+//! single `spawn(coord)` entry, 5s poll cadence, shutdown-aware.
+//!
+//! ## Catch-up semantics
+//!
+//! When hive-c0re comes back from being down, a recurring row
+//! whose `next_fire_at` is well in the past would otherwise fire
+//! N delayed pulses in a row. Instead we fire ONCE and let
+//! `ScheduledPrompts::rearm` bump `next_fire_at` to the next
+//! interval slot ≥ `now`, recording the skipped-cycle count in
+//! the per-target `last_result` so operators see how many
+//! firings were caught up rather than losing the signal.
+//!
+//! ## Missing-target failure
+//!
+//! When a target name doesn't resolve to a known agent (the
+//! container has been destroyed, the operator typo'd a name,
+//! etc.) the worker:
+//!  1. records `last_result = "no such agent: <name>"` against
+//!     the per-target row,
+//!  2. sends a single advisory `Message` from `system` to
+//!     `operator` describing the schedule + target + reason,
+//!  3. continues fanning out to the other (live) targets.
+//!
+//! Transient broker errors (sqlite lock contention, etc.) get
+//! the per-target `last_result` annotated AND a `tracing::warn`,
+//! but the post-fire bookkeeping treats the row the same way it
+//! does on a clean fire:
+//!  - **recurring** rows re-arm — the next interval slot tries
+//!    the broker send again, so transient errors self-heal.
+//!  - **one-shots** delete unconditionally after their single
+//!    fan-out pass; a broker failure on a one-shot is NOT
+//!    retried (the operator advisory + last_result are the only
+//!    audit trail).
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use hive_sh4re::Message;
+
+use crate::coordinator::Coordinator;
+use crate::scheduled_prompts::Schedule;
+
+/// Per-tick cap. Each schedule fires once per tick at most;
+/// 100/tick × 5s tick = sustained throughput cap of ~20/sec,
+/// matching `reminder_scheduler::REMINDER_BATCH_LIMIT`. Bump
+/// together if real-world rates push past this.
+const SCHEDULE_BATCH_LIMIT: u64 = 100;
+
+/// Poll interval. Same 5s as the reminder scheduler — picking
+/// up freshly-due rows within at most one tick keeps the
+/// dashboard's "next fire in ..." countdown honest without
+/// burning CPU on empty sweeps.
+const POLL_INTERVAL: Duration = Duration::from_secs(5);
+
+/// Reap cancelled schedules older than this from the table so
+/// the dashboard list view doesn't accrue tombstones forever.
+/// Cancelled rows live long enough that the operator can still
+/// see what they cancelled in the recent past.
+const CANCELLED_REAP_AGE: Duration = Duration::from_secs(3600);
+
+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!("scheduled_prompts worker: shutdown signal received");
+                    break;
+                }
+            }
+        }
+    });
+}
+
+fn tick(coord: &Arc<Coordinator>) {
+    let now = now_unix();
+    let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) {
+        Ok(rows) => rows,
+        Err(e) => {
+            tracing::warn!(error = ?e, "scheduled_prompts: query due rows failed");
+            return;
+        }
+    };
+    if due.is_empty() {
+        // Periodic reaper still gets a chance even on empty ticks.
+        let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
+        if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
+            tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
+        }
+        return;
+    }
+    for schedule in due {
+        fire_schedule(coord, &schedule, now);
+    }
+    let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
+    if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
+        tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
+    }
+}
+
+/// Fan out one schedule's body to every active target. Records
+/// per-target last_result; advances or reaps the parent row at
+/// the end depending on whether `interval_seconds` is set.
+fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
+    let known: std::collections::HashSet<String> = known_agents(coord);
+    for target_row in &schedule.targets {
+        if target_row.cancelled_at_unix.is_some() {
+            continue;
+        }
+        let target = &target_row.target;
+        // `operator` is a valid recipient (mara c4) — operator
+        // delivery uses the regular broker path; the dashboard
+        // mirrors `to == operator` into its own pane.
+        if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
+            let reason = format!("no such agent: {target}");
+            if let Err(e) = coord.scheduled_prompts.record_target_result(
+                schedule.id,
+                target,
+                now,
+                &reason,
+            ) {
+                tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed");
+            }
+            notify_operator_missing_target(coord, schedule, target);
+            continue;
+        }
+        let msg = Message {
+            from: "scheduled".to_owned(),
+            to: target.clone(),
+            body: schedule.body.clone(),
+            in_reply_to: None,
+        };
+        let result = coord.broker.send(&msg);
+        let result_str = match &result {
+            Ok(()) => "ok".to_owned(),
+            Err(e) => format!("broker send failed: {e:#}"),
+        };
+        if let Err(e) = result {
+            tracing::warn!(
+                schedule = schedule.id,
+                %target,
+                error = ?e,
+                "scheduled_prompts: broker send failed (will retry on next interval)"
+            );
+        }
+        if let Err(e) =
+            coord
+                .scheduled_prompts
+                .record_target_result(schedule.id, target, now, &result_str)
+        {
+            tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed");
+        }
+    }
+    // Advance or reap. One-shots delete; recurring re-arm with
+    // catch-up clamp.
+    if schedule.interval_seconds.is_some() {
+        match coord.scheduled_prompts.rearm(schedule.id, now) {
+            Ok(0) => {}
+            Ok(skipped) => {
+                tracing::info!(
+                    schedule = schedule.id,
+                    skipped,
+                    "scheduled_prompts: caught up missed cycles"
+                );
+            }
+            Err(e) => {
+                tracing::warn!(error = ?e, schedule = schedule.id, "rearm failed");
+            }
+        }
+    } else if let Err(e) = coord.scheduled_prompts.delete(schedule.id) {
+        tracing::warn!(error = ?e, schedule = schedule.id, "delete one-shot failed");
+    }
+}
+
+/// Snapshot of live container names for the missing-target check.
+/// Always seeds the manager name (which is always reachable);
+/// adds every live nspawn container that matches the `h-` prefix.
+/// On `lifecycle::list` failure the set stays at just the manager
+/// — fail-CLOSED, meaning every non-operator/non-manager target
+/// looks missing this tick and gets the same treatment as a
+/// genuinely-destroyed agent: operator advisory + per-target
+/// `last_result` annotation + skipped delivery. Recurring
+/// schedules recover automatically on the next tick (the lifecycle
+/// listing usually works); one-shots that land on this window
+/// lose their single delivery. Logged at `warn`, not propagated.
+fn known_agents(_coord: &Coordinator) -> std::collections::HashSet<String> {
+    // `lifecycle::list` is async; the worker tick is sync. Use the
+    // blocking variant via a small `tokio::runtime::Handle::block_on`
+    // wrapper. The worker runs in its own tokio task so this is
+    // safe (we're not in a `current_thread` runtime).
+    use std::collections::HashSet;
+    let mut out: HashSet<String> = HashSet::new();
+    out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
+    let containers = tokio::task::block_in_place(|| {
+        tokio::runtime::Handle::current().block_on(crate::lifecycle::list())
+    });
+    match containers {
+        Ok(list) => {
+            for raw in list {
+                if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) {
+                    out.insert(name.to_owned());
+                } else if raw == crate::lifecycle::MANAGER_NAME {
+                    out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
+                }
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "scheduled_prompts: container listing failed");
+        }
+    }
+    out
+}
+
+/// Send the operator a one-line advisory when a schedule fires
+/// against an agent that no longer exists. Best-effort — failure
+/// to send just gets logged; the schedule continues firing.
+fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, target: &str) {
+    let body = format!(
+        "scheduled prompt #{id} fired but target `{target}` is not a live agent. \
+         body was:\n\n{body}",
+        id = schedule.id,
+        target = target,
+        body = schedule.body
+    );
+    let msg = Message {
+        from: "scheduled".to_owned(),
+        to: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
+        body,
+        in_reply_to: None,
+    };
+    if let Err(e) = coord.broker.send(&msg) {
+        tracing::warn!(error = ?e, schedule = schedule.id, %target, "operator advisory send failed");
+    }
+}
+
+fn now_unix() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .ok()
+        .and_then(|d| i64::try_from(d.as_secs()).ok())
+        .unwrap_or(0)
+}

hive-c0re/src/topology.rs

@@ -62,6 +62,32 @@ pub fn parent_of(name: &str) -> Option<String> {
     read().get(name).cloned().flatten()
 }
 
+/// True when `candidate` is `ancestor` or any descendant of
+/// `ancestor` per the current topology. Walks parents from
+/// `candidate` upward; the walk terminates at root or on a cycle
+/// (cycle defence: bounded to 32 hops, more than any plausible
+/// hive depth). Used by the cancel-authorization check in
+/// `manager_server::handle_cancel_schedule` to enforce
+/// "managers can cancel anything their subtree owns."
+#[must_use]
+pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool {
+    if candidate == ancestor {
+        return true;
+    }
+    let topo = read();
+    let mut cur = candidate.to_owned();
+    for _ in 0..32 {
+        let Some(parent) = topo.get(&cur).cloned().flatten() else {
+            return false;
+        };
+        if parent == ancestor {
+            return true;
+        }
+        cur = parent;
+    }
+    false
+}
+
 /// Persist the topology map. Sorted JSON output (BTreeMap is sorted by
 /// key) keeps git diffs minimal across re-writes. Best-effort —
 /// returns an `io::Error` so callers can decide whether a failure

hive-sh4re/src/lib.rs

@@ -110,6 +110,14 @@ pub enum ApprovalKind {
     /// JSON-encoded inputs array (`"[]"` = all inputs). Agent field is
     /// set to `hm1nd` (the requesting manager).
     UpdateMetaInputs,
+    /// Add a scheduled prompt (closes #444). On approval hive-c0re
+    /// inserts a row into `scheduled_prompts` with
+    /// `source = Approval { id }`; the worker fans the body out as
+    /// inbox messages to each target at the scheduled time, recurring
+    /// when `interval_seconds` is set. `commit_ref` stores the
+    /// JSON-encoded `SchedulePromptPayload` so the approval row carries
+    /// the full submission verbatim (target list, body, schedule).
+    SchedulePrompt,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -910,6 +918,93 @@ pub enum ManagerRequest {
         #[serde(default, skip_serializing_if = "Option::is_none")]
         description: Option<String>,
     },
+    /// Queue an approval to add a scheduled prompt (#444). The
+    /// requester (caller of this request) is recorded as the schedule
+    /// owner; on operator approval hive-c0re inserts the schedule and
+    /// the worker fans the body out at fire time. Even agent-self
+    /// schedules go through approval — the existing `remind` MCP tool
+    /// is the unapproved self-wake path.
+    RequestSchedulePrompt(SchedulePromptPayload),
+    /// Cancel a scheduled prompt (#444). `targets = None` cancels the
+    /// whole schedule; `Some(list)` cancels just those recipients,
+    /// auto-cancelling the parent when no active targets remain.
+    /// Authorization: manager can cancel its own schedules + any
+    /// sub-agent schedules (i.e. owner reachable via topology); the
+    /// operator surface bypasses this check.
+    CancelSchedule {
+        id: i64,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        targets: Option<Vec<String>>,
+    },
+    /// List every schedule in the queue. Manager-side this is
+    /// unfiltered — the dashboard does the topology-filter for the
+    /// per-agent view.
+    ListSchedules,
+}
+
+/// Submission payload for `RequestSchedulePrompt`. Lives outside the
+/// enum so it can also serialize into the approval row's `commit_ref`
+/// (the dispatcher re-parses it on approve and inserts the schedule).
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+pub struct SchedulePromptPayload {
+    /// Names of recipient agents. Operator + `hm1nd` allowed.
+    pub targets: Vec<String>,
+    /// Message body delivered to each target's inbox at fire time.
+    /// Same size budget as `Send.body` — soft cap at the broker level.
+    pub body: String,
+    /// Absolute unix timestamp (seconds) for the FIRST fire. For
+    /// recurring schedules the worker then re-arms in
+    /// `interval_seconds` steps.
+    pub first_fire_at_unix: i64,
+    /// `None` = one-shot. `Some(n > 0)` = recurring every `n` seconds.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub interval_seconds: Option<u64>,
+    /// Optional description shown on the dashboard approval card AND
+    /// stored on the resulting schedule row for the operator's
+    /// "what is this?" reference later.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+}
+
+/// Schedule row shape on the wire — mirror of
+/// `scheduled_prompts::Schedule` but in the public crate so dashboard
+/// + agent surfaces can deserialize without depending on
+/// hive-c0re-internal types. Kept structurally identical to the
+/// in-process type; the conversion is field-by-field in
+/// `manager_server` / `dashboard`.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WireSchedule {
+    pub id: i64,
+    pub owner: String,
+    pub body: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub interval_seconds: Option<u64>,
+    pub next_fire_at_unix: i64,
+    pub created_at_unix: i64,
+    pub source: WireScheduleSource,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub cancelled_at_unix: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub description: Option<String>,
+    pub targets: Vec<WireScheduleTarget>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(tag = "kind", rename_all = "snake_case")]
+pub enum WireScheduleSource {
+    Operator,
+    Approval { id: i64 },
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct WireScheduleTarget {
+    pub target: String,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub cancelled_at_unix: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub last_fired_at_unix: Option<i64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub last_result: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -944,6 +1039,12 @@ pub enum ManagerResponse {
     Logs {
         content: String,
     },
+    /// `ListSchedules` result (#444). Snapshot of every schedule
+    /// (active + cancelled-but-not-yet-reaped); the dashboard does the
+    /// per-agent topology filter on top.
+    Schedules {
+        schedules: Vec<WireSchedule>,
+    },
     /// `GetLooseEnds` result: hive-wide loose ends (approvals +
     /// unanswered questions). Same `LooseEnd` variants as the
     /// agent surface; the manager's view is unfiltered.