From 232d28e05abd03886cc59fc23757e897f5516cdc Mon Sep 17 00:00:00 2001 From: Mega Mind <146339422+M3gA-Mind@users.noreply.github.com> Date: Wed, 24 Jun 2026 00:18:49 +0530 Subject: [PATCH] fix(notifications): persist core notifications so missed events survive (#3805) (#3982) --- src/core/jsonrpc.rs | 2 +- src/openhuman/notifications/README.md | 16 ++- src/openhuman/notifications/bus.rs | 52 ++++++++- src/openhuman/notifications/bus_tests.rs | 45 ++++++++ src/openhuman/notifications/rpc.rs | 49 +++++++++ src/openhuman/notifications/schemas.rs | 81 +++++++++++++- src/openhuman/notifications/store.rs | 121 ++++++++++++++++++++- src/openhuman/notifications/store_tests.rs | 98 +++++++++++++++++ 8 files changed, 453 insertions(+), 11 deletions(-) diff --git a/src/core/jsonrpc.rs b/src/core/jsonrpc.rs index 1adcae7ac..9531cfe53 100644 --- a/src/core/jsonrpc.rs +++ b/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(), ); diff --git a/src/openhuman/notifications/README.md b/src/openhuman/notifications/README.md index 9ce853f33..6fd38be9b 100644 --- a/src/openhuman/notifications/README.md +++ b/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`. diff --git a/src/openhuman/notifications/bus.rs b/src/openhuman/notifications/bus.rs index 908448f99..09e804b86 100644 --- a/src/openhuman/notifications/bus.rs +++ b/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, +} + +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, ¬ification) { + 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"); diff --git a/src/openhuman/notifications/bus_tests.rs b/src/openhuman/notifications/bus_tests.rs index 1a8c19452..331716faf 100644 --- a/src/openhuman/notifications/bus_tests.rs +++ b/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. +} diff --git a/src/openhuman/notifications/rpc.rs b/src/openhuman/notifications/rpc.rs index 61ee4b5de..0e0ada232 100644 --- a/src/openhuman/notifications/rpc.rs +++ b/src/openhuman/notifications/rpc.rs @@ -370,6 +370,55 @@ pub async fn handle_mark_acted(params: Map) -> Result) -> Result { + 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) -> Result { + 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 // ───────────────────────────────────────────────────────────────────────────── diff --git a/src/openhuman/notifications/schemas.rs b/src/openhuman/notifications/schemas.rs index 6bae6b64e..4ed1b6e45 100644 --- a/src/openhuman/notifications/schemas.rs +++ b/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 { @@ -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) -> ControllerFuture { Box::pin(async move { super::rpc::handle_stats(params).await }) } +fn handle_core_list_wrap(params: Map) -> ControllerFuture { + Box::pin(async move { super::rpc::handle_core_list(params).await }) +} + +fn handle_core_mark_read_wrap(params: Map) -> 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", ] ); } diff --git a/src/openhuman/notifications/store.rs b/src/openhuman/notifications/store.rs index 0e927d841..6bd7bdca2 100644 --- a/src/openhuman/notifications/store.rs +++ b/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 { + 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> { + 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::(&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 { + 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 { + 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. diff --git a/src/openhuman/notifications/store_tests.rs b/src/openhuman/notifications/store_tests.rs index b9a9b132e..f3d590fd3 100644 --- a/src/openhuman/notifications/store_tests.rs +++ b/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" + ); +}