c0re: scheduled_prompts sqlite layer + tests (#444 step 1)

hive-c0re/src/main.rs

@@ -21,6 +21,7 @@ mod stats_vacuum;
 mod flake_check;
 mod forge;
 mod lifecycle;
+mod scheduled_prompts;
 mod limits;
 mod loose_ends;
 mod manager_server;

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());
+    }
+}