fix(notifications): persist core notifications so missed events survive (#3805) (#3982)

src/core/jsonrpc.rs

@@ -2209,7 +2209,7 @@ fn register_domain_subscribers(
         }
 
         crate::openhuman::health::bus::register_health_subscriber();
-        crate::openhuman::notifications::register_notification_bridge_subscriber();
+        crate::openhuman::notifications::register_notification_bridge_subscriber(config.clone());
         crate::openhuman::memory_conversations::register_conversation_persistence_subscriber(
             workspace_dir.clone(),
         );

src/openhuman/notifications/README.md

@@ -21,7 +21,7 @@ The notifications domain owns two complementary sub-systems. The **core-bridge**
 | `src/openhuman/notifications/bus.rs` | `NotificationBridgeSubscriber` (`EventHandler`), the `NOTIFICATION_BUS` broadcast static, `publish_core_notification`/`subscribe_core_notifications`, the pure `event_to_notification` translator, and `register_notification_bridge_subscriber`. |
 | `src/openhuman/notifications/rpc.rs` | Async RPC handler fns (`handle_ingest`, `handle_list`, `handle_mark_read`, `handle_dismiss`, `handle_mark_acted`, `handle_settings_get`/`_set`, `handle_stats`) + `triage_action_to_score` heuristic. |
 | `src/openhuman/notifications/schemas.rs` | Controller schema defs, the `NOTIFICATION_CONTROLLER_DEFS` table, `all_controller_schemas`/`all_registered_controllers`, and handler wrappers delegating to `rpc.rs`. |
-| `src/openhuman/notifications/store.rs` | SQLite persistence (`integration_notifications` + `notification_settings` tables) via a per-call `with_connection` helper; insert/list/dedup/triage-update/status/settings/stats queries. |
+| `src/openhuman/notifications/store.rs` | SQLite persistence (`integration_notifications` + `notification_settings` + `core_notifications` tables) via a per-call `with_connection` helper; insert/list/dedup/triage-update/status/settings/stats queries plus core-notification persistence (#3805). |
 | `src/openhuman/notifications/bus_tests.rs` | Sibling test suite for the bridge. |
 | `src/openhuman/notifications/store_tests.rs` | Sibling test suite for the store. |
 
@@ -35,7 +35,7 @@ Re-exported from `mod.rs`:
 
 ## RPC / controllers
 
-Namespace `notification` (8 controllers, registered via `all_notifications_registered_controllers`):
+Namespace `notification` (10 controllers, registered via `all_notifications_registered_controllers`):
 
 | Function | Inputs | Output |
 | --- | --- | --- |
@@ -47,9 +47,21 @@ Namespace `notification` (8 controllers, registered via `all_notifications_regis
 | `settings_get` | `provider` | `{ settings }` (defaulted if absent) |
 | `settings_set` | `provider`, `enabled`, `importance_threshold`, `route_to_orchestrator` | `{ ok, settings }` — threshold clamped to 0.0–1.0. |
 | `stats` | — | `{ total, unread, unscored, by_provider, by_action }` |
+| `core_list` | `only_unread?` (true), `limit?` (100) | `{ items, unread_count }` — persisted core notifications (#3805), newest first; sync-down for events fired while the app was closed. |
+| `core_mark_read` | `id` | `{ ok }` (true when a row matched) |
 
 Schemas + handlers are wired into the controller registry in `src/core/all.rs`.
 
+### Core-notification persistence (#3805)
+
+Core notifications are broadcast-only; if no client is connected when the
+event fires (app closed / minimised / disconnected) the broadcast reaches zero
+receivers and the notification is lost. `NotificationBridgeSubscriber` therefore
+**persists** each translated `CoreNotificationEvent` to a `core_notifications`
+table (keyed by event id, so re-publishes dedupe) *before* broadcasting, and the
+`core_list` / `core_mark_read` controllers let the frontend sync down and
+acknowledge anything missed on the next app open.
+
 ## Agent tools
 
 None. This domain owns no `tools.rs`.

src/openhuman/notifications/bus.rs

@@ -14,6 +14,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 use tokio::sync::broadcast;
 
 use crate::core::event_bus::{DomainEvent, EventHandler};
+use crate::openhuman::config::Config;
 use async_trait::async_trait;
 
 use super::types::{CoreNotificationCategory, CoreNotificationEvent};
@@ -46,9 +47,27 @@ pub fn publish_core_notification(event: CoreNotificationEvent) -> usize {
 }
 
 /// Subscribes to selected DomainEvent variants and translates each into a
-/// [`CoreNotificationEvent`]. Pure translation — no I/O, no locks.
+/// [`CoreNotificationEvent`], persisting it (#3805) and broadcasting it to any
+/// connected client.
+///
+/// `config` is `None` only in unit tests that exercise the pure translation /
+/// subscriber-name contract without a workspace on disk; in production it is
+/// always `Some`, so every core notification is durably stored before being
+/// broadcast and therefore survives an app-closed / disconnected window.
 #[derive(Default)]
-pub struct NotificationBridgeSubscriber;
+pub struct NotificationBridgeSubscriber {
+    config: Option<Config>,
+}
+
+impl NotificationBridgeSubscriber {
+    /// Construct a subscriber that persists notifications to the workspace
+    /// store backed by `config` before broadcasting them.
+    pub fn new(config: Config) -> Self {
+        Self {
+            config: Some(config),
+        }
+    }
+}
 
 fn now_ms() -> u64 {
     SystemTime::now()
@@ -186,6 +205,27 @@ impl EventHandler for NotificationBridgeSubscriber {
 
     async fn handle(&self, event: &DomainEvent) {
         if let Some(notification) = event_to_notification(event) {
+            // #3805: persist BEFORE broadcasting so the event is durable even
+            // when no client is currently subscribed (app closed / minimised /
+            // disconnected) — otherwise the broadcast send finds zero
+            // receivers and the notification is lost forever. Best-effort: a
+            // store failure must not suppress the live broadcast.
+            if let Some(config) = &self.config {
+                match super::store::insert_core_notification(config, &notification) {
+                    Ok(true) => log::debug!(
+                        "{LOG_PREFIX} persisted core notification id={}",
+                        notification.id
+                    ),
+                    Ok(false) => log::debug!(
+                        "{LOG_PREFIX} core notification id={} already persisted (dedup)",
+                        notification.id
+                    ),
+                    Err(err) => log::warn!(
+                        "{LOG_PREFIX} failed to persist core notification id={}: {err}",
+                        notification.id
+                    ),
+                }
+            }
             publish_core_notification(notification);
         }
     }
@@ -194,11 +234,11 @@ impl EventHandler for NotificationBridgeSubscriber {
 /// Register the notification bridge subscriber on the global event bus.
 /// Safe to call multiple times — each call produces a fresh subscription,
 /// but the caller (`register_domain_subscribers`) is Once-guarded.
-pub fn register_notification_bridge_subscriber() {
+pub fn register_notification_bridge_subscriber(config: Config) {
     use std::sync::Arc;
-    if let Some(handle) =
-        crate::core::event_bus::subscribe_global(Arc::new(NotificationBridgeSubscriber::default()))
-    {
+    if let Some(handle) = crate::core::event_bus::subscribe_global(Arc::new(
+        NotificationBridgeSubscriber::new(config),
+    )) {
         // SAFETY: intentional leak; handle's Drop would cancel the subscriber.
         std::mem::forget(handle);
         log::info!("{LOG_PREFIX} notification bridge subscriber registered");

src/openhuman/notifications/bus_tests.rs

@@ -275,3 +275,48 @@ fn notification_bridge_subscriber_name_is_stable() {
     // silently because callers (loggers, dedup guards) depend on it.
     assert_eq!(sub.name(), "notifications::bridge");
 }
+
+// ── persistence: subscriber stores core notifications (#3805) ───────────────
+
+#[tokio::test]
+async fn subscriber_persists_core_notification_so_it_survives_disconnect() {
+    use crate::openhuman::notifications::store;
+    use tempfile::TempDir;
+
+    let dir = TempDir::new().unwrap();
+    let mut config = crate::openhuman::config::Config::default();
+    config.workspace_dir = dir.path().to_path_buf();
+
+    let subscriber = NotificationBridgeSubscriber::new(config.clone());
+    // No client is subscribed to the broadcast bus here — the in-memory send
+    // reaches zero receivers. The notification must still be durably stored.
+    subscriber
+        .handle(&DomainEvent::CronJobCompleted {
+            job_id: "daily-digest".into(),
+            success: true,
+            output: "ok".into(),
+        })
+        .await;
+
+    let items = store::list_core_notifications(&config, true, 50).unwrap();
+    assert_eq!(items.len(), 1, "cron completion must be persisted");
+    assert_eq!(items[0].category, CoreNotificationCategory::Agents);
+    assert!(items[0].id.starts_with("cron:daily-digest:"));
+    assert_eq!(store::unread_core_notification_count(&config).unwrap(), 1);
+}
+
+#[tokio::test]
+async fn subscriber_without_config_does_not_persist_but_still_translates() {
+    // The Default subscriber (config = None, used only in tests) must not panic
+    // and simply skips persistence.
+    let subscriber = NotificationBridgeSubscriber::default();
+    subscriber
+        .handle(&DomainEvent::CronJobCompleted {
+            job_id: "j".into(),
+            success: false,
+            output: String::new(),
+        })
+        .await;
+    // Nothing to assert beyond "did not panic / persist" — the translation is
+    // covered by event_to_notification tests above.
+}

src/openhuman/notifications/rpc.rs

@@ -370,6 +370,55 @@ pub async fn handle_mark_acted(params: Map<String, Value>) -> Result<Value, Stri
     outcome.into_cli_compatible_json()
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// notification_core_list / notification_core_mark_read (#3805)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// List persisted core notifications (cron, webhook, sub-agent, triage, …).
+///
+/// Lets the frontend sync down notifications that fired while the app was
+/// closed / disconnected — these are otherwise lost because the live channel
+/// is broadcast-only. Optional params: `only_unread` (bool, default true),
+/// `limit` (u64, default 100).
+pub async fn handle_core_list(params: Map<String, Value>) -> Result<Value, String> {
+    let config = config_rpc::load_config_with_timeout().await?;
+
+    let only_unread = params
+        .get("only_unread")
+        .and_then(|v| v.as_bool())
+        .unwrap_or(true);
+    let limit = params
+        .get("limit")
+        .and_then(|v| v.as_u64())
+        .map(|v| v as usize)
+        .unwrap_or(100);
+
+    let items = store::list_core_notifications(&config, only_unread, limit)
+        .map_err(|e| format!("[notification_intel] core_list failed: {e}"))?;
+    let unread = store::unread_core_notification_count(&config)
+        .map_err(|e| format!("[notification_intel] core unread_count failed: {e}"))?;
+
+    let outcome = RpcOutcome::new(json!({ "items": items, "unread_count": unread }), vec![]);
+    outcome.into_cli_compatible_json()
+}
+
+/// Mark a persisted core notification as read so it isn't re-surfaced on the
+/// next sync-down.
+pub async fn handle_core_mark_read(params: Map<String, Value>) -> Result<Value, String> {
+    let config = config_rpc::load_config_with_timeout().await?;
+    let id = params
+        .get("id")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| "[notification_intel] missing required param 'id'".to_string())?
+        .to_string();
+
+    let updated = store::mark_core_notification_read(&config, &id)
+        .map_err(|e| format!("[notification_intel] core_mark_read failed: {e}"))?;
+    tracing::debug!(id = %id, updated = updated, "[notification_intel] core notification marked read");
+    let outcome = RpcOutcome::new(json!({ "ok": updated }), vec![]);
+    outcome.into_cli_compatible_json()
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // notification_stats
 // ─────────────────────────────────────────────────────────────────────────────

src/openhuman/notifications/schemas.rs

@@ -62,6 +62,16 @@ const NOTIFICATION_CONTROLLER_DEFS: &[NotificationControllerDef] = &[
         schema: schema_stats,
         handler: handle_stats_wrap,
     },
+    NotificationControllerDef {
+        function: "core_list",
+        schema: schema_core_list,
+        handler: handle_core_list_wrap,
+    },
+    NotificationControllerDef {
+        function: "core_mark_read",
+        schema: schema_core_mark_read,
+        handler: handle_core_mark_read_wrap,
+    },
 ];
 
 pub fn all_controller_schemas() -> Vec<ControllerSchema> {
@@ -372,6 +382,63 @@ fn schema_stats() -> ControllerSchema {
     }
 }
 
+fn schema_core_list() -> ControllerSchema {
+    ControllerSchema {
+        namespace: "notification",
+        function: "core_list",
+        description: "List persisted core notifications (cron, webhook, sub-agent, triage) so \
+                      the frontend can sync down events fired while the app was closed (#3805).",
+        inputs: vec![
+            FieldSchema {
+                name: "only_unread",
+                ty: TypeSchema::Option(Box::new(TypeSchema::Bool)),
+                comment: "Return only unread notifications. Defaults to true.",
+                required: false,
+            },
+            FieldSchema {
+                name: "limit",
+                ty: TypeSchema::Option(Box::new(TypeSchema::U64)),
+                comment: "Maximum number of records to return; defaults to 100.",
+                required: false,
+            },
+        ],
+        outputs: vec![
+            FieldSchema {
+                name: "items",
+                ty: TypeSchema::Array(Box::new(TypeSchema::Ref("CoreNotificationEvent"))),
+                comment: "Persisted core notifications, newest first.",
+                required: true,
+            },
+            FieldSchema {
+                name: "unread_count",
+                ty: TypeSchema::I64,
+                comment: "Total count of unread core notifications.",
+                required: true,
+            },
+        ],
+    }
+}
+
+fn schema_core_mark_read() -> ControllerSchema {
+    ControllerSchema {
+        namespace: "notification",
+        function: "core_mark_read",
+        description: "Mark a persisted core notification as read by its id (#3805).",
+        inputs: vec![FieldSchema {
+            name: "id",
+            ty: TypeSchema::String,
+            comment: "Id of the core notification to mark as read.",
+            required: true,
+        }],
+        outputs: vec![FieldSchema {
+            name: "ok",
+            ty: TypeSchema::Bool,
+            comment: "True when a matching notification was updated.",
+            required: true,
+        }],
+    }
+}
+
 fn schema_unknown() -> ControllerSchema {
     ControllerSchema {
         namespace: "notification",
@@ -428,6 +495,14 @@ fn handle_stats_wrap(params: Map<String, Value>) -> ControllerFuture {
     Box::pin(async move { super::rpc::handle_stats(params).await })
 }
 
+fn handle_core_list_wrap(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { super::rpc::handle_core_list(params).await })
+}
+
+fn handle_core_mark_read_wrap(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move { super::rpc::handle_core_mark_read(params).await })
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Tests
 // ─────────────────────────────────────────────────────────────────────────────
@@ -453,6 +528,8 @@ mod tests {
                 "dismiss",
                 "mark_acted",
                 "stats",
+                "core_list",
+                "core_mark_read",
             ]
         );
     }
@@ -460,7 +537,7 @@ mod tests {
     #[test]
     fn all_registered_controllers_has_handler_per_schema() {
         let controllers = all_registered_controllers();
-        assert_eq!(controllers.len(), 8);
+        assert_eq!(controllers.len(), 10);
         let names: Vec<_> = controllers.iter().map(|c| c.schema.function).collect();
         assert_eq!(
             names,
@@ -473,6 +550,8 @@ mod tests {
                 "dismiss",
                 "mark_acted",
                 "stats",
+                "core_list",
+                "core_mark_read",
             ]
         );
     }

src/openhuman/notifications/store.rs

@@ -9,7 +9,9 @@ use rusqlite::{params, Connection};
 
 use crate::openhuman::config::Config;
 
-use super::types::{IntegrationNotification, NotificationSettings, NotificationStatus};
+use super::types::{
+    CoreNotificationEvent, IntegrationNotification, NotificationSettings, NotificationStatus,
+};
 
 /// SQL schema applied on every `with_connection` call (idempotent).
 const SCHEMA: &str = "
@@ -42,6 +44,24 @@ CREATE TABLE IF NOT EXISTS notification_settings (
     importance_threshold  REAL NOT NULL DEFAULT 0.0,
     route_to_orchestrator INTEGER NOT NULL DEFAULT 1
 );
+
+-- #3805: core notification events (cron completions, webhook failures,
+-- sub-agent results, triaged alerts, …) are broadcast over an in-memory
+-- channel that drops the event when no client is connected. Persist them
+-- here so a notification fired while the app is closed/minimised/disconnected
+-- survives and is surfaced on the next app open. The full wire payload is
+-- stored as JSON; `id` is the PK so re-publishes dedupe naturally.
+CREATE TABLE IF NOT EXISTS core_notifications (
+    id            TEXT PRIMARY KEY,
+    payload       TEXT NOT NULL,
+    timestamp_ms  INTEGER NOT NULL,
+    read          INTEGER NOT NULL DEFAULT 0,
+    created_at    TEXT NOT NULL
+);
+CREATE INDEX IF NOT EXISTS idx_core_notifications_read
+    ON core_notifications(read);
+CREATE INDEX IF NOT EXISTS idx_core_notifications_ts
+    ON core_notifications(timestamp_ms);
 ";
 
 /// Open (and migrate) the notifications DB, then call `f` with the live
@@ -113,6 +133,105 @@ pub fn insert(config: &Config, n: &IntegrationNotification) -> Result<()> {
     })
 }
 
+// ─────────────────────────────────────────────────────────────────────────────
+// Core notification persistence (#3805)
+// ─────────────────────────────────────────────────────────────────────────────
+
+/// Persist a core notification event so it survives an app restart / a window
+/// where no client was connected to the broadcast bus.
+///
+/// Idempotent: the event `id` is the primary key, so a re-publish of the same
+/// event is ignored (no duplicates). Returns `true` when a new row was written,
+/// `false` when an event with the same id already existed.
+pub fn insert_core_notification(config: &Config, event: &CoreNotificationEvent) -> Result<bool> {
+    with_connection(config, |conn| {
+        let payload = serde_json::to_string(event)
+            .context("[notifications::store] serialize core notification failed")?;
+        let affected = conn
+            .execute(
+                "INSERT OR IGNORE INTO core_notifications
+                 (id, payload, timestamp_ms, read, created_at)
+                 VALUES (?1, ?2, ?3, 0, ?4)",
+                params![
+                    event.id,
+                    payload,
+                    event.timestamp_ms as i64,
+                    Utc::now().to_rfc3339(),
+                ],
+            )
+            .context("[notifications::store] insert_core_notification failed")?;
+        Ok(affected > 0)
+    })
+}
+
+/// Return persisted core notifications, newest first. When `only_unread` is
+/// true only rows with `read = 0` are returned. Used by the frontend to
+/// sync-down notifications missed while the app was closed.
+pub fn list_core_notifications(
+    config: &Config,
+    only_unread: bool,
+    limit: usize,
+) -> Result<Vec<CoreNotificationEvent>> {
+    with_connection(config, |conn| {
+        let sql = if only_unread {
+            "SELECT payload FROM core_notifications WHERE read = 0
+             ORDER BY timestamp_ms DESC LIMIT ?1"
+        } else {
+            "SELECT payload FROM core_notifications
+             ORDER BY timestamp_ms DESC LIMIT ?1"
+        };
+        let mut stmt = conn
+            .prepare(sql)
+            .context("[notifications::store] prepare list_core_notifications failed")?;
+        let rows = stmt
+            .query_map(params![limit as i64], |row| row.get::<_, String>(0))
+            .context("[notifications::store] query list_core_notifications failed")?;
+
+        let mut out = Vec::new();
+        for row in rows {
+            let payload =
+                row.context("[notifications::store] read core notification row failed")?;
+            match serde_json::from_str::<CoreNotificationEvent>(&payload) {
+                Ok(event) => out.push(event),
+                // A single corrupt row must not break the whole sync-down.
+                Err(err) => tracing::warn!(
+                    error = %err,
+                    "[notifications::store] skipping undeserializable core notification"
+                ),
+            }
+        }
+        Ok(out)
+    })
+}
+
+/// Mark a persisted core notification as read so it isn't re-surfaced on the
+/// next sync-down. Returns `true` when a row was updated.
+pub fn mark_core_notification_read(config: &Config, id: &str) -> Result<bool> {
+    with_connection(config, |conn| {
+        let affected = conn
+            .execute(
+                "UPDATE core_notifications SET read = 1 WHERE id = ?1",
+                params![id],
+            )
+            .context("[notifications::store] mark_core_notification_read failed")?;
+        Ok(affected > 0)
+    })
+}
+
+/// Count unread persisted core notifications.
+pub fn unread_core_notification_count(config: &Config) -> Result<i64> {
+    with_connection(config, |conn| {
+        let count: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM core_notifications WHERE read = 0",
+                [],
+                |row| row.get(0),
+            )
+            .context("[notifications::store] unread_core_notification_count failed")?;
+        Ok(count)
+    })
+}
+
 /// Atomically insert a notification unless a matching one arrived recently.
 ///
 /// Returns `true` when inserted, `false` when skipped as duplicate.

src/openhuman/notifications/store_tests.rs

@@ -279,3 +279,101 @@ fn insert_if_not_recent_after_expiry_inserts_again() {
         "both old and fresh entries should be stored"
     );
 }
+
+// ── core notification persistence (#3805) ───────────────────────────────────
+
+fn sample_core_event(id: &str, ts: u64) -> CoreNotificationEvent {
+    CoreNotificationEvent {
+        id: id.to_string(),
+        category: super::super::types::CoreNotificationCategory::Agents,
+        title: "Cron job completed".to_string(),
+        body: "Job daily-digest finished successfully.".to_string(),
+        deep_link: Some("/settings/cron-jobs".to_string()),
+        timestamp_ms: ts,
+        actions: None,
+    }
+}
+
+#[test]
+fn core_notification_insert_persists_and_lists() {
+    let dir = TempDir::new().unwrap();
+    let config = test_config(&dir);
+
+    assert!(insert_core_notification(&config, &sample_core_event("cron:1", 100)).unwrap());
+    assert!(insert_core_notification(&config, &sample_core_event("cron:2", 200)).unwrap());
+
+    let items = list_core_notifications(&config, true, 50).unwrap();
+    assert_eq!(items.len(), 2);
+    // Newest first.
+    assert_eq!(items[0].id, "cron:2");
+    assert_eq!(items[1].id, "cron:1");
+    assert_eq!(unread_core_notification_count(&config).unwrap(), 2);
+}
+
+#[test]
+fn core_notification_insert_is_idempotent_on_id() {
+    let dir = TempDir::new().unwrap();
+    let config = test_config(&dir);
+
+    assert!(insert_core_notification(&config, &sample_core_event("cron:dup", 100)).unwrap());
+    // Same id re-published — must not create a duplicate row.
+    assert!(!insert_core_notification(&config, &sample_core_event("cron:dup", 100)).unwrap());
+
+    assert_eq!(
+        list_core_notifications(&config, false, 50).unwrap().len(),
+        1
+    );
+    assert_eq!(unread_core_notification_count(&config).unwrap(), 1);
+}
+
+#[test]
+fn core_notification_mark_read_excludes_from_unread_list() {
+    let dir = TempDir::new().unwrap();
+    let config = test_config(&dir);
+
+    insert_core_notification(&config, &sample_core_event("cron:a", 100)).unwrap();
+    insert_core_notification(&config, &sample_core_event("cron:b", 200)).unwrap();
+
+    assert!(mark_core_notification_read(&config, "cron:a").unwrap());
+    // Marking a missing id returns false.
+    assert!(!mark_core_notification_read(&config, "cron:missing").unwrap());
+
+    let unread = list_core_notifications(&config, true, 50).unwrap();
+    assert_eq!(unread.len(), 1);
+    assert_eq!(unread[0].id, "cron:b");
+    assert_eq!(unread_core_notification_count(&config).unwrap(), 1);
+
+    // Non-filtered list still returns both (read + unread).
+    assert_eq!(
+        list_core_notifications(&config, false, 50).unwrap().len(),
+        2
+    );
+}
+
+#[test]
+fn core_notification_list_respects_limit() {
+    let dir = TempDir::new().unwrap();
+    let config = test_config(&dir);
+
+    for i in 0..5 {
+        insert_core_notification(&config, &sample_core_event(&format!("cron:{i}"), 100 + i))
+            .unwrap();
+    }
+    assert_eq!(list_core_notifications(&config, true, 3).unwrap().len(), 3);
+}
+
+#[test]
+fn core_notification_roundtrip_preserves_payload() {
+    let dir = TempDir::new().unwrap();
+    let config = test_config(&dir);
+
+    let event = sample_core_event("cron:rt", 1234);
+    insert_core_notification(&config, &event).unwrap();
+
+    let items = list_core_notifications(&config, true, 1).unwrap();
+    assert_eq!(items.len(), 1);
+    assert_eq!(
+        items[0], event,
+        "payload must round-trip through the store unchanged"
+    );
+}